feat: custom service spec

custom_components/xiaomi_home/__init__.py

@@ -155,7 +155,8 @@ async def async_setup_entry(
                     for entity in filter_entities:
                         device.entity_list[platform].remove(entity)
                         entity_id = device.gen_service_entity_id(
-                            ha_domain=platform, siid=entity.spec.iid)
+                            ha_domain=platform, siid=entity.spec.iid,
+                            description=entity.spec.description)
                         if er.async_get(entity_id_or_uuid=entity_id):
                             er.async_remove(entity_id=entity_id)
                 if platform in device.prop_list:

custom_components/xiaomi_home/miot/miot_device.py

@@ -298,10 +298,11 @@ class MIoTDevice:
             f'{ha_domain}.{self._model_strs[0][:9]}_{self.did_tag}_'
             f'{self._model_strs[-1][:20]}')
 
-    def gen_service_entity_id(self, ha_domain: str, siid: int) -> str:
+    def gen_service_entity_id(self, ha_domain: str, siid: int,
+                              description: str) -> str:
         return (
             f'{ha_domain}.{self._model_strs[0][:9]}_{self.did_tag}_'
-            f'{self._model_strs[-1][:20]}_s_{siid}')
+            f'{self._model_strs[-1][:20]}_s_{siid}_{description}')
 
     def gen_prop_entity_id(
         self, ha_domain: str, spec_name: str, siid: int, piid: int
@@ -731,7 +732,8 @@ class MIoTServiceEntity(Entity):
             self._attr_name = f' {self.entity_data.spec.description_trans}'
         elif isinstance(entity_data.spec, MIoTSpecService):
             self.entity_id = miot_device.gen_service_entity_id(
-                DOMAIN, siid=entity_data.spec.iid)
+                DOMAIN, siid=entity_data.spec.iid,
+                description=entity_data.spec.description)
             self._attr_name = (
                 f'{"* "if self.entity_data.spec.proprietary else " "}'
                 f'{self.entity_data.spec.description_trans}')

custom_components/xiaomi_home/miot/miot_spec.py

@@ -61,6 +61,7 @@ from .miot_storage import (
     MIoTStorage,
     SpecBoolTranslation,
     SpecFilter,
+    SpecCustomService,
     SpecMultiLang)
 
 _LOGGER = logging.getLogger(__name__)
@@ -466,6 +467,7 @@ class MIoTSpecParser:
     _bool_trans: SpecBoolTranslation
     _multi_lang: SpecMultiLang
     _spec_filter: SpecFilter
+    _custom_service: SpecCustomService
 
     def __init__(
         self, lang: str = DEFAULT_INTEGRATION_LANGUAGE,
@@ -484,6 +486,7 @@ class MIoTSpecParser:
             lang=self._lang, loop=self._main_loop)
         self._multi_lang = SpecMultiLang(lang=self._lang, loop=self._main_loop)
         self._spec_filter = SpecFilter(loop=self._main_loop)
+        self._custom_service = SpecCustomService(loop=self._main_loop)
 
     async def init_async(self) -> None:
         if self._init_done is True:
@@ -491,6 +494,7 @@ class MIoTSpecParser:
         await self._bool_trans.init_async()
         await self._multi_lang.init_async()
         await self._spec_filter.init_async()
