Compare commits

...

3 Commits

Author SHA1 Message Date
topsworld
3399e3bb20 feat: update entity value-range 2025-01-08 19:29:39 +08:00
topsworld
078adfbd4c feat: update std_lib and multi_lang logic 2025-01-08 15:23:31 +08:00
topsworld
e4dfdf68ab feat: remove spec cache storage 2025-01-08 11:00:41 +08:00
11 changed files with 303 additions and 274 deletions

View File

@ -180,26 +180,26 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
self._attr_hvac_modes = list(self._hvac_mode_map.values())
self._prop_mode = prop
elif prop.name == 'target-temperature':
if not isinstance(prop.value_range, dict):
if not prop.value_range:
_LOGGER.error(
'invalid target-temperature value_range format, %s',
self.entity_id)
continue
self._attr_min_temp = prop.value_range['min']
self._attr_max_temp = prop.value_range['max']
self._attr_target_temperature_step = prop.value_range['step']
self._attr_min_temp = prop.value_range.min_
self._attr_max_temp = prop.value_range.max_
self._attr_target_temperature_step = prop.value_range.step
self._attr_temperature_unit = prop.external_unit
self._attr_supported_features |= (
ClimateEntityFeature.TARGET_TEMPERATURE)
self._prop_target_temp = prop
elif prop.name == 'target-humidity':
if not isinstance(prop.value_range, dict):
if not prop.value_range:
_LOGGER.error(
'invalid target-humidity value_range format, %s',
self.entity_id)
continue
self._attr_min_humidity = prop.value_range['min']
self._attr_max_humidity = prop.value_range['max']
self._attr_min_humidity = prop.value_range.min_
self._attr_max_humidity = prop.value_range.max_
self._attr_supported_features |= (
ClimateEntityFeature.TARGET_HUMIDITY)
self._prop_target_humi = prop
@ -517,14 +517,14 @@ class Heater(MIoTServiceEntity, ClimateEntity):
ClimateEntityFeature.TURN_OFF)
self._prop_on = prop
elif prop.name == 'target-temperature':
if not isinstance(prop.value_range, dict):
if not prop.value_range:
_LOGGER.error(
'invalid target-temperature value_range format, %s',
self.entity_id)
continue
self._attr_min_temp = prop.value_range['min']
self._attr_max_temp = prop.value_range['max']
self._attr_target_temperature_step = prop.value_range['step']
self._attr_min_temp = prop.value_range.min_
self._attr_max_temp = prop.value_range.max_
self._attr_target_temperature_step = prop.value_range.step
self._attr_temperature_unit = prop.external_unit
self._attr_supported_features |= (
ClimateEntityFeature.TARGET_TEMPERATURE)

View File

@ -172,13 +172,13 @@ class Cover(MIoTServiceEntity, CoverEntity):
elif prop.name == 'current-position':
self._prop_current_position = prop
elif prop.name == 'target-position':
if not isinstance(prop.value_range, dict):
if not prop.value_range:
_LOGGER.error(
'invalid target-position value_range format, %s',
self.entity_id)
continue
self._prop_position_value_min = prop.value_range['min']
self._prop_position_value_max = prop.value_range['max']
self._prop_position_value_min = prop.value_range.min_
self._prop_position_value_max = prop.value_range.max_
self._prop_position_value_range = (
self._prop_position_value_max -
self._prop_position_value_min)

View File

@ -119,11 +119,11 @@ class Fan(MIoTServiceEntity, FanEntity):
self._attr_supported_features |= FanEntityFeature.TURN_OFF
self._prop_on = prop
elif prop.name == 'fan-level':
if isinstance(prop.value_range, dict):
if prop.value_range:
# Fan level with value-range
self._speed_min = prop.value_range['min']
self._speed_max = prop.value_range['max']
self._speed_step = prop.value_range['step']
self._speed_min = prop.value_range.min_
self._speed_max = prop.value_range.max_
self._speed_step = prop.value_range.step
self._attr_speed_count = self._speed_max - self._speed_min+1
self._attr_supported_features |= FanEntityFeature.SET_SPEED
self._prop_fan_level = prop

