diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 0125d578a..c1803ba63 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -1,11 +1,14 @@ import logging from functools import partial -from typing import TypeVar +from pathlib import Path +from typing import TypeVar, cast import attr +import click +import yaml from miio import MiotDevice -from miio.click_common import command +from miio.click_common import command, format_output from miio.descriptors import AccessFlags, ActionDescriptor, PropertyDescriptor from miio.miot_cloud import MiotCloud from miio.miot_device import MiotMapping @@ -89,7 +92,7 @@ def _enrich_with_metadata(self, entity: MiotBaseModel, desc: _D) -> _D: access the raw device-given name if needed. """ meta = self._meta.get_metadata(entity) - if meta is None or meta.description == desc.name: + if meta is None or meta.description is None or meta.description == desc.name: return desc _LOGGER.debug("Renamed %s to %s", desc.name, meta.description) @@ -120,6 +123,7 @@ def _create_properties(self, serv: MiotService): if prop.access == [MiotAccess.Notify]: _LOGGER.debug("Skipping notify-only property: %s", prop) continue + if not prop.access: # some properties are defined only to be used as inputs or outputs for actions _LOGGER.debug( @@ -181,6 +185,99 @@ def device_type(self) -> str | None: return self._miot_model.urn.type return None + @command( + click.option( + "--output-dir", + type=click.Path(file_okay=False), + default=None, + help="Write one YAML file per namespace to this directory.", + ), + default_output=format_output("", ""), + ) + def metadata(self, output_dir: str | None = None): + """Show metadata coverage and optionally generate YAML templates for missing items.""" + if not self._initialized: + self._initialize_descriptors() + + miot_model = cast(DeviceModel, self._miot_model) + + for serv in miot_model.services: + if serv.siid == 1: + continue + + click.echo(f"\n{serv}") + ns_name = serv.urn.namespace + nd_lines: list[str] = [] + for entity in [*serv.properties, *serv.actions]: + direct = self._meta.lookup_in_namespace( + ns_name, serv.name, entity.urn.type, entity.urn.name + ) + if direct: + if direct.description is None: + nd_lines.append( + f" [??] {entity!s:50} (fill in description if known)" + ) + else: + click.echo(f" [ok] {entity!s:50} -> {direct}") + continue + + fallback = self._meta.get_metadata(entity) + if fallback: + if fallback.description is None: + nd_lines.append( + f" [??] {entity!s:50} (fill in description if known)" + ) + else: + click.echo( + f" [fb] {entity!s:50} -> {fallback} ({fallback.source})" + ) + else: + click.echo(f" [--] {entity!s:50} {entity.description!r}") + + for line in nd_lines: + click.echo(line) + + cov = self._meta.collect_coverage(miot_model) + + click.echo( + f"\nCoverage: {cov.ok} ok, {cov.fb} via fallback, " + f"{cov.no_desc} without description, {cov.missing} missing " + f"(total {cov.total})" + ) + + if not cov.missing_by_ns: + if cov.no_desc: + click.echo( + f"{cov.no_desc} entries lack a description " + "- fill them in if you know what they are." + ) + else: + click.echo("All entities are covered.") + return + + for ns_name, services in cov.missing_by_ns.items(): + ns_meta = self._meta.build_namespace_metadata(ns_name, services) + suggested = self._meta.suggested_filename(ns_name) + + if output_dir is not None: + out = Path(output_dir) / suggested + created = self._meta.write_namespace_metadata(ns_meta, out) + click.echo(f"{'Written' if created else 'Updated'}: {out}") + base_file = Path(output_dir) / "base.yaml" + if base_file.exists(): + if self._meta.register_namespace(ns_name, suggested, base_file): + click.echo(f"Registered in {base_file}") + else: + click.echo(f"\n--- {ns_name} (save as {suggested}) ---") + click.echo( + yaml.dump( + ns_meta.model_dump(exclude_defaults=True), + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + ) + @classmethod def get_device_group(cls): """Return device command group. diff --git a/miio/integrations/genericmiot/meta.py b/miio/integrations/genericmiot/meta.py index 731f03ca7..bea33e67f 100644 --- a/miio/integrations/genericmiot/meta.py +++ b/miio/integrations/genericmiot/meta.py @@ -1,41 +1,65 @@ import logging +from collections import defaultdict from pathlib import Path +from typing import NamedTuple import yaml from pydantic import BaseModel, ConfigDict -from miio.miot_models import MiotBaseModel +from miio.miot_models import DeviceModel, MiotBaseModel _LOGGER = logging.getLogger(__name__) _ANY_SERVICE = "__ANY__" +class CoverageResult(NamedTuple): + """Aggregated metadata coverage for a device model.""" + + total: int + ok: int + fb: int + missing: int + no_desc: int + missing_by_ns: dict + + class MetaBase(BaseModel): """Base metadata with description.""" description: str + source: str | None = None # namespace that provided this metadata, set on lookup model_config = ConfigDict(extra="forbid") + def __str__(self) -> str: + return self.description if self.description else "(no description)" + class ActionMeta(MetaBase): """Metadata for actions.""" + description: str | None = None # type: ignore[assignment] + class PropertyMeta(MetaBase): """Metadata for properties.""" + description: str | None = None # type: ignore[assignment] + class ServiceMeta(MetaBase): """Metadata for a service, containing per-action and per-property metadata.""" description: str | None = None # type: ignore[assignment] - action: dict[str, ActionMeta] = {} property: dict[str, PropertyMeta] = {} + action: dict[str, ActionMeta] = {} event: dict = {} model_config = ConfigDict(extra="forbid") + def __str__(self) -> str: + return self.description if self.description else "(no description)" + def get(self, type_: str, name: str) -> MetaBase | None: """Return metadata for the given type and name, or None if not found.""" return getattr(self, type_).get(name) @@ -44,8 +68,21 @@ def get(self, type_: str, name: str) -> MetaBase | None: class Namespace(MetaBase): """A namespace (e.g. miot-spec-v2) containing service definitions.""" - fallback: str | None = None services: dict[str, ServiceMeta] = {} + source_file: str | None = None + + def merge(self, other: "Namespace") -> None: + """Add entries from other that are not already present in this namespace.""" + for svc_name, other_svc in other.services.items(): + if svc_name not in self.services: + self.services[svc_name] = other_svc + continue + + svc = self.services[svc_name] + for name, prop in other_svc.property.items(): + svc.property.setdefault(name, prop) + for name, act in other_svc.action.items(): + svc.action.setdefault(name, act) class Metadata(BaseModel): @@ -67,67 +104,179 @@ def load(cls, file: Path | None = None) -> "Metadata": with file.open() as f: data = yaml.safe_load(f) + missing = [] for ns_name, ns_value in data["namespaces"].items(): if isinstance(ns_value, str): ns_path = file.parent / ns_value + if not ns_path.exists(): + _LOGGER.warning("Namespace file not found, skipping: %s", ns_path) + missing.append(ns_name) + continue + _LOGGER.debug("Loading namespace %s from %s", ns_name, ns_path) with ns_path.open() as f: - data["namespaces"][ns_name] = yaml.safe_load(f) + ns_data = yaml.safe_load(f) + ns_data["source_file"] = ns_value + data["namespaces"][ns_name] = ns_data + + for ns_name in missing: + del data["namespaces"][ns_name] return cls(**data) + def suggested_filename(self, ns_name: str) -> str: + """Return the filename for a namespace metadata file.""" + ns = self.namespaces.get(ns_name) + if ns and ns.source_file: + return ns.source_file + + return f"{ns_name.replace('-', '')}.yaml" + def _lookup_in_namespace( self, ns: "Namespace", service_name: str, type_: str, entity_name: str ) -> MetaBase | None: - """Look up metadata within a single namespace, following fallback if needed.""" + """Look up metadata in a namespace's own services.""" for svc_name in (service_name, _ANY_SERVICE): if (serv := ns.services.get(svc_name)) and ( meta := serv.get(type_, entity_name) ): return meta - common = self.namespaces.get("common") - fallback_ns = self.namespaces.get(ns.fallback or "common", common) + return None - if fallback_ns is not None and fallback_ns is not ns: - return self._lookup_in_namespace( - fallback_ns, service_name, type_, entity_name + def build_namespace_metadata( + self, + ns_name: str, + missing: dict[str, list[MiotBaseModel]], + ) -> "Namespace": + """Build a Namespace with template entries for entities that lack coverage.""" + services = {} + for svc_name, entities in missing.items(): + props = {} + acts = {} + for entity in entities: + if entity.urn.type == "property": + props[entity.urn.name] = PropertyMeta( + description=entity.description + ) + elif entity.urn.type == "action": + acts[entity.urn.name] = ActionMeta(description=entity.description) + + svc_desc = entities[0].service.description if entities[0].service else None + services[svc_name] = ServiceMeta( + description=svc_desc, + property=props, + action=acts, ) - return None + return Namespace( + description=f"Metadata for {ns_name} namespace", + services=services, + ) - def get_metadata(self, entity: MiotBaseModel) -> MetaBase | None: - """Look up metadata for a miot entity (property or action). + def register_namespace(self, ns_name: str, filename: str, base_file: Path) -> bool: + """Add a namespace entry to the index file if not already listed.""" + data = yaml.safe_load(base_file.read_text()) + if ns_name in data["namespaces"]: + return False + + data["namespaces"][ns_name] = filename + base_file.write_text( + yaml.dump( + data, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + ) + + return True + + def write_namespace_metadata(self, ns_meta: "Namespace", path: Path) -> bool: + """Write namespace metadata to a file, merging into any existing content.""" + created = not path.exists() + if not created: + existing = Namespace.model_validate(yaml.safe_load(path.read_text())) + existing.merge(ns_meta) + ns_meta = existing + + data = ns_meta.model_dump(exclude_defaults=True) + for svc in data.get("services", {}).values(): + for key in ("property", "action"): + if key in svc: + svc[key] = dict(sorted(svc[key].items())) + + path.write_text( + yaml.dump( + data, default_flow_style=False, sort_keys=False, allow_unicode=True + ) + ) + + return created - Returns a MetaBase object, or None if no metadata was found. - """ - urn = entity.extras.get("urn") - if urn is None: + def lookup_in_namespace( + self, ns_name: str, service_name: str, type_: str, entity_name: str + ) -> MetaBase | None: + """Look up metadata in a specific namespace's own services.""" + ns = self.namespaces.get(ns_name) + if ns is None: return None + return self._lookup_in_namespace(ns, service_name, type_, entity_name) + + def collect_coverage(self, device_model: DeviceModel) -> CoverageResult: + """Count metadata coverage for a device model and collect missing entities.""" + missing_by_ns: dict = defaultdict(dict) + total = ok = fb = missing = no_desc = 0 + + for serv in device_model.services: + if serv.siid == 1: + continue + + ns_name = serv.urn.namespace + for entity in [*serv.properties, *serv.actions]: + total += 1 + direct = self.lookup_in_namespace( + ns_name, serv.name, entity.urn.type, entity.urn.name + ) + if direct: + if direct.description is None: + no_desc += 1 + else: + ok += 1 + continue + + fallback = self.get_metadata(entity) + if fallback: + if fallback.description is None: + no_desc += 1 + else: + fb += 1 + else: + missing += 1 + missing_by_ns[ns_name].setdefault(serv.name, []).append(entity) + + return CoverageResult(total, ok, fb, missing, no_desc, missing_by_ns) + + def get_metadata(self, entity: MiotBaseModel) -> MetaBase | None: + """Look up metadata for a miot entity, returning it with source namespace set.""" if entity.service is None: return None - ns_name: str = urn.namespace - service_name: str = entity.service.name - type_: str = urn.type - entity_name: str = urn.name + ns_name = entity.urn.namespace + service_name = entity.service.name - ns = self.namespaces.get(ns_name, self.namespaces["common"]) + for try_name in dict.fromkeys([ns_name, "miot-spec-v2", "common"]): + ns = self.namespaces.get(try_name) + if ns is None: + continue - meta = self._lookup_in_namespace(ns, service_name, type_, entity_name) - if meta is None: - _LOGGER.debug( - "No metadata for %s:%s:%s:%s", ns_name, service_name, type_, entity_name + meta = self._lookup_in_namespace( + ns, service_name, entity.urn.type, entity.urn.name ) - return None + if meta is not None: + _LOGGER.debug("Found metadata for %s in %s", entity, try_name) + return meta.model_copy(update={"source": try_name}) - _LOGGER.debug( - "Found metadata for %s:%s:%s:%s: %s", - ns_name, - service_name, - type_, - entity_name, - meta, - ) - return meta + _LOGGER.debug("No metadata for %s", entity) + return None diff --git a/miio/integrations/genericmiot/metadata/base.yaml b/miio/integrations/genericmiot/metadata/base.yaml index 1039365d7..4bc1abb65 100644 --- a/miio/integrations/genericmiot/metadata/base.yaml +++ b/miio/integrations/genericmiot/metadata/base.yaml @@ -2,3 +2,4 @@ namespaces: miot-spec-v2: miotspec.yaml dreame-spec: dreamespec.yaml common: common.yaml + cgllc-spec: cgllcspec.yaml diff --git a/miio/integrations/genericmiot/metadata/cgllcspec.yaml b/miio/integrations/genericmiot/metadata/cgllcspec.yaml new file mode 100644 index 000000000..e4624cdad --- /dev/null +++ b/miio/integrations/genericmiot/metadata/cgllcspec.yaml @@ -0,0 +1,72 @@ +description: Metadata for cgllc-spec namespace +services: + settings: + description: Device settings + property: + auto-slideing-time: + description: Screensaver interval + carbondioxide-led-th: + description: CO2 LED threshold + device-off: + description: Auto power off + device-off-new: {} + end-time: {} + humi-led-th: + description: Humidity LED threshold + is-twelve-hours-sys: + description: Use 12-hour clock + monitoring-frequency: + description: Monitoring frequency + page-sequence: + description: Screensaver order + pm-t-led-th: {} + pm-tpf-led-th: {} + pm-tpf-standard: {} + screen-off: + description: Screen off timeout + screensaver-time: + description: Screensaver timeout + screensaver-type: + description: Screensaver type + start-time: {} + temp-led-th: + description: Temperature LED threshold + tempature-unit: + description: Temperature unit + time-zone: + description: Timezone + action: + reset-carbon-dioxide: + description: Reset CO2 calibration + set-autoslideingtime: + description: Set screensaver interval + set-co-two-led-th: + description: Set CO2 LED threshold + set-device-off: + description: Set auto power off + set-device-off-new: {} + set-end-time: {} + set-frequency: + description: Set monitoring frequency + set-humi-led-th: + description: Set humidity LED threshold + set-page-sequence: + description: Set screensaver order + set-pm-t-led-th: {} + set-pm-tpf-led-th: {} + set-pm-tpf-standard: {} + set-screen-off: + description: Set screen off timeout + set-screensaver-time: + description: Set screensaver timeout + set-screensaver-type: + description: Set screensaver type + set-start-time: {} + set-temp-led-th: + description: Set temperature LED threshold + set-temp-unit: + description: Set temperature unit + set-time-zone: + description: Set timezone + set-twelve-hours-sys: + description: Use 12-hour clock diff --git a/miio/integrations/genericmiot/metadata/dreamespec.yaml b/miio/integrations/genericmiot/metadata/dreamespec.yaml index 8c67abb72..3500d27f8 100644 --- a/miio/integrations/genericmiot/metadata/dreamespec.yaml +++ b/miio/integrations/genericmiot/metadata/dreamespec.yaml @@ -1,5 +1,4 @@ description: Metadata for dreame-specific services -fallback: miot-spec-v2 services: vacuum-extend: description: Extended vacuum services for dreame diff --git a/miio/integrations/genericmiot/tests/test_meta.py b/miio/integrations/genericmiot/tests/test_meta.py index 16e5837ea..55db60dd5 100644 --- a/miio/integrations/genericmiot/tests/test_meta.py +++ b/miio/integrations/genericmiot/tests/test_meta.py @@ -2,12 +2,13 @@ from unittest.mock import Mock import pytest +import yaml from miio.descriptors import AccessFlags, ActionDescriptor -from miio.miot_models import URN, MiotBaseModel +from miio.miot_models import URN, MiotBaseModel, MiotService from ..genericmiot import GenericMiot -from ..meta import MetaBase, Metadata +from ..meta import ActionMeta, MetaBase, Metadata, Namespace, PropertyMeta, ServiceMeta @pytest.fixture(scope="module") @@ -23,7 +24,7 @@ def _make_entity( service: Mock = Mock() service.name = service_name entity: Mock = Mock() - entity.extras = {"urn": urn} + entity.urn = urn entity.service = service return entity @@ -71,15 +72,37 @@ def test_action_found(meta: Metadata) -> None: assert result.description == "Start cleaning" +def test_unknown_namespace_falls_back_to_miotspec(meta: Metadata) -> None: + entity: MiotBaseModel = _make_entity( + "unknown-spec", "property", "battery-level", "battery" + ) + result = meta.get_metadata(entity) + assert result is not None + assert result.description == "Battery level" + + def test_unknown_namespace_falls_back_to_common(meta: Metadata) -> None: entity: MiotBaseModel = _make_entity( "unknown-spec", "property", "temperature", "environment" ) - result: MetaBase | None = meta.get_metadata(entity) + result = meta.get_metadata(entity) assert result is not None assert result.description == "Temperature" +def test_registered_namespace_without_fallback_reaches_miotspec(meta: Metadata) -> None: + meta_copy = Metadata( + namespaces={ + **meta.namespaces, + "no-fallback-spec": Namespace(description="no fallback"), + } + ) + entity = _make_entity("no-fallback-spec", "property", "battery-level", "battery") + result = meta_copy.get_metadata(entity) + assert result is not None + assert result.description == "Battery level" + + def test_unknown_service(meta: Metadata) -> None: entity: MiotBaseModel = _make_entity( "miot-spec-v2", "property", "battery-level", "nonexistent" @@ -96,13 +119,6 @@ def test_unknown_property(meta: Metadata) -> None: assert result is None -def test_no_urn_in_extras(meta: Metadata) -> None: - entity: Mock = Mock(spec=MiotBaseModel) - entity.extras = {} - result: MetaBase | None = meta.get_metadata(entity) - assert result is None - - def test_dreame_property(meta: Metadata) -> None: entity: MiotBaseModel = _make_entity( "dreame-spec", "property", "mop-mode", "vacuum-extend" @@ -121,34 +137,6 @@ def test_dreame_action(meta: Metadata) -> None: assert result.description == "Stop cleaning" -def test_fallback_namespace() -> None: - fallback_ns = { - "description": "fallback", - "services": { - "vacuum": { - "description": "Vacuum service", - "property": { - "status": {"description": "Status from fallback"}, - }, - } - }, - } - primary_ns = {"description": "primary", "fallback": "fallback-ns"} - common_ns = {"description": "common"} - meta = Metadata( - namespaces={ - "primary-ns": primary_ns, - "fallback-ns": fallback_ns, - "common": common_ns, - } - ) - - entity = _make_entity("primary-ns", "property", "status", "vacuum") - result = meta.get_metadata(entity) - assert result is not None - assert result.description == "Status from fallback" - - def test_implicit_common_fallback() -> None: common_ns = { "description": "common", @@ -194,9 +182,7 @@ def test_load_explicit_file() -> None: def test_no_service_returns_none(meta: Metadata) -> None: entity: Mock = Mock(spec=MiotBaseModel) - entity.extras = { - "urn": URN.model_validate("urn:miot-spec-v2:property:battery-level:1:mock:1") - } + entity.urn = URN.model_validate("urn:miot-spec-v2:property:battery-level:1:mock:1") entity.service = None assert meta.get_metadata(entity) is None @@ -233,3 +219,313 @@ def test_enrich_applies_metadata(device: GenericMiot) -> None: assert result.name == "Start cleaning" assert result.extras["original"] is desc assert result.extras["original"].name == "start-sweep" + + +def test_namespace_merge_adds_new_service() -> None: + base = Namespace(description="base", services={}) + stub = Namespace( + description="stub", + services={ + "env": ServiceMeta( + property={"temperature": PropertyMeta(description="Temperature")} + ) + }, + ) + base.merge(stub) + assert "env" in base.services + assert "temperature" in base.services["env"].property + + +def test_namespace_merge_adds_to_existing_service() -> None: + base = Namespace( + description="base", + services={ + "env": ServiceMeta( + property={"temperature": PropertyMeta(description="Temperature")} + ) + }, + ) + stub = Namespace( + description="stub", + services={ + "env": ServiceMeta( + property={"humidity": PropertyMeta(description="Humidity")} + ) + }, + ) + base.merge(stub) + assert "temperature" in base.services["env"].property + assert "humidity" in base.services["env"].property + + +def test_namespace_merge_preserves_existing_descriptions() -> None: + base = Namespace( + description="base", + services={ + "env": ServiceMeta( + property={ + "temperature": PropertyMeta(description="My custom description") + } + ) + }, + ) + stub = Namespace( + description="stub", + services={ + "env": ServiceMeta( + property={"temperature": PropertyMeta(description="temperature")} + ) + }, + ) + base.merge(stub) + assert ( + base.services["env"].property["temperature"].description + == "My custom description" + ) + + +def test_namespace_merge_adds_actions() -> None: + base = Namespace(description="base", services={"settings": ServiceMeta()}) + stub = Namespace( + description="stub", + services={ + "settings": ServiceMeta(action={"reset": ActionMeta(description="Reset")}) + }, + ) + base.merge(stub) + assert "reset" in base.services["settings"].action + + +def test_suggested_filename_uses_source_file() -> None: + ns = Namespace(description="test", source_file="existing.yaml") + meta = Metadata(namespaces={"my-spec": ns}) + assert meta.suggested_filename("my-spec") == "existing.yaml" + + +def test_suggested_filename_generates_name() -> None: + meta = Metadata(namespaces={}) + assert meta.suggested_filename("my-new-spec") == "mynewspec.yaml" + + +def test_build_namespace_metadata(meta: Metadata) -> None: + entity = _make_entity("cgllc-spec", "property", "my-prop", "settings") + entity.description = "My property" + entity.service.description = "Settings" # type: ignore[union-attr] + + ns = meta.build_namespace_metadata("cgllc-spec", {"settings": [entity]}) + + assert "settings" in ns.services + assert "my-prop" in ns.services["settings"].property + + +def test_write_namespace_metadata_creates_file(tmp_path: Path, meta: Metadata) -> None: + path = tmp_path / "test.yaml" + ns = Namespace(description="Test") + assert meta.write_namespace_metadata(ns, path) is True + assert path.exists() + + +def test_write_namespace_metadata_merges_into_existing( + tmp_path: Path, meta: Metadata +) -> None: + path = tmp_path / "test.yaml" + existing = Namespace( + description="Existing", + services={"env": ServiceMeta(property={"a": PropertyMeta(description="A")})}, + ) + path.write_text(yaml.dump(existing.model_dump(exclude_defaults=True))) + + new_ns = Namespace( + description="New", + services={"env": ServiceMeta(property={"b": PropertyMeta(description="B")})}, + ) + assert meta.write_namespace_metadata(new_ns, path) is False + + merged = Namespace.model_validate(yaml.safe_load(path.read_text())) + assert "a" in merged.services["env"].property + assert "b" in merged.services["env"].property + + +def test_multi_namespace_routing(tmp_path: Path, meta: Metadata) -> None: + """Missing entities from different namespaces go to separate files.""" + ns_a = Namespace(description="ns-a", source_file="a.yaml") + ns_b = Namespace(description="ns-b", source_file="b.yaml") + routing_meta = Metadata(namespaces={"spec-a": ns_a, "spec-b": ns_b}) + + entity_a = _make_entity("spec-a", "property", "prop", "svc") + entity_a.description = "Prop" + entity_a.service.description = "Svc" # type: ignore[union-attr] + + entity_b = _make_entity("spec-b", "property", "prop", "svc") + entity_b.description = "Prop" + entity_b.service.description = "Svc" # type: ignore[union-attr] + + missing_by_ns = {"spec-a": {"svc": [entity_a]}, "spec-b": {"svc": [entity_b]}} + for ns_name, services in missing_by_ns.items(): + ns = routing_meta.build_namespace_metadata(ns_name, services) + routing_meta.write_namespace_metadata( + ns, tmp_path / routing_meta.suggested_filename(ns_name) + ) + + assert (tmp_path / "a.yaml").exists() + assert (tmp_path / "b.yaml").exists() + + +@pytest.fixture +def battery_service() -> MiotService: + return MiotService.model_validate_json("""{ + "iid": 2, + "description": "Battery", + "type": "urn:miot-spec-v2:service:battery:00000003:dummy:1", + "properties": [{ + "iid": 1, + "type": "urn:miot-spec-v2:property:battery-level:00000014:dummy:1", + "description": "Battery Level", + "format": "uint8", + "access": ["read"] + }], + "actions": [], + "events": [] + }""") + + +@pytest.fixture +def device_info_service() -> MiotService: + return MiotService.model_validate_json("""{ + "iid": 1, + "description": "Device Information", + "type": "urn:miot-spec-v2:service:device-information:00000001:dummy:1", + "properties": [{ + "iid": 1, + "type": "urn:miot-spec-v2:property:manufacturer:00000001:dummy:1", + "description": "Manufacturer", + "format": "string", + "access": ["read"] + }], + "actions": [], + "events": [] + }""") + + +def _device_model(services: list) -> Mock: + model = Mock() + model.services = services + return model + + +@pytest.mark.parametrize( + ("meta_obj", "expected"), + [ + (PropertyMeta(description="Battery level"), "Battery level"), + (PropertyMeta(description=None), "(no description)"), + (ServiceMeta(description="Battery"), "Battery"), + (ServiceMeta(description=None), "(no description)"), + ], +) +def test_str_representation(meta_obj, expected) -> None: + assert str(meta_obj) == expected + + +def test_load_skips_missing_namespace_file(tmp_path: Path) -> None: + base = tmp_path / "base.yaml" + base.write_text(yaml.dump({"namespaces": {"miot-spec-v2": "nonexistent.yaml"}})) + + meta = Metadata.load(file=base) + + assert "miot-spec-v2" not in meta.namespaces + + +def test_build_namespace_metadata_with_action(meta: Metadata) -> None: + entity = _make_entity("cgllc-spec", "action", "my-action", "settings") + entity.description = "My action" + entity.service.description = "Settings" # type: ignore[union-attr] + + ns = meta.build_namespace_metadata("cgllc-spec", {"settings": [entity]}) + + assert "my-action" in ns.services["settings"].action + + +def test_register_namespace(tmp_path: Path, meta: Metadata) -> None: + base = tmp_path / "base.yaml" + base.write_text(yaml.dump({"namespaces": {}})) + + assert meta.register_namespace("new-spec", "newspec.yaml", base) is True + assert meta.register_namespace("new-spec", "newspec.yaml", base) is False + + +@pytest.mark.parametrize( + ("ns_name", "service", "type_", "name", "expected_desc"), + [ + ("miot-spec-v2", "battery", "property", "battery-level", "Battery level"), + ("nonexistent-spec", "battery", "property", "battery-level", None), + ], +) +def test_lookup_in_namespace( + meta: Metadata, ns_name, service, type_, name, expected_desc +) -> None: + result = meta.lookup_in_namespace(ns_name, service, type_, name) + if expected_desc is None: + assert result is None + else: + assert result is not None + assert result.description == expected_desc + + +def test_collect_coverage_ok(meta: Metadata, battery_service: MiotService) -> None: + cov = meta.collect_coverage(_device_model([battery_service])) + + assert cov.ok == 1 + assert cov.total == 1 + assert cov.missing == 0 + + +def test_collect_coverage_fallback( + meta: Metadata, battery_service: MiotService +) -> None: + dreame_battery = battery_service.model_copy(deep=True) + dreame_battery.urn.namespace = "dreame-spec" + dreame_battery.properties[0].urn.namespace = "dreame-spec" + cov = meta.collect_coverage(_device_model([dreame_battery])) + + assert cov.fb == 1 + assert cov.missing == 0 + + +def test_collect_coverage_missing(meta: Metadata, battery_service: MiotService) -> None: + unknown_service = battery_service.model_copy(deep=True) + unknown_service.urn.namespace = "unknown-spec" + unknown_service.urn.name = "unknown-svc" + unknown_service.properties[0].urn.namespace = "unknown-spec" + unknown_service.properties[0].urn.name = "unknown-prop" + cov = meta.collect_coverage(_device_model([unknown_service])) + + assert cov.missing == 1 + assert "unknown-spec" in cov.missing_by_ns + + +def test_collect_coverage_no_desc(meta: Metadata, battery_service: MiotService) -> None: + meta_no_desc = Metadata( + namespaces={ + "miot-spec-v2": Namespace( + description="miot-spec-v2", + services={ + "battery": ServiceMeta( + property={"battery-level": PropertyMeta(description=None)} + ) + }, + ) + } + ) + cov = meta_no_desc.collect_coverage(_device_model([battery_service])) + + assert cov.no_desc == 1 + assert cov.ok == 0 + assert cov.missing == 0 + + +def test_collect_coverage_skips_siid_1( + meta: Metadata, device_info_service: MiotService +) -> None: + cov = meta.collect_coverage(_device_model([device_info_service])) + + assert cov.total == 0 diff --git a/miio/miot_models.py b/miio/miot_models.py index ba567c676..6f6fa8138 100644 --- a/miio/miot_models.py +++ b/miio/miot_models.py @@ -171,6 +171,9 @@ def normalized_name(self) -> str: """ return self.name.replace(":", "_").replace("-", "_") + def __str__(self) -> str: + return f"{self.urn.type}:{self.urn.name!r}" + @property @abstractmethod def unique_identifier(self) -> str: @@ -282,10 +285,10 @@ def pretty_value(self): "days": timedelta(days=1), } - unit = unit_map.get(self.unit) + unit = unit_map.get(self.unit, self.unit) if isinstance(unit, timedelta): value = value * unit - else: + elif unit: value = f"{value} {unit}" return value @@ -480,6 +483,11 @@ def normalized_name(self) -> str: """ return self.urn.name.replace(":", "_").replace("-", "_") + def __str__(self) -> str: + return ( + f"Service: {self.description} (siid={self.siid}, ns={self.urn.namespace})" + ) + model_config = ConfigDict(extra="allow")