feat: update std_lib and multi_lang logic

custom_components/xiaomi_home/miot/miot_spec.py

@@ -59,8 +59,7 @@ from .miot_error import MIoTSpecError
 from .miot_storage import (
     MIoTStorage,
     SpecBoolTranslation,
-    SpecFilter,
+    SpecFilter)
-    SpecMultiLang)
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -693,21 +692,16 @@ class MIoTSpecInstance:
         }
 
 
-class MIoTSpecParser:
+class _MIoTSpecMultiLang:
-    """MIoT SPEC parser."""
+    """MIoT SPEC multi lang class."""
-    # pylint: disable=inconsistent-quotes
+    # pylint: disable=broad-exception-caught
-    VERSION: int = 1
+    _DOMAIN: str = 'miot_specs_multi_lang'
-    DOMAIN: str = 'miot_specs'
     _lang: str
     _storage: MIoTStorage
     _main_loop: asyncio.AbstractEventLoop
 
-    _init_done: bool
+    _custom_cache: dict[str, dict]
-
+    _current_data: Optional[dict[str, str]]
-    _std_lib: _SpecStdLib
-    _bool_trans: SpecBoolTranslation
-    _multi_lang: SpecMultiLang
-    _spec_filter: SpecFilter
 
     def __init__(
         self, lang: Optional[str],
@@ -718,22 +712,137 @@ class MIoTSpecParser:
         self._storage = storage
         self._main_loop = loop or asyncio.get_running_loop()
 
+        self._custom_cache = {}
+        self._current_data = None
+
+    async def set_spec_async(self, urn: str) -> None:
+        if urn in self._custom_cache:
+            self._current_data = self._custom_cache[urn]
+            return
+
+        trans_cache: dict[str, str] = {}
+        trans_cloud: dict = {}
+        trans_local: dict = {}
+        # Get multi lang from cloud
+        try:
+            trans_cloud = await self.__get_multi_lang_async(urn)
+            if self._lang == 'zh-Hans':
+                # Simplified Chinese
+                trans_cache = trans_cloud.get('zh_cn', {})
+            elif self._lang == 'zh-Hant':
+                # Traditional Chinese, zh_hk or zh_tw
+                trans_cache = trans_cloud.get('zh_hk', {})
+                if not trans_cache:
+                    trans_cache = trans_cloud.get('zh_tw', {})
+            else:
+                trans_cache = trans_cloud.get(self._lang, {})
+        except Exception as err:
+            trans_cloud = {}
+            _LOGGER.info('get multi lang from cloud failed, %s, %s', urn, err)
+        # Get multi lang from local
+        try:
+            trans_local = await self._storage.load_async(
+                domain=self._DOMAIN, name=urn, type_=dict)  # type: ignore
+            if (
+                isinstance(trans_local, dict)
+                and self._lang in trans_local
+            ):
+                trans_cache.update(trans_local[self._lang])
+        except Exception as err:
+            trans_local = {}
+            _LOGGER.info('get multi lang from local failed, %s, %s', urn, err)
+        # Default language
+        if not trans_cache:
+            if DEFAULT_INTEGRATION_LANGUAGE in trans_cloud:
+                trans_cache = trans_cloud[DEFAULT_INTEGRATION_LANGUAGE]
+            if DEFAULT_INTEGRATION_LANGUAGE in trans_local:
+                trans_cache.update(
+                    trans_local[DEFAULT_INTEGRATION_LANGUAGE])
+        trans_data: dict[str, str] = {}
+        for tag, value in trans_cache.items():
+            if value is None or value.strip() == '':
+                continue
+            # The dict key is like:
+            # 'service:002:property:001:valuelist:000' or
+            # 'service:002:property:001' or 'service:002'
+            strs: list = tag.split(':')
+            strs_len = len(strs)
+            if strs_len == 2:
+                trans_data[f's:{int(strs[1])}'] = value
+            elif strs_len == 4:
+                type_ = 'p' if strs[2] == 'property' else (
+                    'a' if strs[2] == 'action' else 'e')
+                trans_data[
+                    f'{type_}:{int(strs[1])}:{int(strs[3])}'
+                ] = value
+            elif strs_len == 6:
+                trans_data[
+                    f'v:{int(strs[1])}:{int(strs[3])}:{int(strs[5])}'
+                ] = value
+
+        self._custom_cache[urn] = trans_data
+        self._current_data = trans_data
+
+    def translate(self, key: str) -> Optional[str]:
+        if not self._current_data:
+            return None
+        return self._current_data.get(key, None)
+
+    async def __get_multi_lang_async(self, urn: str) -> dict:
+        res_trans = await MIoTHttp.get_json_async(
+            url='https://miot-spec.org/instance/v2/multiLanguage',
+            params={'urn': urn})
+        if (
+            not isinstance(res_trans, dict)
+            or 'data' not in res_trans
+            or not isinstance(res_trans['data'], dict)
+        ):
+            raise MIoTSpecError('invalid translation data')
+        return res_trans['data']
+
+
+class MIoTSpecParser:
+    """MIoT SPEC parser."""
+    # pylint: disable=inconsistent-quotes
+    VERSION: int = 1
+    _DOMAIN: str = 'miot_specs'
+    _lang: str
+    _storage: MIoTStorage
+    _main_loop: asyncio.AbstractEventLoop
+
+    _std_lib: _SpecStdLib
+    _multi_lang: _MIoTSpecMultiLang
+
+    _init_done: bool
+
+    _bool_trans: SpecBoolTranslation
+    _spec_filter: SpecFilter
+
+    def __init__(
+        self, lang: Optional[str],
+        storage: MIoTStorage,
+        loop: Optional[asyncio.AbstractEventLoop] = None
+    ) -> None:
+        self._lang = lang or DEFAULT_INTEGRATION_LANGUAGE
+        self._storage = storage
+        self._main_loop = loop or asyncio.get_running_loop()
+        self._std_lib = _SpecStdLib(lang=self._lang)
+        self._multi_lang = _MIoTSpecMultiLang(
+            lang=self._lang, storage=self._storage, loop=self._main_loop)
+
         self._init_done = False
 
-        self._std_lib = _SpecStdLib(lang=self._lang)
         self._bool_trans = SpecBoolTranslation(
             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)
 
     async def init_async(self) -> None:
         if self._init_done is True:
             return
         await self._bool_trans.init_async()
-        await self._multi_lang.init_async()
         await self._spec_filter.init_async()
         std_lib_cache = await self._storage.load_async(
-            domain=self.DOMAIN, name='spec_std_lib', type_=dict)
+            domain=self._DOMAIN, name='spec_std_lib', type_=dict)
         if (
             isinstance(std_lib_cache, dict)
             and 'data' in std_lib_cache
@@ -751,7 +860,7 @@ class MIoTSpecParser:
         # Update spec std lib
         if await self._std_lib.refresh_async():
             if not await self._storage.save_async(
-                domain=self.DOMAIN, name='spec_std_lib',
+                domain=self._DOMAIN, name='spec_std_lib',
                 data={
                     'data': self._std_lib.dump(),
                     'ts': int(time.time())
@@ -770,7 +879,6 @@ class MIoTSpecParser:
         self._init_done = False
         # self._std_lib.deinit()
         await self._bool_trans.deinit_async()
-        await self._multi_lang.deinit_async()
         await self._spec_filter.deinit_async()
 
     async def parse(
@@ -797,7 +905,7 @@ class MIoTSpecParser:
             return False
         if await self._std_lib.refresh_async():
             if not await self._storage.save_async(
-                domain=self.DOMAIN, name='spec_std_lib',
+                domain=self._DOMAIN, name='spec_std_lib',
                 data={
                     'data': self._std_lib.dump(),
                     'ts': int(time.time())
@@ -819,7 +927,7 @@ class MIoTSpecParser:
         if platform.system() == 'Windows':
             urn = urn.replace(':', '_')
         return await self._storage.load_async(
-            domain=self.DOMAIN,
+            domain=self._DOMAIN,
             name=f'{urn}_{self._lang}',
             type_=dict)  # type: ignore
 
@@ -827,7 +935,7 @@ class MIoTSpecParser:
         if platform.system() == 'Windows':
             urn = urn.replace(':', '_')
         return await self._storage.save_async(
-            domain=self.DOMAIN, name=f'{urn}_{self._lang}', data=data)
+            domain=self._DOMAIN, name=f'{urn}_{self._lang}', data=data)
 
     def __spec_format2dtype(self, format_: str) -> str:
         # 'string'|'bool'|'uint8'|'uint16'|'uint32'|
@@ -840,11 +948,6 @@ class MIoTSpecParser:
             url='https://miot-spec.org/miot-spec-v2/instance',
             params={'type': urn})
 
-    async def __get_translation(self, urn: str) -> Optional[dict]:
-        return await MIoTHttp.get_json_async(
-            url='https://miot-spec.org/instance/v2/multiLanguage',
-            params={'urn': urn})
-
     async def __parse(self, urn: str) -> MIoTSpecInstance:
         _LOGGER.debug('parse urn, %s', urn)
         # Load spec instance
@@ -856,68 +959,11 @@ class MIoTSpecParser:
             or 'services' not in instance
         ):
             raise MIoTSpecError(f'invalid urn instance, {urn}')
-        translation: dict = {}
         urn_strs: list[str] = urn.split(':')
         urn_key: str = ':'.join(urn_strs[:6])
-        try:
+        # Set translation cache
-            # Load multiple language configuration.
+        await self._multi_lang.set_spec_async(urn=urn_key)
-            res_trans = await self.__get_translation(urn=urn)
+        # Set spec filter
-            if (
-                not isinstance(res_trans, dict)
-                or 'data' not in res_trans
-                or not isinstance(res_trans['data'], dict)
-            ):
-                raise MIoTSpecError('invalid translation data')
-            trans_data: dict[str, str] = {}
-            if self._lang == 'zh-Hans':
-                # Simplified Chinese
-                trans_data = res_trans['data'].get('zh_cn', {})
-            elif self._lang == 'zh-Hant':
-                # Traditional Chinese, zh_hk or zh_tw
-                trans_data = res_trans['data'].get('zh_hk', {})
-                if not trans_data:
-                    trans_data = res_trans['data'].get('zh_tw', {})
-            else:
-                trans_data = res_trans['data'].get(self._lang, {})
-            # Load local multiple language configuration.
-            multi_lang: dict = await self._multi_lang.translate_async(
-                urn_key=urn_key)
-            if multi_lang:
-                trans_data.update(multi_lang)
-            if not trans_data:
-                trans_data = res_trans['data'].get(
-                    DEFAULT_INTEGRATION_LANGUAGE, {})
-                if not trans_data:
-                    raise MIoTSpecError(
-                        f'the language is not supported, {self._lang}')
-                else:
-                    _LOGGER.error(
-                        'the language is not supported, %s, try using the '
-                        'default language, %s, %s',
-                        self._lang, DEFAULT_INTEGRATION_LANGUAGE, urn)
-            for tag, value in trans_data.items():
-                if value is None or value.strip() == '':
-                    continue
-                # The dict key is like:
-                # 'service:002:property:001:valuelist:000' or
-                # 'service:002:property:001' or 'service:002'
-                strs: list = tag.split(':')
-                strs_len = len(strs)
-                if strs_len == 2:
-                    translation[f's:{int(strs[1])}'] = value
-                elif strs_len == 4:
-                    type_ = 'p' if strs[2] == 'property' else (
-                        'a' if strs[2] == 'action' else 'e')
-                    translation[
-                        f'{type_}:{int(strs[1])}:{int(strs[3])}'
-                    ] = value
-                elif strs_len == 6:
-                    translation[
-                        f'v:{int(strs[1])}:{int(strs[3])}:{int(strs[5])}'
-                    ] = value
-        except MIoTSpecError as e:
-            _LOGGER.error('get translation error, %s, %s', urn, e)
-        # Spec filter
         self._spec_filter.filter_spec(urn_key=urn_key)
         # Parse device type
         spec_instance: MIoTSpecInstance = MIoTSpecInstance(
@@ -948,7 +994,7 @@ class MIoTSpecParser:
             if type_strs[1] != 'miot-spec-v2':
                 spec_service.proprietary = True
             spec_service.description_trans = (
-                translation.get(f's:{service["iid"]}', None)
+                self._multi_lang.translate(f's:{service["iid"]}')
                 or self._std_lib.service_translate(key=':'.join(type_strs[:5]))
                 or service['description']
                 or spec_service.name
@@ -979,8 +1025,8 @@ class MIoTSpecParser:
                 if p_type_strs[1] != 'miot-spec-v2':
                     spec_prop.proprietary = spec_service.proprietary or True
                 spec_prop.description_trans = (
-                    translation.get(
+                    self._multi_lang.translate(
-                        f'p:{service["iid"]}:{property_["iid"]}', None)
+                        f'p:{service["iid"]}:{property_["iid"]}')
                     or self._std_lib.property_translate(
                         key=':'.join(p_type_strs[:5]))
                     or property_['description']
@@ -1000,9 +1046,9 @@ class MIoTSpecParser:
                     for index, v in enumerate(v_list):
                         v['name'] = v['description']
                         v['description'] = (
-                            translation.get(
+                            self._multi_lang.translate(
                                 f'v:{service["iid"]}:{property_["iid"]}:'
-                                f'{index}', None)
+                                f'{index}')
                             or self._std_lib.value_translate(
                                 key=f'{type_strs[:5]}|{p_type_strs[3]}|'
                                 f'{v["description"]}')
@@ -1037,8 +1083,8 @@ class MIoTSpecParser:
                 if e_type_strs[1] != 'miot-spec-v2':
                     spec_event.proprietary = spec_service.proprietary or True
                 spec_event.description_trans = (
-                    translation.get(
+                    self._multi_lang.translate(
-                        f'e:{service["iid"]}:{event["iid"]}', None)
+                        f'e:{service["iid"]}:{event["iid"]}')
                     or self._std_lib.event_translate(
                         key=':'.join(e_type_strs[:5]))
                     or event['description']
@@ -1073,8 +1119,8 @@ class MIoTSpecParser:
                 if a_type_strs[1] != 'miot-spec-v2':
                     spec_action.proprietary = spec_service.proprietary or True
                 spec_action.description_trans = (
-                    translation.get(
+                    self._multi_lang.translate(
-                        f'a:{service["iid"]}:{action["iid"]}', None)
+                        f'a:{service["iid"]}:{action["iid"]}')
                     or self._std_lib.action_translate(
                         key=':'.join(a_type_strs[:5]))
                     or action['description']

custom_components/xiaomi_home/miot/miot_storage.py

@@ -719,60 +719,6 @@ class MIoTCert:
         return binascii.hexlify(sha1_hash.finalize()).decode('utf-8')
 
 
-class SpecMultiLang:
-    """
-    MIoT-Spec-V2 multi-language for entities.
-    """
-    MULTI_LANG_FILE = 'specs/multi_lang.json'
-    _main_loop: asyncio.AbstractEventLoop
-    _lang: str
-    _data: Optional[dict[str, dict]]
-
-    def __init__(
-        self, lang: str, loop: Optional[asyncio.AbstractEventLoop] = None
-    ) -> None:
-        self._main_loop = loop or asyncio.get_event_loop()
-        self._lang = lang
-        self._data = None
-
-    async def init_async(self) -> None:
-        if isinstance(self._data, dict):
-            return
-        multi_lang_data = None
-        self._data = {}
-        try:
-            multi_lang_data = await self._main_loop.run_in_executor(
-                None, load_json_file,
-                os.path.join(
-                    os.path.dirname(os.path.abspath(__file__)),
-                    self.MULTI_LANG_FILE))
-        except Exception as err:  # pylint: disable=broad-exception-caught
-            _LOGGER.error('multi lang, load file error, %s', err)
-            return
-        # Check if the file is a valid JSON file
-        if not isinstance(multi_lang_data, dict):
-            _LOGGER.error('multi lang, invalid file data')
-            return
-        for lang_data in multi_lang_data.values():
-            if not isinstance(lang_data, dict):
-                _LOGGER.error('multi lang, invalid lang data')
-                return
-            for data in lang_data.values():
-                if not isinstance(data, dict):
-                    _LOGGER.error('multi lang, invalid lang data item')
-                    return
-        self._data = multi_lang_data
-
-    async def deinit_async(self) -> str:
-        self._data = None
-
-    async def translate_async(self, urn_key: str) -> dict[str, str]:
-        """MUST call init_async() first."""
-        if urn_key in self._data:
-            return self._data[urn_key].get(self._lang, {})
-        return {}
-
-
 class SpecBoolTranslation:
     """
     Boolean value translation.