View File

@ -119,13 +119,13 @@ class Humidifier(MIoTServiceEntity, HumidifierEntity):
self._prop_on = prop
# target-humidity
elif prop.name == 'target-humidity':
if not isinstance(prop.value_range, dict):
if not prop.value_range:
_LOGGER.error(
'invalid target-humidity value_range format, %s',
self.entity_id)
continue
self._attr_min_humidity = prop.value_range['min']
self._attr_max_humidity = prop.value_range['max']
self._attr_min_humidity = prop.value_range.min_
self._attr_max_humidity = prop.value_range.max_
self._prop_target_humidity = prop
# mode
elif prop.name == 'mode':

View File

@ -131,9 +131,9 @@ class Light(MIoTServiceEntity, LightEntity):
self._prop_on = prop
# brightness
if prop.name == 'brightness':
if isinstance(prop.value_range, dict):
if prop.value_range:
self._brightness_scale = (
prop.value_range['min'], prop.value_range['max'])
prop.value_range.min_, prop.value_range.max_)
self._prop_brightness = prop
elif (
self._mode_list is None
@ -153,13 +153,13 @@ class Light(MIoTServiceEntity, LightEntity):
continue
# color-temperature
if prop.name == 'color-temperature':
if not isinstance(prop.value_range, dict):
if not prop.value_range:
_LOGGER.info(
'invalid color-temperature value_range format, %s',
self.entity_id)
continue
self._attr_min_color_temp_kelvin = prop.value_range['min']
self._attr_max_color_temp_kelvin = prop.value_range['max']
self._attr_min_color_temp_kelvin = prop.value_range.min_
self._attr_max_color_temp_kelvin = prop.value_range.max_
self._attr_supported_color_modes.add(ColorMode.COLOR_TEMP)
self._attr_color_mode = ColorMode.COLOR_TEMP
self._prop_color_temp = prop
@ -178,13 +178,13 @@ class Light(MIoTServiceEntity, LightEntity):
mode_list = {
item['value']: item['description']
for item in prop.value_list}
elif isinstance(prop.value_range, dict):
elif prop.value_range:
mode_list = {}
if (
int((
prop.value_range['max']
- prop.value_range['min']
) / prop.value_range['step'])
prop.value_range.max_
- prop.value_range.min_
) / prop.value_range.step)
> self._VALUE_RANGE_MODE_COUNT_MAX
):
_LOGGER.info(
@ -192,9 +192,9 @@ class Light(MIoTServiceEntity, LightEntity):
self.entity_id, prop.name, prop.value_range)
else:
for value in range(
prop.value_range['min'],
prop.value_range['max'],
prop.value_range['step']):
prop.value_range.min_,
prop.value_range.max_,
prop.value_range.step):
mode_list[value] = f'mode {value}'
if mode_list:
self._mode_list = mode_list

View File

@ -94,7 +94,8 @@ from .miot_spec import (
MIoTSpecEvent,
MIoTSpecInstance,
MIoTSpecProperty,
MIoTSpecService
MIoTSpecService,
MIoTSpecValueRange
)
_LOGGER = logging.getLogger(__name__)
@ -1006,8 +1007,7 @@ class MIoTPropertyEntity(Entity):
service: MIoTSpecService
_main_loop: asyncio.AbstractEventLoop
# {'min':int, 'max':int, 'step': int}
_value_range: dict[str, int]
_value_range: Optional[MIoTSpecValueRange]
# {Any: Any}
_value_list: Optional[dict[Any, Any]]
_value: Any

View File

