From 0797fc218788aba06daf53cff31842fec3c17198 Mon Sep 17 00:00:00 2001 From: Teemu R Date: Mon, 7 Nov 2022 22:50:21 +0100 Subject: [PATCH] Add models to parse miotspec files to miio module (#1577) This moves (and extends) the models previously used only by the miot simulator to the main module. This enables creating a generic miot integration that will be added in a separate PR. --- miio/devtools/simulators/miotsimulator.py | 2 +- miio/devtools/simulators/models.py | 106 -------- miio/miot_models.py | 280 ++++++++++++++++++++++ miio/tests/test_miot_models.py | 155 ++++++++++++ 4 files changed, 436 insertions(+), 107 deletions(-) delete mode 100644 miio/devtools/simulators/models.py create mode 100644 miio/miot_models.py create mode 100644 miio/tests/test_miot_models.py diff --git a/miio/devtools/simulators/miotsimulator.py b/miio/devtools/simulators/miotsimulator.py index cea12e078..b806ec09d 100644 --- a/miio/devtools/simulators/miotsimulator.py +++ b/miio/devtools/simulators/miotsimulator.py @@ -8,9 +8,9 @@ from pydantic import Field, validator from miio import PushServer +from miio.miot_models import DeviceModel, MiotProperty, MiotService from .common import create_info_response, mac_from_model -from .models import DeviceModel, MiotProperty, MiotService _LOGGER = logging.getLogger(__name__) UNSET = -10000 diff --git a/miio/devtools/simulators/models.py b/miio/devtools/simulators/models.py deleted file mode 100644 index f6928542a..000000000 --- a/miio/devtools/simulators/models.py +++ /dev/null @@ -1,106 +0,0 @@ -import logging -from typing import Any, List, Optional - -from pydantic import BaseModel, Field - -_LOGGER = logging.getLogger(__name__) - - -class MiotFormat(type): - """Custom type to convert textual presentation to python type.""" - - @classmethod - def __get_validators__(cls): - yield cls.convert_type - - @classmethod - def convert_type(cls, input: str): - if input.startswith("uint") or input.startswith("int"): - return int - type_map = { - "bool": bool, - "string": str, - "float": float, - } - return type_map[input] - - @classmethod - def serialize(cls, v): - return str(v) - - -class MiotEvent(BaseModel): - """Presentation of miot event.""" - - description: str - eiid: int = Field(alias="iid") - urn: str = Field(alias="type") - arguments: Any - - class Config: - extra = "forbid" - - -class MiotEnumValue(BaseModel): - """Enum value for miot.""" - - description: str - value: int - - class Config: - extra = "forbid" - - -class MiotAction(BaseModel): - """Action presentation for miot.""" - - description: str - aiid: int = Field(alias="iid") - urn: str = Field(alias="type") - inputs: Any = Field(alias="in") - output: Any = Field(alias="out") - - class Config: - extra = "forbid" - - -class MiotProperty(BaseModel): - """Property presentation for miot.""" - - description: str - piid: int = Field(alias="iid") - urn: str = Field(alias="type") - unit: str = Field(default="unknown") - format: MiotFormat - access: Any = Field(default=["read"]) - range: Optional[List[int]] = Field(alias="value-range") - choices: Optional[List[MiotEnumValue]] = Field(alias="value-list") - - class Config: - extra = "forbid" - - -class MiotService(BaseModel): - """Service presentation for miot.""" - - description: str - siid: int = Field(alias="iid") - urn: str = Field(alias="type") - properties: List[MiotProperty] = Field(default=[], repr=False) - events: Optional[List[MiotEvent]] = Field(default=[], repr=False) - actions: Optional[List[MiotAction]] = Field(default=[], repr=False) - - class Config: - extra = "forbid" - - -class DeviceModel(BaseModel): - """Device presentation for miot.""" - - description: str - urn: str = Field(alias="type") - services: List[MiotService] = Field(repr=False) - model: Optional[str] = None - - class Config: - extra = "forbid" diff --git a/miio/miot_models.py b/miio/miot_models.py new file mode 100644 index 000000000..509c6f888 --- /dev/null +++ b/miio/miot_models.py @@ -0,0 +1,280 @@ +import logging +from datetime import timedelta +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field, PrivateAttr, root_validator + +_LOGGER = logging.getLogger(__name__) + + +class URN(BaseModel): + """Parsed type URN.""" + + namespace: str + type: str + name: str + internal_id: str + model: str + version: int + + @classmethod + def __get_validators__(cls): + yield cls.validate + + @classmethod + def validate(cls, v): + if not isinstance(v, str) or ":" not in v: + raise TypeError("invalid type") + + _, namespace, type, name, id_, model, version = v.split(":") + + return cls( + namespace=namespace, + type=type, + name=name, + internal_id=id_, + model=model, + version=version, + ) + + def __repr__(self): + return f"urn:{self.namespace}:{self.type}:{self.name}:{self.internal_id}:{self.model}:{self.version}" + + +class MiotFormat(type): + """Custom type to convert textual presentation to python type.""" + + @classmethod + def __get_validators__(cls): + yield cls.convert_type + + @classmethod + def convert_type(cls, input: str): + if input.startswith("uint") or input.startswith("int"): + return int + type_map = { + "bool": bool, + "string": str, + "float": float, + } + return type_map[input] + + +class MiotEvent(BaseModel): + """Presentation of miot event.""" + + eiid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + arguments: Any + + service: Optional["MiotService"] = None # backref to containing service + + class Config: + extra = "forbid" + + +class MiotEnumValue(BaseModel): + """Enum value for miot.""" + + description: str + value: int + + @root_validator + def description_from_value(cls, values): + """If description is empty, use the value instead.""" + if not values["description"]: + values["description"] = str(values["value"]) + return values + + class Config: + extra = "forbid" + + +class MiotAction(BaseModel): + """Action presentation for miot.""" + + aiid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + inputs: Any = Field(alias="in") + outputs: Any = Field(alias="out") + + extras: Dict = Field(default_factory=dict, repr=False) + + service: Optional["MiotService"] = None # backref to containing service + + @property + def siid(self) -> Optional[int]: + """Return siid.""" + if self.service is not None: + return self.service.siid + + return None + + @property + def plain_name(self) -> str: + """Return plain name.""" + return self.urn.name + + @property + def name(self) -> str: + """Return combined name of the service and the action.""" + return f"{self.service.name}:{self.urn.name}" # type: ignore + + class Config: + extra = "forbid" + + +class MiotProperty(BaseModel): + """Property presentation for miot.""" + + piid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + format: MiotFormat + access: Any = Field(default=["read"]) + unit: Optional[str] = None + + range: Optional[List[int]] = Field(alias="value-range") + choices: Optional[List[MiotEnumValue]] = Field(alias="value-list") + + extras: Dict[str, Any] = Field(default_factory=dict, repr=False) + + service: Optional["MiotService"] = None # backref to containing service + + # TODO: currently just used to pass the data for miiocli + # there must be a better way to do this.. + value: Optional[Any] = None + + @property + def siid(self) -> Optional[int]: + """Return siid.""" + if self.service is not None: + return self.service.siid + + return None + + @property + def plain_name(self): + """Return plain name.""" + return self.urn.name + + @property + def name(self) -> str: + """Return combined name of the service and the property.""" + return f"{self.service.name}:{self.urn.name}" # type: ignore + + @property + def pretty_value(self): + value = self.value + + if self.choices is not None: + # TODO: find a nicer way to get the choice by value + selected = next(c.description for c in self.choices if c.value == value) + current = f"{selected} (value: {value})" + return current + + if self.format == bool: + return bool(value) + + unit_map = { + "none": "", + "percentage": "%", + "minutes": timedelta(minutes=1), + "hours": timedelta(hours=1), + "days": timedelta(days=1), + } + + unit = unit_map.get(self.unit) + if isinstance(unit, timedelta): + value = value * unit + else: + value = f"{value} {unit}" + + return value + + class Config: + extra = "forbid" + + +class MiotService(BaseModel): + """Service presentation for miot.""" + + siid: int = Field(alias="iid") + urn: URN = Field(alias="type") + description: str + + properties: List[MiotProperty] = Field(default_factory=list, repr=False) + events: List[MiotEvent] = Field(default_factory=list, repr=False) + actions: List[MiotAction] = Field(default_factory=list, repr=False) + + def __init__(self, *args, **kwargs): + """Initialize a service. + + Overridden to propagate the siid to the children. + """ + super().__init__(*args, **kwargs) + + for prop in self.properties: + prop.service = self + for act in self.actions: + act.service = self + for ev in self.events: + ev.service = self + + @property + def name(self) -> str: + """Return service name.""" + return self.urn.name + + class Config: + extra = "forbid" + + +class DeviceModel(BaseModel): + """Device presentation for miot.""" + + description: str + urn: URN = Field(alias="type") + services: List[MiotService] = Field(repr=False) + + # internal mappings to simplify accesses to a specific (siid, piid) + _properties_by_id: Dict[int, Dict[int, MiotProperty]] = PrivateAttr( + default_factory=dict + ) + _properties_by_name: Dict[str, Dict[str, MiotProperty]] = PrivateAttr( + default_factory=dict + ) + + def __init__(self, *args, **kwargs): + """Presentation of a miot device model scehma. + + Overridden to implement internal (siid, piid) mapping. + """ + super().__init__(*args, **kwargs) + for serv in self.services: + self._properties_by_name[serv.name] = dict() + self._properties_by_id[serv.siid] = dict() + for prop in serv.properties: + self._properties_by_name[serv.name][prop.plain_name] = prop + self._properties_by_id[serv.siid][prop.piid] = prop + + @property + def device_type(self) -> str: + """Return device type as string.""" + return self.urn.type + + def get_property(self, service: str, prop_name: str) -> MiotProperty: + """Return the property model for given service and property name.""" + return self._properties_by_name[service][prop_name] + + def get_property_by_siid_piid(self, siid: int, piid: int) -> MiotProperty: + """Return the property model for given siid, piid.""" + return self._properties_by_id[siid][piid] + + class Config: + extra = "forbid" diff --git a/miio/tests/test_miot_models.py b/miio/tests/test_miot_models.py new file mode 100644 index 000000000..dcca795cc --- /dev/null +++ b/miio/tests/test_miot_models.py @@ -0,0 +1,155 @@ +"""Tests for miot model parsing.""" + +import pytest +from pydantic import BaseModel + +from miio.miot_models import ( + URN, + MiotAction, + MiotEnumValue, + MiotEvent, + MiotFormat, + MiotProperty, + MiotService, +) + + +def test_enum(): + """Test that enum parsing works.""" + data = """ + { + "value": 1, + "description": "dummy" + }""" + en = MiotEnumValue.parse_raw(data) + assert en.value == 1 + assert en.description == "dummy" + + +def test_enum_missing_description(): + """Test that missing description gets replaced by the value.""" + data = '{"value": 1, "description": ""}' + en = MiotEnumValue.parse_raw(data) + assert en.value == 1 + assert en.description == "1" + + +TYPES_FOR_FORMAT = [ + ("bool", bool), + ("string", str), + ("float", float), + ("uint8", int), + ("uint16", int), + ("uint32", int), + ("int8", int), + ("int16", int), + ("int32", int), +] + + +@pytest.mark.parametrize("format,expected_type", TYPES_FOR_FORMAT) +def test_format(format, expected_type): + class Wrapper(BaseModel): + """Need to wrap as plain string is not valid json.""" + + format: MiotFormat + + data = f'{{"format": "{format}"}}' + f = Wrapper.parse_raw(data) + assert f.format == expected_type + + +def test_action(): + """Test the public properties of action.""" + simple_action = """ + { + "iid": 1, + "type": "urn:miot-spec-v2:action:dummy-action:0000001:dummy:1", + "description": "Description", + "in": [], + "out": [] + }""" + act = MiotAction.parse_raw(simple_action) + assert act.aiid == 1 + assert act.urn.type == "action" + assert act.description == "Description" + assert act.inputs == [] + assert act.outputs == [] + + assert act.plain_name == "dummy-action" + + +def test_urn(): + """Test the parsing of URN strings.""" + urn_string = "urn:namespace:type:name:41414141:dummy.model:1" + example_urn = f'{{"urn": "{urn_string}"}}' + + class Wrapper(BaseModel): + """Need to wrap as plain string is not valid json.""" + + urn: URN + + wrapper = Wrapper.parse_raw(example_urn) + urn = wrapper.urn + assert urn.namespace == "namespace" + assert urn.type == "type" + assert urn.name == "name" + assert urn.internal_id == "41414141" + assert urn.model == "dummy.model" + assert urn.version == 1 + + # Check that the serialization works, too + assert repr(urn) == urn_string + + +def test_service(): + data = """ + { + "iid": 1, + "description": "test service", + "type": "urn:miot-spec-v2:service:device-information:00000001:dummy:1" + } + """ + serv = MiotService.parse_raw(data) + assert serv.siid == 1 + assert serv.urn.type == "service" + assert serv.actions == [] + assert serv.properties == [] + assert serv.events == [] + + +def test_event(): + data = '{"iid": 1, "type": "urn:spect:event:example_event:00000001:dummymodel:1", "description": "dummy", "arguments": []}' + ev = MiotEvent.parse_raw(data) + assert ev.eiid == 1 + assert ev.urn.type == "event" + assert ev.description == "dummy" + assert ev.arguments == [] + + +def test_property(): + data = """ + { + "iid": 1, + "type": "urn:miot-spec-v2:property:manufacturer:00000001:dummy:1", + "description": "Device Manufacturer", + "format": "string", + "access": [ + "read" + ] + } + """ + prop: MiotProperty = MiotProperty.parse_raw(data) + assert prop.piid == 1 + assert prop.urn.type == "property" + assert prop.format == str + assert prop.access == ["read"] + assert prop.description == "Device Manufacturer" + + assert prop.plain_name == "manufacturer" + + +@pytest.mark.xfail(reason="not implemented") +def test_property_pretty_value(): + """Test the pretty value conversions.""" + raise NotImplementedError()