chickliu/ha_xiaomi_home
1
0
Fork 0
mirror of https://github.com/XiaoMi/ha_xiaomi_home.git synced 2026-01-17 23:50:42 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: remove spec cache storage

Browse Source
This commit is contained in:
topsworld 2025-01-08 10:39:07 +08:00
parent d9d8433405
commit e4dfdf68ab
1 changed files with 92 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

141
custom_components/xiaomi_home/miot/miot_spec.py
View File

@ -46,12 +46,9 @@ off Xiaomi or its affiliates' products.
MIoT-Spec-V2 parser. MIoT-Spec-V2 parser.
""" """
import asyncio import asyncio
import json
import platform import platform
import time import time
from typing import Any, Optional from typing import Any, Optional, Union
from urllib.parse import urlencode
from urllib.request import Request, urlopen
import logging import logging
@ -74,13 +71,36 @@ class _MIoTSpecValueRange:
max_: int max_: int
step: int step: int
def from_list(self, value_range: list) -> None: def __init__(self, value_range: Union[dict, list, None]) -> None:
if isinstance(value_range, dict):
self.load(value_range)
elif isinstance(value_range, list):
self.from_spec(value_range)
def load(self, value_range: dict) -> None:
if (
'min' not in value_range
or 'max' not in value_range
or 'step' not in value_range
):
raise MIoTSpecError('invalid value range')
self.min_ = value_range['min']
self.max_ = value_range['max']
self.step = value_range['step']
def from_spec(self, value_range: list) -> None:
if len(value_range) != 3:
raise MIoTSpecError('invalid value range')
self.min_ = value_range[0] self.min_ = value_range[0]
self.max_ = value_range[1] self.max_ = value_range[1]
self.step = value_range[2] self.step = value_range[2]
def to_list(self) -> list: def dump(self) -> dict:
return [self.min_, self.max_, self.step] return {
'min': self.min_,
'max': self.max_,
'step': self.step
}
class _MIoTSpecValueListItem: class _MIoTSpecValueListItem:
@ -92,7 +112,7 @@ class _MIoTSpecValueListItem:
# Descriptions after multilingual conversion. # Descriptions after multilingual conversion.
description: str description: str
def to_dict(self) -> dict: def dump(self) -> dict:
return { return {
'name': self.name, 'name': self.name,
'value': self.value, 'value': self.value,
@ -107,8 +127,8 @@ class _MIoTSpecValueList:
def to_map(self) -> dict: def to_map(self) -> dict:
return {item.value: item.description for item in self.items} return {item.value: item.description for item in self.items}
def to_list(self) -> list: def dump(self) -> list:
return [item.to_dict() for item in self.items] return [item.dump() for item in self.items]
class _SpecStdLib: class _SpecStdLib:
@ -132,7 +152,7 @@ class _SpecStdLib:
self._spec_std_lib = None self._spec_std_lib = None
def from_dict(self, std_lib: dict[str, dict[str, dict[str, str]]]) -> None: def load(self, std_lib: dict[str, dict[str, dict[str, str]]]) -> None:
if ( if (
not isinstance(std_lib, dict) not isinstance(std_lib, dict)
or 'devices' not in std_lib or 'devices' not in std_lib
@ -211,7 +231,7 @@ class _SpecStdLib:
async def refresh_async(self) -> bool: async def refresh_async(self) -> bool:
std_lib_new = await self.__request_from_cloud_async() std_lib_new = await self.__request_from_cloud_async()
if std_lib_new: if std_lib_new:
self.from_dict(std_lib_new) self.load(std_lib_new)
return True return True
return False return False
@ -353,7 +373,7 @@ class _SpecStdLib:
return result return result
class MIoTSpecBase: class _MIoTSpecBase:
"""MIoT SPEC base class.""" """MIoT SPEC base class."""
iid: int iid: int
type_: str type_: str
@ -397,13 +417,13 @@ class MIoTSpecBase:
return self.spec_id == value.spec_id return self.spec_id == value.spec_id
class MIoTSpecProperty(MIoTSpecBase): class MIoTSpecProperty(_MIoTSpecBase):
"""MIoT SPEC property class.""" """MIoT SPEC property class."""
format_: str format_: str
precision: int precision: int
unit: Optional[str] unit: Optional[str]
value_range: Optional[list] _value_range: Optional[_MIoTSpecValueRange]
value_list: Optional[list[dict]] value_list: Optional[list[dict]]
_access: list _access: list
@ -411,12 +431,18 @@ class MIoTSpecProperty(MIoTSpecBase):
_readable: bool _readable: bool
_notifiable: bool _notifiable: bool
service: MIoTSpecBase service: 'MIoTSpecService'
def __init__( def __init__(
self, spec: dict, service: MIoTSpecBase, format_: str, access: list, self,
unit: Optional[str] = None, value_range: Optional[list] = None, spec: dict,
value_list: Optional[list[dict]] = None, precision: int = 0 service: 'MIoTSpecService',
format_: str,
access: list,
unit: Optional[str] = None,
value_range: Optional[dict] = None,
value_list: Optional[list[dict]] = None,
precision: int = 0
) -> None: ) -> None:
super().__init__(spec=spec) super().__init__(spec=spec)
self.service = service self.service = service
@ -454,6 +480,14 @@ class MIoTSpecProperty(MIoTSpecBase):
def notifiable(self): def notifiable(self):
return self._notifiable return self._notifiable
@property
def value_range(self) -> Optional[_MIoTSpecValueRange]:
return self._value_range
@value_range.setter
def value_range(self, value: Union[dict, list, None]) -> None:
self._value_range = _MIoTSpecValueRange(value_range=value)
def value_format(self, value: Any) -> Any: def value_format(self, value: Any) -> Any:
if value is None: if value is None:
return None return None
@ -477,23 +511,24 @@ class MIoTSpecProperty(MIoTSpecBase):
'format': self.format_, 'format': self.format_,
'access': self._access, 'access': self._access,
'unit': self.unit, 'unit': self.unit,
'value_range': self.value_range, 'value_range': (
self.value_range.dump() if self.value_range else None),
'value_list': self.value_list, 'value_list': self.value_list,
'precision': self.precision 'precision': self.precision
} }
class MIoTSpecEvent(MIoTSpecBase): class MIoTSpecEvent(_MIoTSpecBase):
"""MIoT SPEC event class.""" """MIoT SPEC event class."""
argument: list[MIoTSpecProperty] argument: list[MIoTSpecProperty]
service: MIoTSpecBase service: 'MIoTSpecService'
def __init__( def __init__(
self, spec: dict, service: MIoTSpecBase, self, spec: dict, service: 'MIoTSpecService',
argument: list[MIoTSpecProperty] = None argument: Optional[list[MIoTSpecProperty]] = None
) -> None: ) -> None:
super().__init__(spec=spec) super().__init__(spec=spec)
self.argument = argument self.argument = argument or []
self.service = service self.service = service
self.spec_id = hash( self.spec_id = hash(
@ -512,20 +547,20 @@ class MIoTSpecEvent(MIoTSpecBase):
} }
class MIoTSpecAction(MIoTSpecBase): class MIoTSpecAction(_MIoTSpecBase):
"""MIoT SPEC action class.""" """MIoT SPEC action class."""
in_: list[MIoTSpecProperty] in_: list[MIoTSpecProperty]
out: list[MIoTSpecProperty] out: list[MIoTSpecProperty]
service: MIoTSpecBase service: 'MIoTSpecService'
def __init__( def __init__(
self, spec: dict, service: MIoTSpecBase = None, self, spec: dict, service: 'MIoTSpecService',
in_: list[MIoTSpecProperty] = None, in_: Optional[list[MIoTSpecProperty]] = None,
out: list[MIoTSpecProperty] = None out: Optional[list[MIoTSpecProperty]] = None
) -> None: ) -> None:
super().__init__(spec=spec) super().__init__(spec=spec)
self.in_ = in_ self.in_ = in_ or []
self.out = out self.out = out or []
self.service = service self.service = service
self.spec_id = hash( self.spec_id = hash(
@ -545,7 +580,7 @@ class MIoTSpecAction(MIoTSpecBase):
} }
class MIoTSpecService(MIoTSpecBase): class MIoTSpecService(_MIoTSpecBase):
"""MIoT SPEC service class.""" """MIoT SPEC service class."""
properties: list[MIoTSpecProperty] properties: list[MIoTSpecProperty]
events: list[MIoTSpecEvent] events: list[MIoTSpecEvent]
@ -588,8 +623,7 @@ class MIoTSpecInstance:
icon: str icon: str
def __init__( def __init__(
self, urn: str = None, name: str = None, self, urn: str, name: str, description: str, description_trans: str
description: str = None, description_trans: str = None
) -> None: ) -> None:
self.urn = urn self.urn = urn
self.name = name self.name = name
@ -597,12 +631,13 @@ class MIoTSpecInstance:
self.description_trans = description_trans self.description_trans = description_trans
self.services = [] self.services = []
def load(self, specs: dict) -> 'MIoTSpecInstance': @staticmethod
self.urn = specs['urn'] def load(specs: dict) -> 'MIoTSpecInstance':
self.name = specs['name'] instance = MIoTSpecInstance(
self.description = specs['description'] urn=specs['urn'],
self.description_trans = specs['description_trans'] name=specs['name'],
self.services = [] description=specs['description'],
description_trans=specs['description_trans'])
for service in specs['services']: for service in specs['services']:
spec_service = MIoTSpecService(spec=service) spec_service = MIoTSpecService(spec=service)
for prop in service['properties']: for prop in service['properties']:
@ -645,8 +680,8 @@ class MIoTSpecInstance:
break break
spec_action.out = out_list spec_action.out = out_list
spec_service.actions.append(spec_action) spec_service.actions.append(spec_action)
self.services.append(spec_service) instance.services.append(spec_service)
return self return instance
def dump(self) -> dict: def dump(self) -> dict:
return { return {
@ -668,7 +703,6 @@ class MIoTSpecParser:
_main_loop: asyncio.AbstractEventLoop _main_loop: asyncio.AbstractEventLoop
_init_done: bool _init_done: bool
_ram_cache: dict
_std_lib: _SpecStdLib _std_lib: _SpecStdLib
_bool_trans: SpecBoolTranslation _bool_trans: SpecBoolTranslation
@ -685,7 +719,6 @@ class MIoTSpecParser:
self._main_loop = loop or asyncio.get_running_loop() self._main_loop = loop or asyncio.get_running_loop()
self._init_done = False self._init_done = False
self._ram_cache = {}
self._std_lib = _SpecStdLib(lang=self._lang) self._std_lib = _SpecStdLib(lang=self._lang)
self._bool_trans = SpecBoolTranslation( self._bool_trans = SpecBoolTranslation(
@ -712,7 +745,7 @@ class MIoTSpecParser:
# Use the cache if the update time is less than 14 day # Use the cache if the update time is less than 14 day
_LOGGER.debug( _LOGGER.debug(
'use local spec std cache, ts->%s', std_lib_cache['ts']) 'use local spec std cache, ts->%s', std_lib_cache['ts'])
self._std_lib.from_dict(std_lib_cache['data']) self._std_lib.load(std_lib_cache['data'])
self._init_done = True self._init_done = True
return return
# Update spec std lib # Update spec std lib
@ -727,7 +760,7 @@ class MIoTSpecParser:
_LOGGER.error('save spec std lib failed') _LOGGER.error('save spec std lib failed')
else: else:
if isinstance(std_lib_cache, dict) and 'data' in std_lib_cache: if isinstance(std_lib_cache, dict) and 'data' in std_lib_cache:
self._std_lib.from_dict(std_lib_cache['data']) self._std_lib.load(std_lib_cache['data'])
_LOGGER.info('get spec std lib failed, use local cache') _LOGGER.info('get spec std lib failed, use local cache')
else: else:
_LOGGER.error('load spec std lib failed') _LOGGER.error('load spec std lib failed')
@ -739,17 +772,16 @@ class MIoTSpecParser:
await self._bool_trans.deinit_async() await self._bool_trans.deinit_async()
await self._multi_lang.deinit_async() await self._multi_lang.deinit_async()
await self._spec_filter.deinit_async() await self._spec_filter.deinit_async()
self._ram_cache.clear()
async def parse( async def parse(
self, urn: str, skip_cache: bool = False, self, urn: str, skip_cache: bool = False,
) -> MIoTSpecInstance: ) -> Optional[MIoTSpecInstance]:
"""MUST await init first !!!""" """MUST await init first !!!"""
if not skip_cache: if not skip_cache:
cache_result = await self.__cache_get(urn=urn) cache_result = await self.__cache_get(urn=urn)
if isinstance(cache_result, dict): if isinstance(cache_result, dict):
_LOGGER.debug('get from cache, %s', urn) _LOGGER.debug('get from cache, %s', urn)
return MIoTSpecInstance().load(specs=cache_result) return MIoTSpecInstance.load(specs=cache_result)
# Retry three times # Retry three times
for index in range(3): for index in range(3):
try: try:
@ -784,21 +816,18 @@ class MIoTSpecParser:
return success_count return success_count
async def __cache_get(self, urn: str) -> Optional[dict]: async def __cache_get(self, urn: str) -> Optional[dict]:
if self._storage is not None:
if platform.system() == 'Windows': if platform.system() == 'Windows':
urn = urn.replace(':', '_') urn = urn.replace(':', '_')
return await self._storage.load_async( return await self._storage.load_async(
domain=self.DOMAIN, name=f'{urn}_{self._lang}', type_=dict) domain=self.DOMAIN,
return self._ram_cache.get(urn, None) name=f'{urn}_{self._lang}',
type_=dict) # type: ignore
async def __cache_set(self, urn: str, data: dict) -> bool: async def __cache_set(self, urn: str, data: dict) -> bool:
if self._storage is not None:
if platform.system() == 'Windows': if platform.system() == 'Windows':
urn = urn.replace(':', '_') urn = urn.replace(':', '_')
return await self._storage.save_async( 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)
self._ram_cache[urn] = data
return True
def __spec_format2dtype(self, format_: str) -> str: def __spec_format2dtype(self, format_: str) -> str:
# 'string'|'bool'|'uint8'|'uint16'|'uint32'| # 'string'|'bool'|'uint8'|'uint16'|'uint32'|