@ -46,12 +46,9 @@ off Xiaomi or its affiliates' products.
MIoT-Spec-V2 parser.
"""
import asyncio
import json
import platform
import time
from typing import Any, Optional
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from typing import Any, Optional, Union
import logging
@ -62,25 +59,50 @@ from .miot_error import MIoTSpecError
from .miot_storage import (
MIoTStorage,
SpecBoolTranslation,
SpecFilter,
SpecMultiLang)
SpecFilter)
_LOGGER = logging.getLogger(__name__)
class _MIoTSpecValueRange:
class MIoTSpecValueRange:
"""MIoT SPEC value range class."""
min_: int
max_: int
step: int
def from_list(self, value_range: list) -> None:
def __init__(self, value_range: Union[dict, list]) -> 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.max_ = value_range[1]
self.step = value_range[2]
def to_list(self) -> list:
return [self.min_, self.max_, self.step]
def dump(self) -> dict:
return {
'min': self.min_,
'max': self.max_,
'step': self.step
}
def __str__(self) -> str:
return f'[{self.min_}, {self.max_}, {self.step}'
class _MIoTSpecValueListItem:
@ -92,7 +114,7 @@ class _MIoTSpecValueListItem:
# Descriptions after multilingual conversion.
description: str
def to_dict(self) -> dict:
def dump(self) -> dict:
return {
'name': self.name,
'value': self.value,
@ -107,8 +129,8 @@ class _MIoTSpecValueList:
def to_map(self) -> dict:
return {item.value: item.description for item in self.items}
def to_list(self) -> list:
return [item.to_dict() for item in self.items]
def dump(self) -> list:
return [item.dump() for item in self.items]
class _SpecStdLib:
@ -132,7 +154,7 @@ class _SpecStdLib:
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 (
not isinstance(std_lib, dict)
or 'devices' not in std_lib
@ -211,7 +233,7 @@ class _SpecStdLib:
async def refresh_async(self) -> bool:
std_lib_new = await self.__request_from_cloud_async()
if std_lib_new:
self.from_dict(std_lib_new)
self.load(std_lib_new)
return True
return False
@ -353,7 +375,7 @@ class _SpecStdLib:
return result
class MIoTSpecBase:
class _MIoTSpecBase:
"""MIoT SPEC base class."""
iid: int
type_: str
@ -397,13 +419,14 @@ class MIoTSpecBase:
return self.spec_id == value.spec_id
class MIoTSpecProperty(MIoTSpecBase):
class MIoTSpecProperty(_MIoTSpecBase):
"""MIoT SPEC property class."""
format_: str
precision: int
unit: Optional[str]
precision: int
_value_range: Optional[MIoTSpecValueRange]
value_range: Optional[list]
value_list: Optional[list[dict]]
_access: list
@ -411,12 +434,18 @@ class MIoTSpecProperty(MIoTSpecBase):
_readable: bool
_notifiable: bool
service: MIoTSpecBase
service: 'MIoTSpecService'
def __init__(
self, spec: dict, service: MIoTSpecBase, format_: str, access: list,
unit: Optional[str] = None, value_range: Optional[list] = None,
value_list: Optional[list[dict]] = None, precision: int = 0
self,
spec: dict,
service: 'MIoTSpecService',
format_: str,
access: list,
unit: Optional[str] = None,
value_range: Optional[dict] = None,
value_list: Optional[list[dict]] = None,
precision: Optional[int] = None
) -> None:
super().__init__(spec=spec)
self.service = service
@ -425,7 +454,7 @@ class MIoTSpecProperty(MIoTSpecBase):
self.unit = unit
self.value_range = value_range
self.value_list = value_list
self.precision = precision
self.precision = precision or 1
self.spec_id = hash(
f'p.{self.name}.{self.service.iid}.{self.iid}')
@ -454,6 +483,21 @@ class MIoTSpecProperty(MIoTSpecBase):
def notifiable(self):
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:
"""Set value-range, precision."""
if not value:
self._value_range = None
return
self._value_range = MIoTSpecValueRange(value_range=value)
if isinstance(value, list):
self.precision = len(str(value[2]).split(
'.')[1].rstrip('0')) if '.' in str(value[2]) else 0
def value_format(self, value: Any) -> Any:
if value is None:
return None
@ -477,23 +521,24 @@ class MIoTSpecProperty(MIoTSpecBase):
'format': self.format_,
'access': self._access,
'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,
'precision': self.precision
}
class MIoTSpecEvent(MIoTSpecBase):
class MIoTSpecEvent(_MIoTSpecBase):
"""MIoT SPEC event class."""
argument: list[MIoTSpecProperty]
service: MIoTSpecBase
service: 'MIoTSpecService'
def __init__(
self, spec: dict, service: MIoTSpecBase,
argument: list[MIoTSpecProperty] = None
self, spec: dict, service: 'MIoTSpecService',
argument: Optional[list[MIoTSpecProperty]] = None
) -> None:
super().__init__(spec=spec)
self.argument = argument
self.argument = argument or []
self.service = service
self.spec_id = hash(
@ -512,20 +557,20 @@ class MIoTSpecEvent(MIoTSpecBase):
}
class MIoTSpecAction(MIoTSpecBase):
class MIoTSpecAction(_MIoTSpecBase):
"""MIoT SPEC action class."""
in_: list[MIoTSpecProperty]
out: list[MIoTSpecProperty]
service: MIoTSpecBase
service: 'MIoTSpecService'
def __init__(
self, spec: dict, service: MIoTSpecBase = None,
in_: list[MIoTSpecProperty] = None,
out: list[MIoTSpecProperty] = None
self, spec: dict, service: 'MIoTSpecService',
in_: Optional[list[MIoTSpecProperty]] = None,
out: Optional[list[MIoTSpecProperty]] = None
) -> None:
super().__init__(spec=spec)
self.in_ = in_
self.out = out
self.in_ = in_ or []
self.out = out or []
self.service = service
self.spec_id = hash(
@ -545,7 +590,7 @@ class MIoTSpecAction(MIoTSpecBase):
}
class MIoTSpecService(MIoTSpecBase):
class MIoTSpecService(_MIoTSpecBase):
"""MIoT SPEC service class."""
properties: list[MIoTSpecProperty]
events: list[MIoTSpecEvent]
@ -588,8 +633,7 @@ class MIoTSpecInstance:
icon: str
def __init__(
self, urn: str = None, name: str = None,
description: str = None, description_trans: str = None
self, urn: str, name: str, description: str, description_trans: str
) -> None:
self.urn = urn
self.name = name
@ -597,12 +641,13 @@ class MIoTSpecInstance:
self.description_trans = description_trans
self.services = []
def load(self, specs: dict) -> 'MIoTSpecInstance':
self.urn = specs['urn']
self.name = specs['name']
self.description = specs['description']
self.description_trans = specs['description_trans']
self.services = []
@staticmethod
def load(specs: dict) -> 'MIoTSpecInstance':
instance = MIoTSpecInstance(
urn=specs['urn'],
name=specs['name'],
description=specs['description'],
description_trans=specs['description_trans'])
for service in specs['services']:
spec_service = MIoTSpecService(spec=service)
for prop in service['properties']:
@ -614,7 +659,7 @@ class MIoTSpecInstance:
unit=prop['unit'],
value_range=prop['value_range'],
value_list=prop['value_list'],
precision=prop.get('precision', 0))
precision=prop.get('precision', None))
spec_service.properties.append(spec_prop)
for event in service['events']:
spec_event = MIoTSpecEvent(
@ -645,8 +690,8 @@ class MIoTSpecInstance:
break
spec_action.out = out_list
spec_service.actions.append(spec_action)
self.services.append(spec_service)
return self
instance.services.append(spec_service)
return instance
def dump(self) -> dict:
return {
@ -658,22 +703,16 @@ class MIoTSpecInstance:
}
class MIoTSpecParser:
"""MIoT SPEC parser."""
# pylint: disable=inconsistent-quotes
VERSION: int = 1
DOMAIN: str = 'miot_specs'
class _MIoTSpecMultiLang:
"""MIoT SPEC multi lang class."""
# pylint: disable=broad-exception-caught
_DOMAIN: str = 'miot_specs_multi_lang'
_lang: str
_storage: MIoTStorage
_main_loop: asyncio.AbstractEventLoop
_init_done: bool
_ram_cache: dict
_std_lib: _SpecStdLib
_bool_trans: SpecBoolTranslation
_multi_lang: SpecMultiLang
_spec_filter: SpecFilter
_custom_cache: dict[str, dict]
_current_data: Optional[dict[str, str]]
def __init__(
self, lang: Optional[str],
@ -684,23 +723,137 @@ class MIoTSpecParser:
self._storage = storage
self._main_loop = loop or asyncio.get_running_loop()
self._init_done = False
self._ram_cache = {}
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 trans_cloud and DEFAULT_INTEGRATION_LANGUAGE in trans_cloud:
trans_cache = trans_cloud[DEFAULT_INTEGRATION_LANGUAGE]
if trans_local and 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._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
@ -712,13 +865,13 @@ class MIoTSpecParser:
# Use the cache if the update time is less than 14 day
_LOGGER.debug(
'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
return
# 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())
@ -727,7 +880,7 @@ class MIoTSpecParser:
_LOGGER.error('save spec std lib failed')
else:
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')
else:
_LOGGER.error('load spec std lib failed')
@ -737,26 +890,24 @@ 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()
self._ram_cache.clear()
async def parse(
self, urn: str, skip_cache: bool = False,
) -> MIoTSpecInstance:
) -> Optional[MIoTSpecInstance]:
"""MUST await init first !!!"""
if not skip_cache:
cache_result = await self.__cache_get(urn=urn)
if isinstance(cache_result, dict):
_LOGGER.debug('get from cache, %s', urn)
return MIoTSpecInstance().load(specs=cache_result)
return MIoTSpecInstance.load(specs=cache_result)
# Retry three times
for index in range(3):
try:
return await self.__parse(urn=urn)
except Exception as err: # pylint: disable=broad-exception-caught
_LOGGER.error(
'parse error, retry, %d, %s, %s', index, urn, err)
# try:
return await self.__parse(urn=urn)
# except Exception as err: # pylint: disable=broad-exception-caught
# _LOGGER.error(
# 'parse error, retry, %d, %s, %s', index, urn, err)
return None
async def refresh_async(self, urn_list: list[str]) -> int:
@ -765,7 +916,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())
@ -784,21 +935,18 @@ class MIoTSpecParser:
return success_count
async def __cache_get(self, urn: str) -> Optional[dict]:
if self._storage is not None:
if platform.system() == 'Windows':
urn = urn.replace(':', '_')
return await self._storage.load_async(
domain=self.DOMAIN, name=f'{urn}_{self._lang}', type_=dict)
return self._ram_cache.get(urn, None)
if platform.system() == 'Windows':
urn = urn.replace(':', '_')
return await self._storage.load_async(
domain=self._DOMAIN,
name=f'{urn}_{self._lang}',
type_=dict) # type: ignore
async def __cache_set(self, urn: str, data: dict) -> bool:
if self._storage is not None:
if platform.system() == 'Windows':
urn = urn.replace(':', '_')
return await self._storage.save_async(
domain=self.DOMAIN, name=f'{urn}_{self._lang}', data=data)
self._ram_cache[urn] = data
return True
if platform.system() == 'Windows':
urn = urn.replace(':', '_')
return await self._storage.save_async(
domain=self._DOMAIN, name=f'{urn}_{self._lang}', data=data)
def __spec_format2dtype(self, format_: str) -> str:
# 'string'|'bool'|'uint8'|'uint16'|'uint32'|
@ -811,11 +959,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
@ -827,68 +970,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:
# Load multiple language configuration.
res_trans = await self.__get_translation(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')
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
# Set translation cache
await self._multi_lang.set_spec_async(urn=urn_key)
# Set spec filter
self._spec_filter.filter_spec(urn_key=urn_key)
# Parse device type
spec_instance: MIoTSpecInstance = MIoTSpecInstance(
@ -919,7 +1005,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
@ -950,30 +1036,22 @@ class MIoTSpecParser:
if p_type_strs[1] != 'miot-spec-v2':
spec_prop.proprietary = spec_service.proprietary or True
spec_prop.description_trans = (
translation.get(
f'p:{service["iid"]}:{property_["iid"]}', None)
self._multi_lang.translate(
f'p:{service["iid"]}:{property_["iid"]}')
or self._std_lib.property_translate(
key=':'.join(p_type_strs[:5]))
or property_['description']
or spec_prop.name)
if 'value-range' in property_:
spec_prop.value_range = {
'min': property_['value-range'][0],
'max': property_['value-range'][1],
'step': property_['value-range'][2]
}
spec_prop.precision = len(str(
property_['value-range'][2]).split(
'.')[1].rstrip('0')) if '.' in str(
property_['value-range'][2]) else 0
spec_prop.value_range = property_['value-range']
elif 'value-list' in property_:
v_list: list[dict] = property_['value-list']
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"]}')
@ -1008,8 +1086,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(
f'e:{service["iid"]}:{event["iid"]}', None)
self._multi_lang.translate(
f'e:{service["iid"]}:{event["iid"]}')
or self._std_lib.event_translate(
key=':'.join(e_type_strs[:5]))
or event['description']
@ -1044,8 +1122,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(
f'a:{service["iid"]}:{action["iid"]}', None)
self._multi_lang.translate(
f'a:{service["iid"]}:{action["iid"]}')
or self._std_lib.action_translate(
key=':'.join(a_type_strs[:5]))
or action['description']

View File

@ -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.

View File

@ -92,9 +92,9 @@ class Number(MIoTPropertyEntity, NumberEntity):
self._attr_icon = self.spec.icon
# Set value range
if self._value_range:
self._attr_native_min_value = self._value_range['min']
self._attr_native_max_value = self._value_range['max']
self._attr_native_step = self._value_range['step']
self._attr_native_min_value = self._value_range.min_
self._attr_native_max_value = self._value_range.max_
self._attr_native_step = self._value_range.step
@property
def native_value(self) -> Optional[float]:

View File

@ -115,8 +115,8 @@ class Sensor(MIoTPropertyEntity, SensorEntity):
"""Return the current value of the sensor."""
if self._value_range and isinstance(self._value, (int, float)):
if (
self._value < self._value_range['min']
or self._value > self._value_range['max']
self._value < self._value_range.min_
or self._value > self._value_range.max_
):
_LOGGER.info(
'%s, data exception, out of range, %s, %s',

View File

@ -115,7 +115,7 @@ class WaterHeater(MIoTServiceEntity, WaterHeaterEntity):
self._prop_on = prop
# temperature
if prop.name == 'temperature':
if isinstance(prop.value_range, dict):
if prop.value_range:
if (
self._attr_temperature_unit is None
and prop.external_unit
@ -128,9 +128,14 @@ class WaterHeater(MIoTServiceEntity, WaterHeaterEntity):
self.entity_id)
# target-temperature
if prop.name == 'target-temperature':
self._attr_min_temp = prop.value_range['min']
self._attr_max_temp = prop.value_range['max']
self._attr_precision = prop.value_range['step']
if not prop.value_range:
_LOGGER.error(
'invalid target-temperature value_range format, %s',
self.entity_id)
continue
self._attr_min_temp = prop.value_range.min_
self._attr_max_temp = prop.value_range.max_
self._attr_precision = prop.value_range.step
if self._attr_temperature_unit is None and prop.external_unit:
self._attr_temperature_unit = prop.external_unit
self._attr_supported_features |= (