mirror of
https://github.com/XiaoMi/ha_xiaomi_home.git
synced 2026-01-20 01:09:36 +08:00
Compare commits
3 Commits
d9d8433405
...
3399e3bb20
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
3399e3bb20 | ||
|
|
078adfbd4c | ||
|
|
e4dfdf68ab |
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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':
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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']
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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]:
|
||||
|
||||
@ -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',
|
||||
|
||||
@ -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 |= (
|
||||
|
||||
Loading…
Reference in New Issue
Block a user