From 528a59dc62bf5859019311ec88b0508f54b32f62 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 17:14:39 +0300 Subject: [PATCH 01/11] Simplify metadata lookup and fallback chain Unify the lookup API so get_metadata returns the source namespace on the result, removing the need for a separate paired method. Make the miot-spec-v2 -> common fallback implicit for all namespaces, dropping the per-namespace fallback field from the YAML files. --- miio/integrations/genericmiot/meta.py | 167 ++++++++++++++---- .../genericmiot/metadata/dreamespec.yaml | 1 - .../genericmiot/tests/test_meta.py | 142 ++++++++++----- miio/miot_models.py | 12 +- 4 files changed, 242 insertions(+), 80 deletions(-) diff --git a/miio/integrations/genericmiot/meta.py b/miio/integrations/genericmiot/meta.py index 731f03ca7..f51112df4 100644 --- a/miio/integrations/genericmiot/meta.py +++ b/miio/integrations/genericmiot/meta.py @@ -14,28 +14,39 @@ 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 +55,20 @@ 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 +90,139 @@ 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 to use for a namespace stub.""" + 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 + return None - common = self.namespaces.get("common") - fallback_ns = self.namespaces.get(ns.fallback or "common", common) - - 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_stub( + self, + ns_name: str, + missing: dict[str, list[MiotBaseModel]], + ) -> "Namespace": + """Build a stub Namespace for entities that lack metadata 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 Namespace( + description=f"Metadata for {ns_name} namespace", + services=services, + ) - return None + def register_namespace(self, ns_name: str, filename: str, base_file: Path) -> None: + """Add a namespace entry to the index file if not already listed.""" + data = yaml.safe_load(base_file.read_text()) + if ns_name not in data["namespaces"]: + data["namespaces"][ns_name] = filename + base_file.write_text( + yaml.dump( + data, + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + ) - def get_metadata(self, entity: MiotBaseModel) -> MetaBase | None: - """Look up metadata for a miot entity (property or action). + def write_namespace_stub(self, stub: "Namespace", path: Path) -> bool: + """Write a namespace stub 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(stub) + stub = existing + data = stub.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 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 + ns_name: str = entity.urn.namespace service_name: str = entity.service.name - type_: str = urn.type - entity_name: str = urn.name - - ns = self.namespaces.get(ns_name, self.namespaces["common"]) - - 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 - ) - return None + type_: str = entity.urn.type + entity_name: str = entity.urn.name + + 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 not None: + _LOGGER.debug( + "Found metadata for %s:%s:%s:%s in %s", + ns_name, + service_name, + type_, + entity_name, + 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, + "No metadata for %s:%s:%s:%s", ns_name, service_name, type_, entity_name ) - return meta + return None 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..c380963f4 100644 --- a/miio/integrations/genericmiot/tests/test_meta.py +++ b/miio/integrations/genericmiot/tests/test_meta.py @@ -7,7 +7,7 @@ from miio.miot_models import URN, MiotBaseModel from ..genericmiot import GenericMiot -from ..meta import MetaBase, Metadata +from ..meta import ActionMeta, MetaBase, Metadata, Namespace, PropertyMeta, ServiceMeta @pytest.fixture(scope="module") @@ -23,7 +23,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 +71,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 +118,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 +136,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 +181,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 +218,78 @@ 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 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") From 84a9a5745eb7719549661b73abb7de1500424157 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 17:15:09 +0300 Subject: [PATCH 02/11] Add metadata coverage command to GenericMiot Adds a 'metadata' command that reports per-entity coverage against the loaded namespace metadata, distinguishing direct hits from fallbacks. Missing items can be saved as YAML stubs, which are merged into existing namespace files and automatically registered in the index. --- miio/integrations/genericmiot/genericmiot.py | 147 +++++++++++++++++- miio/integrations/genericmiot/meta.py | 38 ++--- .../genericmiot/tests/test_meta.py | 76 +++++++++ 3 files changed, 239 insertions(+), 22 deletions(-) diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 0125d578a..2826d7683 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -1,11 +1,15 @@ import logging +from collections import defaultdict from functools import partial -from typing import TypeVar +from pathlib import Path +from typing import NamedTuple, TypeVar 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 @@ -24,6 +28,15 @@ _D = TypeVar("_D", ActionDescriptor, PropertyDescriptor) +class _CoverageResult(NamedTuple): + total: int + ok: int + fb: int + missing: int + no_desc: int + missing_by_ns: dict + + class GenericMiot(MiotDevice): # we support all devices, if not, it is a responsibility of caller to verify that _supported_models = ["*"] @@ -89,7 +102,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) @@ -151,7 +164,7 @@ def _create_descriptors(self): """Create descriptors based on the miot model.""" for serv in self._miot_model.services: if serv.siid == 1: - continue # Skip device details + continue self._create_actions(serv) self._create_properties(serv) @@ -181,6 +194,132 @@ def device_type(self) -> str | None: return self._miot_model.urn.type return None + def _collect_coverage( + self, + miot_model: DeviceModel, + ) -> _CoverageResult: + """Report per-entity metadata coverage for the device model.""" + missing_by_ns: dict = defaultdict(dict) + total = ok = fb = missing = no_desc = 0 + + for serv in miot_model.services: + if serv.siid == 1: + continue + + ns_name = serv.urn.namespace + click.echo(f"\n{serv}") + + nd_lines: list[str] = [] + for entity in [*serv.properties, *serv.actions]: + total += 1 + + direct = self._meta.lookup_in_namespace( + ns_name, serv.name, entity.urn.type, entity.urn.name + ) + if direct: + if direct.description is None: + no_desc += 1 + nd_lines.append( + f" [??] {entity!s:50} (fill in description if known)" + ) + else: + ok += 1 + click.echo(f" [ok] {entity!s:50} -> {direct}") + continue + + fallback = self._meta.get_metadata(entity) + if fallback: + if fallback.description is None: + no_desc += 1 + nd_lines.append( + f" [??] {entity!s:50} (fill in description if known)" + ) + else: + fb += 1 + click.echo( + f" [fb] {entity!s:50} -> {fallback} ({fallback.source})" + ) + else: + missing += 1 + click.echo(f" [--] {entity!s:50} {entity.description!r}") + missing_by_ns[ns_name].setdefault(serv.name, []).append(entity) + + for line in nd_lines: + click.echo(line) + + return _CoverageResult(total, ok, fb, missing, no_desc, missing_by_ns) + + @command( + click.option( + "--generate-template", + is_flag=True, + default=False, + help="Print namespace metadata YAML for entities that need coverage.", + ), + 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, generate_template: bool = False, output_dir: str | None = None): + """Show metadata coverage and optionally generate YAML templates for missing items.""" + if not self._initialized: + self._initialize_descriptors() + + if self._miot_model is None: + raise RuntimeError("Device model not initialized") + + cov = self._collect_coverage(self._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 + + if not generate_template and output_dir is None: + click.echo(f"{cov.missing} items need namespace metadata.") + if click.confirm("Save namespace metadata?", default=False): + default_dir = str(Path(__file__).parent / "metadata") + output_dir = click.prompt("Output directory", default=default_dir) + else: + generate_template = True + + for ns_name, services in cov.missing_by_ns.items(): + ns_meta = self._meta.build_namespace_metadata(ns_name, services) + yaml_text = yaml.dump( + ns_meta.model_dump(exclude_defaults=True), + default_flow_style=False, + sort_keys=False, + allow_unicode=True, + ) + 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_text) + @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 f51112df4..d4b7acf18 100644 --- a/miio/integrations/genericmiot/meta.py +++ b/miio/integrations/genericmiot/meta.py @@ -109,7 +109,7 @@ def load(cls, file: Path | None = None) -> "Metadata": return cls(**data) def suggested_filename(self, ns_name: str) -> str: - """Return the filename to use for a namespace stub.""" + """Return the filename for a namespace metadata file.""" ns = self.namespaces.get(ns_name) if ns and ns.source_file: return ns.source_file @@ -126,12 +126,12 @@ def _lookup_in_namespace( return meta return None - def build_namespace_stub( + def build_namespace_metadata( self, ns_name: str, missing: dict[str, list[MiotBaseModel]], ) -> "Namespace": - """Build a stub Namespace for entities that lack metadata coverage.""" + """Build a Namespace with template entries for entities that lack coverage.""" services = {} for svc_name, entities in missing.items(): props = {} @@ -154,28 +154,30 @@ def build_namespace_stub( services=services, ) - def register_namespace(self, ns_name: str, filename: str, base_file: Path) -> None: + 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 not in data["namespaces"]: - data["namespaces"][ns_name] = filename - base_file.write_text( - yaml.dump( - data, - default_flow_style=False, - sort_keys=False, - allow_unicode=True, - ) + 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_stub(self, stub: "Namespace", path: Path) -> bool: - """Write a namespace stub to a file, merging into any existing content.""" + 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(stub) - stub = existing - data = stub.model_dump(exclude_defaults=True) + 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: diff --git a/miio/integrations/genericmiot/tests/test_meta.py b/miio/integrations/genericmiot/tests/test_meta.py index c380963f4..259c2a636 100644 --- a/miio/integrations/genericmiot/tests/test_meta.py +++ b/miio/integrations/genericmiot/tests/test_meta.py @@ -2,6 +2,7 @@ from unittest.mock import Mock import pytest +import yaml from miio.descriptors import AccessFlags, ActionDescriptor from miio.miot_models import URN, MiotBaseModel @@ -293,3 +294,78 @@ def test_namespace_merge_adds_actions() -> None: ) 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() From b51f3071b78c691391046bafcf1fa6f80b601773 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 17:59:14 +0300 Subject: [PATCH 03/11] Add cgllc-spec namespace metadata for cgllc.airm.cgd1st Known settings get human-readable descriptions; entries with unclear purpose use empty entries to stay registered without name overrides. --- .../genericmiot/metadata/base.yaml | 1 + .../genericmiot/metadata/cgllcspec.yaml | 76 +++++++++++++++++++ 2 files changed, 77 insertions(+) create mode 100644 miio/integrations/genericmiot/metadata/cgllcspec.yaml 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..288d6f0ba --- /dev/null +++ b/miio/integrations/genericmiot/metadata/cgllcspec.yaml @@ -0,0 +1,76 @@ +description: Metadata for cgllc-spec namespace +services: + settings: + description: Device settings + property: + auto-slideing-time: + description: Auto slideshow interval + carbondioxide-led-th: + description: CO2 LED threshold + device-off: + description: Auto power off + device-off-new: {} + end-time: + description: End time + humi-led-th: + description: Humidity LED threshold + is-twelve-hours-sys: + description: 12-hour clock + monitoring-frequency: + description: Monitoring frequency + page-sequence: + description: Display page 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: + description: 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 auto slideshow 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: + description: Set end time + set-frequency: + description: Set monitoring frequency + set-humi-led-th: + description: Set humidity LED threshold + set-page-sequence: + description: Set display page 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: + description: 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: Set 12-hour clock From 56352332bbd561b019b6ace5feb675dff67e2cb9 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 18:19:27 +0300 Subject: [PATCH 04/11] Simplify debug logging in get_metadata to use the entity directly --- miio/integrations/genericmiot/meta.py | 23 +++++++---------------- 1 file changed, 7 insertions(+), 16 deletions(-) diff --git a/miio/integrations/genericmiot/meta.py b/miio/integrations/genericmiot/meta.py index d4b7acf18..2f5567dcc 100644 --- a/miio/integrations/genericmiot/meta.py +++ b/miio/integrations/genericmiot/meta.py @@ -203,28 +203,19 @@ def get_metadata(self, entity: MiotBaseModel) -> MetaBase | None: if entity.service is None: return None - ns_name: str = entity.urn.namespace - service_name: str = entity.service.name - type_: str = entity.urn.type - entity_name: str = entity.urn.name + ns_name = entity.urn.namespace + service_name = entity.service.name 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) + meta = self._lookup_in_namespace( + ns, service_name, entity.urn.type, entity.urn.name + ) if meta is not None: - _LOGGER.debug( - "Found metadata for %s:%s:%s:%s in %s", - ns_name, - service_name, - type_, - entity_name, - try_name, - ) + _LOGGER.debug("Found metadata for %s in %s", entity, try_name) return meta.model_copy(update={"source": try_name}) - _LOGGER.debug( - "No metadata for %s:%s:%s:%s", ns_name, service_name, type_, entity_name - ) + _LOGGER.debug("No metadata for %s", entity) return None From 8023c4f274e5815cd3c799212a5be3afc6e3149d Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 18:27:57 +0300 Subject: [PATCH 05/11] Remove --generate-template flag from metadata command --- miio/integrations/genericmiot/genericmiot.py | 31 ++++++-------------- 1 file changed, 9 insertions(+), 22 deletions(-) diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 2826d7683..499ed3cf5 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -250,12 +250,6 @@ def _collect_coverage( return _CoverageResult(total, ok, fb, missing, no_desc, missing_by_ns) @command( - click.option( - "--generate-template", - is_flag=True, - default=False, - help="Print namespace metadata YAML for entities that need coverage.", - ), click.option( "--output-dir", type=click.Path(file_okay=False), @@ -264,7 +258,7 @@ def _collect_coverage( ), default_output=format_output("", ""), ) - def metadata(self, generate_template: bool = False, output_dir: str | None = None): + 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() @@ -290,22 +284,8 @@ def metadata(self, generate_template: bool = False, output_dir: str | None = Non click.echo("All entities are covered.") return - if not generate_template and output_dir is None: - click.echo(f"{cov.missing} items need namespace metadata.") - if click.confirm("Save namespace metadata?", default=False): - default_dir = str(Path(__file__).parent / "metadata") - output_dir = click.prompt("Output directory", default=default_dir) - else: - generate_template = True - for ns_name, services in cov.missing_by_ns.items(): ns_meta = self._meta.build_namespace_metadata(ns_name, services) - yaml_text = yaml.dump( - ns_meta.model_dump(exclude_defaults=True), - default_flow_style=False, - sort_keys=False, - allow_unicode=True, - ) suggested = self._meta.suggested_filename(ns_name) if output_dir is not None: @@ -318,7 +298,14 @@ def metadata(self, generate_template: bool = False, output_dir: str | None = Non click.echo(f"Registered in {base_file}") else: click.echo(f"\n--- {ns_name} (save as {suggested}) ---") - click.echo(yaml_text) + 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): From 6f798ba881e9dd30930eb4df4c617579de318295 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 18:51:27 +0300 Subject: [PATCH 06/11] Move coverage computation to Metadata.collect_coverage --- miio/integrations/genericmiot/genericmiot.py | 69 +++++--------------- miio/integrations/genericmiot/meta.py | 47 ++++++++++++- 2 files changed, 64 insertions(+), 52 deletions(-) diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 499ed3cf5..901b538a1 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -1,8 +1,7 @@ import logging -from collections import defaultdict from functools import partial from pathlib import Path -from typing import NamedTuple, TypeVar +from typing import TypeVar, cast import attr import click @@ -28,15 +27,6 @@ _D = TypeVar("_D", ActionDescriptor, PropertyDescriptor) -class _CoverageResult(NamedTuple): - total: int - ok: int - fb: int - missing: int - no_desc: int - missing_by_ns: dict - - class GenericMiot(MiotDevice): # we support all devices, if not, it is a responsibility of caller to verify that _supported_models = ["*"] @@ -194,79 +184,56 @@ def device_type(self) -> str | None: return self._miot_model.urn.type return None - def _collect_coverage( - self, - miot_model: DeviceModel, - ) -> _CoverageResult: - """Report per-entity metadata coverage for the device model.""" - missing_by_ns: dict = defaultdict(dict) - total = ok = fb = missing = no_desc = 0 + @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 - - ns_name = serv.urn.namespace click.echo(f"\n{serv}") - + ns_name = serv.urn.namespace nd_lines: list[str] = [] for entity in [*serv.properties, *serv.actions]: - total += 1 - direct = self._meta.lookup_in_namespace( ns_name, serv.name, entity.urn.type, entity.urn.name ) if direct: if direct.description is None: - no_desc += 1 nd_lines.append( f" [??] {entity!s:50} (fill in description if known)" ) else: - ok += 1 click.echo(f" [ok] {entity!s:50} -> {direct}") continue - fallback = self._meta.get_metadata(entity) if fallback: if fallback.description is None: - no_desc += 1 nd_lines.append( f" [??] {entity!s:50} (fill in description if known)" ) else: - fb += 1 click.echo( f" [fb] {entity!s:50} -> {fallback} ({fallback.source})" ) else: - missing += 1 click.echo(f" [--] {entity!s:50} {entity.description!r}") - missing_by_ns[ns_name].setdefault(serv.name, []).append(entity) - for line in nd_lines: click.echo(line) - return _CoverageResult(total, ok, fb, missing, no_desc, missing_by_ns) - - @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() - - if self._miot_model is None: - raise RuntimeError("Device model not initialized") - - cov = self._collect_coverage(self._miot_model) + cov = self._meta.collect_coverage(miot_model) click.echo( f"\nCoverage: {cov.ok} ok, {cov.fb} via fallback, " diff --git a/miio/integrations/genericmiot/meta.py b/miio/integrations/genericmiot/meta.py index 2f5567dcc..acfa9d82b 100644 --- a/miio/integrations/genericmiot/meta.py +++ b/miio/integrations/genericmiot/meta.py @@ -1,15 +1,28 @@ 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.""" @@ -198,6 +211,38 @@ def lookup_in_namespace( 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: From 0b15065917d9d0dd5e700db038c396dd029180df Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 18:55:28 +0300 Subject: [PATCH 07/11] Improve descriptions --- .../genericmiot/metadata/cgllcspec.yaml | 24 ++++++++----------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/miio/integrations/genericmiot/metadata/cgllcspec.yaml b/miio/integrations/genericmiot/metadata/cgllcspec.yaml index 288d6f0ba..e4624cdad 100644 --- a/miio/integrations/genericmiot/metadata/cgllcspec.yaml +++ b/miio/integrations/genericmiot/metadata/cgllcspec.yaml @@ -4,22 +4,21 @@ services: description: Device settings property: auto-slideing-time: - description: Auto slideshow interval + description: Screensaver interval carbondioxide-led-th: description: CO2 LED threshold device-off: description: Auto power off device-off-new: {} - end-time: - description: End time + end-time: {} humi-led-th: description: Humidity LED threshold is-twelve-hours-sys: - description: 12-hour clock + description: Use 12-hour clock monitoring-frequency: description: Monitoring frequency page-sequence: - description: Display page order + description: Screensaver order pm-t-led-th: {} pm-tpf-led-th: {} pm-tpf-standard: {} @@ -29,8 +28,7 @@ services: description: Screensaver timeout screensaver-type: description: Screensaver type - start-time: - description: Start time + start-time: {} temp-led-th: description: Temperature LED threshold tempature-unit: @@ -41,20 +39,19 @@ services: reset-carbon-dioxide: description: Reset CO2 calibration set-autoslideingtime: - description: Set auto slideshow interval + 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: - description: Set end time + set-end-time: {} set-frequency: description: Set monitoring frequency set-humi-led-th: description: Set humidity LED threshold set-page-sequence: - description: Set display page order + description: Set screensaver order set-pm-t-led-th: {} set-pm-tpf-led-th: {} set-pm-tpf-standard: {} @@ -64,8 +61,7 @@ services: description: Set screensaver timeout set-screensaver-type: description: Set screensaver type - set-start-time: - description: Set start time + set-start-time: {} set-temp-led-th: description: Set temperature LED threshold set-temp-unit: @@ -73,4 +69,4 @@ services: set-time-zone: description: Set timezone set-twelve-hours-sys: - description: Set 12-hour clock + description: Use 12-hour clock From f4add3bb3293d60f386b1015ec9f26f2e7a2037f Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 18:58:01 +0300 Subject: [PATCH 08/11] Remove unrelated change --- miio/integrations/genericmiot/genericmiot.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 901b538a1..0db243f5e 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -154,7 +154,7 @@ def _create_descriptors(self): """Create descriptors based on the miot model.""" for serv in self._miot_model.services: if serv.siid == 1: - continue + continue # Skip device details self._create_actions(serv) self._create_properties(serv) From fd57a45438b6e305a0f0e430d352ec180b9b3d08 Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 19:04:45 +0300 Subject: [PATCH 09/11] Separate logical units with blank lines for readability --- miio/integrations/genericmiot/genericmiot.py | 5 +++++ miio/integrations/genericmiot/meta.py | 8 ++++++++ 2 files changed, 13 insertions(+) diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 0db243f5e..8fac0b41f 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -123,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( @@ -162,6 +163,7 @@ def _create_descriptors(self): _LOGGER.debug("Created %s actions", len(self._actions)) for act in self._actions.values(): _LOGGER.debug(f"\t{act}") + _LOGGER.debug("Created %s properties", len(self._properties)) for sensor in self._properties.values(): _LOGGER.debug(f"\t{sensor}") @@ -203,6 +205,7 @@ def metadata(self, output_dir: str | None = None): 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] = [] @@ -218,6 +221,7 @@ def metadata(self, output_dir: str | None = None): else: click.echo(f" [ok] {entity!s:50} -> {direct}") continue + fallback = self._meta.get_metadata(entity) if fallback: if fallback.description is None: @@ -230,6 +234,7 @@ def metadata(self, output_dir: str | None = None): ) else: click.echo(f" [--] {entity!s:50} {entity.description!r}") + for line in nd_lines: click.echo(line) diff --git a/miio/integrations/genericmiot/meta.py b/miio/integrations/genericmiot/meta.py index acfa9d82b..99fb768c3 100644 --- a/miio/integrations/genericmiot/meta.py +++ b/miio/integrations/genericmiot/meta.py @@ -77,9 +77,11 @@ def merge(self, other: "Namespace") -> None: 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) @@ -111,6 +113,7 @@ def load(cls, file: Path | None = None) -> "Metadata": _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: ns_data = yaml.safe_load(f) @@ -156,12 +159,14 @@ def build_namespace_metadata( ) 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 Namespace( description=f"Metadata for {ns_name} namespace", services=services, @@ -219,6 +224,7 @@ def collect_coverage(self, device_model: DeviceModel) -> CoverageResult: 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 @@ -231,6 +237,7 @@ def collect_coverage(self, device_model: DeviceModel) -> CoverageResult: else: ok += 1 continue + fallback = self.get_metadata(entity) if fallback: if fallback.description is None: @@ -255,6 +262,7 @@ def get_metadata(self, entity: MiotBaseModel) -> MetaBase | None: ns = self.namespaces.get(try_name) if ns is None: continue + meta = self._lookup_in_namespace( ns, service_name, entity.urn.type, entity.urn.name ) From 8f9f7852df19a1db8b5ac8d5999d2e9a3fc9a7ef Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 19:19:19 +0300 Subject: [PATCH 10/11] Cleanup --- miio/integrations/genericmiot/genericmiot.py | 1 - miio/integrations/genericmiot/meta.py | 10 +++++++++- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/miio/integrations/genericmiot/genericmiot.py b/miio/integrations/genericmiot/genericmiot.py index 8fac0b41f..c1803ba63 100644 --- a/miio/integrations/genericmiot/genericmiot.py +++ b/miio/integrations/genericmiot/genericmiot.py @@ -163,7 +163,6 @@ def _create_descriptors(self): _LOGGER.debug("Created %s actions", len(self._actions)) for act in self._actions.values(): _LOGGER.debug(f"\t{act}") - _LOGGER.debug("Created %s properties", len(self._properties)) for sensor in self._properties.values(): _LOGGER.debug(f"\t{sensor}") diff --git a/miio/integrations/genericmiot/meta.py b/miio/integrations/genericmiot/meta.py index 99fb768c3..bea33e67f 100644 --- a/miio/integrations/genericmiot/meta.py +++ b/miio/integrations/genericmiot/meta.py @@ -81,7 +81,6 @@ def merge(self, other: "Namespace") -> None: 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) @@ -119,6 +118,7 @@ def load(cls, file: Path | None = None) -> "Metadata": 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] @@ -129,6 +129,7 @@ def suggested_filename(self, ns_name: str) -> str: 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( @@ -140,6 +141,7 @@ def _lookup_in_namespace( meta := serv.get(type_, entity_name) ): return meta + return None def build_namespace_metadata( @@ -177,6 +179,7 @@ def register_namespace(self, ns_name: str, filename: str, base_file: Path) -> bo 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( @@ -186,6 +189,7 @@ def register_namespace(self, ns_name: str, filename: str, base_file: Path) -> bo allow_unicode=True, ) ) + return True def write_namespace_metadata(self, ns_meta: "Namespace", path: Path) -> bool: @@ -195,16 +199,19 @@ def write_namespace_metadata(self, ns_meta: "Namespace", path: Path) -> bool: 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 def lookup_in_namespace( @@ -214,6 +221,7 @@ def lookup_in_namespace( 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: From b040b127131f262b987a14c622186a1b0b40e80b Mon Sep 17 00:00:00 2001 From: Teemu Rytilahti Date: Sat, 25 Jul 2026 19:39:52 +0300 Subject: [PATCH 11/11] code coverage --- .../genericmiot/tests/test_meta.py | 162 +++++++++++++++++- 1 file changed, 161 insertions(+), 1 deletion(-) diff --git a/miio/integrations/genericmiot/tests/test_meta.py b/miio/integrations/genericmiot/tests/test_meta.py index 259c2a636..55db60dd5 100644 --- a/miio/integrations/genericmiot/tests/test_meta.py +++ b/miio/integrations/genericmiot/tests/test_meta.py @@ -5,7 +5,7 @@ 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 ActionMeta, MetaBase, Metadata, Namespace, PropertyMeta, ServiceMeta @@ -369,3 +369,163 @@ def test_multi_namespace_routing(tmp_path: Path, meta: Metadata) -> None: 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