+        await self._custom_service.init_async()
         std_lib_cache: dict = None
         if self._storage:
             std_lib_cache: dict = await self._storage.load_async(
@@ -536,6 +540,7 @@ class MIoTSpecParser:
         await self._bool_trans.deinit_async()
         await self._multi_lang.deinit_async()
         await self._spec_filter.deinit_async()
+        await self._custom_service.deinit_async()
         self._ram_cache.clear()
 
     async def parse(
@@ -779,6 +784,8 @@ class MIoTSpecParser:
         _LOGGER.debug('parse urn, %s', urn)
         # Load spec instance
         instance: dict = await self.__get_instance(urn=urn)
+        # Modify the spec instance by custom spec
+        instance = self._custom_service.modify_spec(urn=urn, spec=instance)
         if (
             not isinstance(instance, dict)
             or 'type' not in instance

custom_components/xiaomi_home/miot/miot_storage.py

@@ -1033,3 +1033,63 @@ class DeviceManufacturer:
         except Exception as err:  # pylint: disable=broad-exception-caught
             _LOGGER.error('get manufacturer info failed, %s', err)
         return None
+
+
+class SpecCustomService:
+    """Custom MIoT-Spec-V2 service defined by the user."""
+    CUSTOM_SPEC_FILE = 'specs/custom_service.json'
+    _main_loop: asyncio.AbstractEventLoop
+    _data: dict[str, dict[str, any]]
+
+    def __init__(self, loop: Optional[asyncio.AbstractEventLoop]) -> None:
+        self._main_loop = loop or asyncio.get_event_loop()
+        self._data = None
+
+    async def init_async(self) -> None:
+        if isinstance(self._data, dict):
+            return
+        custom_data = None
+        self._data = {}
+        try:
+            custom_data = await self._main_loop.run_in_executor(
+                None, load_json_file,
+                os.path.join(
+                    os.path.dirname(os.path.abspath(__file__)),
+                    self.CUSTOM_SPEC_FILE))
+        except Exception as err:  # pylint: disable=broad-exception-caught
+            _LOGGER.error('custom service, load file error, %s', err)
+            return
+        if not isinstance(custom_data, dict):
+            _LOGGER.error('custom service, invalid spec content')
+            return
+        for values in list(custom_data.values()):
+            if not isinstance(values, dict):
+                _LOGGER.error('custom service, invalid spec data')
+                return
+        self._data = custom_data
+
+    async def deinit_async(self) -> None:
+        self._data = None
+
+    def modify_spec(self, urn: str, spec: dict) -> dict | None:
+        """MUST call init_async() first."""
+        if not self._data:
+            _LOGGER.error('self._data is None')
+            return spec
+        if urn not in self._data:
+            return spec
+        if 'services' not in spec:
+            return spec
+        spec_services = spec['services']
+        custom_spec = self._data.get(urn, None)
+        # Replace services by custom defined spec
+        for i, service in enumerate(spec_services):
+            siid = str(service['iid'])
+            if siid in custom_spec:
+                spec_services[i] = custom_spec[siid]
+        # Add new services
+        if 'new' in custom_spec:
+            for service in custom_spec['new']:
+                spec_services.append(service)
+
+        return spec

custom_components/xiaomi_home/miot/specs/custom_service.json

@@ -0,0 +1,59 @@
+{
+    "urn:miot-spec-v2:device:airer:0000A00D:hyd-lyjpro:1": {
+        "3": {
+            "iid": 3,
+            "type": "urn:miot-spec-v2:service:light:00007802:hyd-lyjpro:1",
+            "description": "Light",
+            "properties": [
+                {
+                    "iid": 1,
+                    "type": "urn:miot-spec-v2:property:on:00000006:hyd-lyjpro:1",
+                    "description": "Sunlight",
+                    "format": "bool",
+                    "access": [
+                        "read",
+                        "write",
+                        "notify"
+                    ]
+                },
+                {
+                    "iid": 3,
+                    "type": "urn:miot-spec-v2:property:flex-switch:000000EC:hyd-lyjpro:1",
+                    "description": "Flex Switch",
+                    "format": "uint8",
+                    "access": [
+                        "read",
+                        "write",
+                        "notify"
+                    ],
+                    "value-list": [
+                        {
+                            "value": 1,
+                            "description": "Overturn"
+                        }
+                    ]
+                }
+            ]
+        },
+        "new": [
+            {
+                "iid": 3,
+                "type": "urn:miot-spec-v2:service:light:00007802:hyd-lyjpro:1",
+                "description": "Moonlight",
+                "properties": [
+                    {
+                        "iid": 2,
+                        "type": "urn:miot-spec-v2:property:on:00000006:hyd-lyjpro:1",
+                        "description": "Switch Status",
+                        "format": "bool",
+                        "access": [
+                            "read",
+                            "write",
+                            "notify"
+                        ]
+                    }
+                ]
+            }
+        ]
+    }
+}