Compare commits

...

3 Commits

Author SHA1 Message Date
topsworld
fca97e03f0 fix: fix miot cloud log error 2025-01-09 15:04:45 +08:00
topsworld
93f04b1aee feat: update prop.format_ logic 2025-01-09 15:04:23 +08:00
topsworld
d25d3f6a93 feat: update value-list logic 2025-01-09 13:23:35 +08:00
14 changed files with 277 additions and 274 deletions

View File

@ -156,27 +156,24 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
_LOGGER.error(
'unknown on property, %s', self.entity_id)
elif prop.name == 'mode':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'invalid mode value_list, %s', self.entity_id)
continue
self._hvac_mode_map = {}
for item in prop.value_list:
if item['name'].lower() in {'off', 'idle'}:
self._hvac_mode_map[item['value']] = HVACMode.OFF
elif item['name'].lower() in {'auto'}:
self._hvac_mode_map[item['value']] = HVACMode.AUTO
elif item['name'].lower() in {'cool'}:
self._hvac_mode_map[item['value']] = HVACMode.COOL
elif item['name'].lower() in {'heat'}:
self._hvac_mode_map[item['value']] = HVACMode.HEAT
elif item['name'].lower() in {'dry'}:
self._hvac_mode_map[item['value']] = HVACMode.DRY
elif item['name'].lower() in {'fan'}:
self._hvac_mode_map[item['value']] = HVACMode.FAN_ONLY
for item in prop.value_list.items:
if item.name in {'off', 'idle'}:
self._hvac_mode_map[item.value] = HVACMode.OFF
elif item.name in {'auto'}:
self._hvac_mode_map[item.value] = HVACMode.AUTO
elif item.name in {'cool'}:
self._hvac_mode_map[item.value] = HVACMode.COOL
elif item.name in {'heat'}:
self._hvac_mode_map[item.value] = HVACMode.HEAT
elif item.name in {'dry'}:
self._hvac_mode_map[item.value] = HVACMode.DRY
elif item.name in {'fan'}:
self._hvac_mode_map[item.value] = HVACMode.FAN_ONLY
self._attr_hvac_modes = list(self._hvac_mode_map.values())
self._prop_mode = prop
elif prop.name == 'target-temperature':
@ -204,16 +201,11 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
ClimateEntityFeature.TARGET_HUMIDITY)
self._prop_target_humi = prop
elif prop.name == 'fan-level':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'invalid fan-level value_list, %s', self.entity_id)
continue
self._fan_mode_map = {
item['value']: item['description']
for item in prop.value_list}
self._fan_mode_map = prop.value_list.to_map()
self._attr_fan_modes = list(self._fan_mode_map.values())
self._attr_supported_features |= ClimateEntityFeature.FAN_MODE
self._prop_fan_level = prop
@ -269,8 +261,8 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
elif self.get_prop_value(prop=self._prop_on) is False:
await self.set_property_async(prop=self._prop_on, value=True)
# set mode
mode_value = self.get_map_value(
map_=self._hvac_mode_map, description=hvac_mode)
mode_value = self.get_map_key(
map_=self._hvac_mode_map, value=hvac_mode)
if (
mode_value is None or
not await self.set_property_async(
@ -339,8 +331,8 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
async def async_set_fan_mode(self, fan_mode):
"""Set new target fan mode."""
mode_value = self.get_map_value(
map_=self._fan_mode_map, description=fan_mode)
mode_value = self.get_map_key(
map_=self._fan_mode_map, value=fan_mode)
if mode_value is None or not await self.set_property_async(
prop=self._prop_fan_level, value=mode_value):
raise RuntimeError(
@ -376,9 +368,9 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
"""Return the hvac mode. e.g., heat, cool mode."""
if self.get_prop_value(prop=self._prop_on) is False:
return HVACMode.OFF
return self.get_map_description(
return self.get_map_key(
map_=self._hvac_mode_map,
key=self.get_prop_value(prop=self._prop_mode))
value=self.get_prop_value(prop=self._prop_mode))
@property
def fan_mode(self) -> Optional[str]:
@ -386,7 +378,7 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
Requires ClimateEntityFeature.FAN_MODE.
"""
return self.get_map_description(
return self.get_map_value(
map_=self._fan_mode_map,
key=self.get_prop_value(prop=self._prop_fan_level))
@ -446,8 +438,8 @@ class AirConditioner(MIoTServiceEntity, ClimateEntity):
}.get(v_ac_state['M'], None)
if mode:
self.set_prop_value(
prop=self._prop_mode, value=self.get_map_value(
map_=self._hvac_mode_map, description=mode))
prop=self._prop_mode, value=self.get_map_key(
map_=self._hvac_mode_map, value=mode))
# T: target temperature
if 'T' in v_ac_state and self._prop_target_temp:
self.set_prop_value(prop=self._prop_target_temp,
@ -530,16 +522,11 @@ class Heater(MIoTServiceEntity, ClimateEntity):
ClimateEntityFeature.TARGET_TEMPERATURE)
self._prop_target_temp = prop
elif prop.name == 'heat-level':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'invalid heat-level value_list, %s', self.entity_id)
continue
self._heat_level_map = {
item['value']: item['description']
for item in prop.value_list}
self._heat_level_map = prop.value_list.to_map()
self._attr_preset_modes = list(self._heat_level_map.values())
self._attr_supported_features |= (
ClimateEntityFeature.PRESET_MODE)
@ -582,8 +569,8 @@ class Heater(MIoTServiceEntity, ClimateEntity):
"""Set the preset mode."""
await self.set_property_async(
self._prop_heat_level,
value=self.get_map_value(
map_=self._heat_level_map, description=preset_mode))
value=self.get_map_key(
map_=self._heat_level_map, value=preset_mode))
@property
def target_temperature(self) -> Optional[float]:
@ -613,7 +600,7 @@ class Heater(MIoTServiceEntity, ClimateEntity):
@property
def preset_mode(self) -> Optional[str]:
return (
self.get_map_description(
self.get_map_value(
map_=self._heat_level_map,
key=self.get_prop_value(prop=self._prop_heat_level))
if self._prop_heat_level else None)

View File

@ -132,42 +132,36 @@ class Cover(MIoTServiceEntity, CoverEntity):
# properties
for prop in entity_data.props:
if prop.name == 'motor-control':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'motor-control value_list is None, %s', self.entity_id)
continue
for item in prop.value_list:
if item['name'].lower() in ['open']:
for item in prop.value_list.items:
if item.name in {'open'}:
self._attr_supported_features |= (
CoverEntityFeature.OPEN)
self._prop_motor_value_open = item['value']
elif item['name'].lower() in ['close']:
self._prop_motor_value_open = item.value
elif item.name in {'close'}:
self._attr_supported_features |= (
CoverEntityFeature.CLOSE)
self._prop_motor_value_close = item['value']
elif item['name'].lower() in ['pause']:
self._prop_motor_value_close = item.value
elif item.name in {'pause'}:
self._attr_supported_features |= (
CoverEntityFeature.STOP)
self._prop_motor_value_pause = item['value']
self._prop_motor_value_pause = item.value
self._prop_motor_control = prop
elif prop.name == 'status':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'status value_list is None, %s', self.entity_id)
continue
for item in prop.value_list:
if item['name'].lower() in ['opening', 'open']:
self._prop_status_opening = item['value']
elif item['name'].lower() in ['closing', 'close']:
self._prop_status_closing = item['value']
elif item['name'].lower() in ['stop', 'pause']:
self._prop_status_stop = item['value']
for item in prop.value_list.items:
if item.name in {'opening', 'open'}:
self._prop_status_opening = item.value
elif item.name in {'closing', 'close'}:
self._prop_status_closing = item.value
elif item.name in {'stop', 'pause'}:
self._prop_status_stop = item.value
self._prop_status = prop
elif prop.name == 'current-position':
self._prop_current_position = prop

View File

@ -90,10 +90,10 @@ class Fan(MIoTServiceEntity, FanEntity):
_prop_mode: Optional[MIoTSpecProperty]
_prop_horizontal_swing: Optional[MIoTSpecProperty]
_speed_min: Optional[int]
_speed_max: Optional[int]
_speed_step: Optional[int]
_mode_list: Optional[dict[Any, Any]]
_speed_min: int
_speed_max: int
_speed_step: int
_mode_map: Optional[dict[Any, Any]]
def __init__(
self, miot_device: MIoTDevice, entity_data: MIoTEntityData
@ -110,7 +110,7 @@ class Fan(MIoTServiceEntity, FanEntity):
self._speed_min = 65535
self._speed_max = 0
self._speed_step = 1
self._mode_list = None
self._mode_map = None
# properties
for prop in entity_data.props:
@ -129,47 +129,28 @@ class Fan(MIoTServiceEntity, FanEntity):
self._prop_fan_level = prop
elif (
self._prop_fan_level is None
and isinstance(prop.value_list, list)
and prop.value_list
):
# Fan level with value-list
for item in prop.value_list:
self._speed_min = min(self._speed_min, item['value'])
self._speed_max = max(self._speed_max, item['value'])
for item in prop.value_list.items:
self._speed_min = min(self._speed_min, item.value)
self._speed_max = max(self._speed_max, item.value)
self._attr_speed_count = self._speed_max - self._speed_min+1
self._attr_supported_features |= FanEntityFeature.SET_SPEED
self._prop_fan_level = prop
elif prop.name == 'mode':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'mode value_list is None, %s', self.entity_id)
continue
self._mode_list = {
item['value']: item['description']
for item in prop.value_list}
self._attr_preset_modes = list(self._mode_list.values())
self._mode_map = prop.value_list.to_map()
self._attr_preset_modes = list(self._mode_map.values())
self._attr_supported_features |= FanEntityFeature.PRESET_MODE
self._prop_mode = prop
elif prop.name == 'horizontal-swing':
self._attr_supported_features |= FanEntityFeature.OSCILLATE
self._prop_horizontal_swing = prop
def __get_mode_description(self, key: int) -> Optional[str]:
if self._mode_list is None:
return None
return self._mode_list.get(key, None)
def __get_mode_value(self, description: str) -> Optional[int]:
if self._mode_list is None:
return None
for key, value in self._mode_list.items():
if value == description:
return key
return None
async def async_turn_on(
self, percentage: int = None, preset_mode: str = None, **kwargs: Any
) -> None:
@ -189,7 +170,8 @@ class Fan(MIoTServiceEntity, FanEntity):
if preset_mode:
await self.set_property_async(
self._prop_mode,
value=self.__get_mode_value(description=preset_mode))
value=self.get_map_key(
map_=self._mode_map, value=preset_mode))
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the fan off."""
@ -217,7 +199,8 @@ class Fan(MIoTServiceEntity, FanEntity):
"""Set the preset mode."""
await self.set_property_async(
self._prop_mode,
value=self.__get_mode_value(description=preset_mode))
value=self.get_map_key(
map_=self._mode_map, value=preset_mode))
async def async_set_direction(self, direction: str) -> None:
"""Set the direction of the fan."""
@ -238,7 +221,8 @@ class Fan(MIoTServiceEntity, FanEntity):
"""Return the current preset mode,
e.g., auto, smart, eco, favorite."""
return (
self.__get_mode_description(
self.get_map_value(
map_=self._mode_map,
key=self.get_prop_value(prop=self._prop_mode))
if self._prop_mode else None)

View File

@ -97,7 +97,7 @@ class Humidifier(MIoTServiceEntity, HumidifierEntity):
_prop_target_humidity: Optional[MIoTSpecProperty]
_prop_humidity: Optional[MIoTSpecProperty]
_mode_list: dict[Any, Any]
_mode_map: dict[Any, Any]
def __init__(
self, miot_device: MIoTDevice, entity_data: MIoTEntityData
@ -110,7 +110,7 @@ class Humidifier(MIoTServiceEntity, HumidifierEntity):
self._prop_mode = None
self._prop_target_humidity = None
self._prop_humidity = None
self._mode_list = None
self._mode_map = None
# properties
for prop in entity_data.props:
@ -129,18 +129,13 @@ class Humidifier(MIoTServiceEntity, HumidifierEntity):
self._prop_target_humidity = prop
# mode
elif prop.name == 'mode':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'mode value_list is None, %s', self.entity_id)
continue
self._mode_list = {
item['value']: item['description']
for item in prop.value_list}
self._mode_map = prop.value_list.to_map()
self._attr_available_modes = list(
self._mode_list.values())
self._mode_map.values())
self._attr_supported_features |= HumidifierEntityFeature.MODES
self._prop_mode = prop
# relative-humidity
@ -163,7 +158,8 @@ class Humidifier(MIoTServiceEntity, HumidifierEntity):
async def async_set_mode(self, mode: str) -> None:
"""Set new target preset mode."""
await self.set_property_async(
prop=self._prop_mode, value=self.__get_mode_value(description=mode))
prop=self._prop_mode,
value=self.get_map_key(map_=self._mode_map, value=mode))
@property
def is_on(self) -> Optional[bool]:
@ -183,20 +179,6 @@ class Humidifier(MIoTServiceEntity, HumidifierEntity):
@property
def mode(self) -> Optional[str]:
"""Return the current preset mode."""
return self.__get_mode_description(
return self.get_map_value(
map_=self._mode_map,
key=self.get_prop_value(prop=self._prop_mode))
def __get_mode_description(self, key: int) -> Optional[str]:
"""Convert mode value to description."""
if self._mode_list is None:
return None
return self._mode_list.get(key, None)
def __get_mode_value(self, description: str) -> Optional[int]:
"""Convert mode description to value."""
if self._mode_list is None:
return None
for key, value in self._mode_list.items():
if value == description:
return key
return None

View File

@ -96,14 +96,14 @@ class Light(MIoTServiceEntity, LightEntity):
"""Light entities for Xiaomi Home."""
# pylint: disable=unused-argument
_VALUE_RANGE_MODE_COUNT_MAX = 30
_prop_on: Optional[MIoTSpecProperty]
_prop_on: MIoTSpecProperty
_prop_brightness: Optional[MIoTSpecProperty]
_prop_color_temp: Optional[MIoTSpecProperty]
_prop_color: Optional[MIoTSpecProperty]
_prop_mode: Optional[MIoTSpecProperty]
_brightness_scale: Optional[tuple[int, int]]
_mode_list: Optional[dict[Any, Any]]
_mode_map: Optional[dict[Any, Any]]
def __init__(
self, miot_device: MIoTDevice, entity_data: MIoTEntityData
@ -122,7 +122,7 @@ class Light(MIoTServiceEntity, LightEntity):
self._prop_color = None
self._prop_mode = None
self._brightness_scale = None
self._mode_list = None
self._mode_map = None
# properties
for prop in entity_data.props:
@ -136,15 +136,12 @@ class Light(MIoTServiceEntity, LightEntity):
prop.value_range.min_, prop.value_range.max_)
self._prop_brightness = prop
elif (
self._mode_list is None
and isinstance(prop.value_list, list)
self._mode_map is None
and prop.value_list
):
# For value-list brightness
self._mode_list = {
item['value']: item['description']
for item in prop.value_list}
self._attr_effect_list = list(self._mode_list.values())
self._mode_map = prop.value_list.to_map()
self._attr_effect_list = list(self._mode_map.values())
self._attr_supported_features |= LightEntityFeature.EFFECT
self._prop_mode = prop
else:
@ -171,13 +168,8 @@ class Light(MIoTServiceEntity, LightEntity):
# mode
if prop.name == 'mode':
mode_list = None
if (
isinstance(prop.value_list, list)
and prop.value_list
):
mode_list = {
item['value']: item['description']
for item in prop.value_list}
if prop.value_list:
mode_list = prop.value_list.to_map()
elif prop.value_range:
mode_list = {}
if (
@ -197,8 +189,8 @@ class Light(MIoTServiceEntity, LightEntity):
prop.value_range.step):
mode_list[value] = f'mode {value}'
if mode_list:
self._mode_list = mode_list
self._attr_effect_list = list(self._mode_list.values())
self._mode_map = mode_list
self._attr_effect_list = list(self._mode_map.values())
self._attr_supported_features |= LightEntityFeature.EFFECT
self._prop_mode = prop
else:
@ -213,21 +205,6 @@ class Light(MIoTServiceEntity, LightEntity):
self._attr_supported_color_modes.add(ColorMode.ONOFF)
self._attr_color_mode = ColorMode.ONOFF
def __get_mode_description(self, key: int) -> Optional[str]:
"""Convert mode value to description."""
if self._mode_list is None:
return None
return self._mode_list.get(key, None)
def __get_mode_value(self, description: str) -> Optional[int]:
"""Convert mode description to value."""
if self._mode_list is None:
return None
for key, value in self._mode_list.items():
if value == description:
return key
return None
@property
def is_on(self) -> Optional[bool]:
"""Return if the light is on."""
@ -264,7 +241,8 @@ class Light(MIoTServiceEntity, LightEntity):
@property
def effect(self) -> Optional[str]:
"""Return the current mode."""
return self.__get_mode_description(
return self.get_map_value(
map_=self._mode_map,
key=self.get_prop_value(prop=self._prop_mode))
async def async_turn_on(self, **kwargs) -> None:
@ -275,7 +253,7 @@ class Light(MIoTServiceEntity, LightEntity):
result: bool = False
# on
# Dirty logic for lumi.gateway.mgl03 indicator light
value_on = True if self._prop_on.format_ == 'bool' else 1
value_on = True if self._prop_on.format_ == bool else 1
result = await self.set_property_async(
prop=self._prop_on, value=value_on)
# brightness
@ -303,11 +281,12 @@ class Light(MIoTServiceEntity, LightEntity):
if ATTR_EFFECT in kwargs:
result = await self.set_property_async(
prop=self._prop_mode,
value=self.__get_mode_value(description=kwargs[ATTR_EFFECT]))
value=self.get_map_key(
map_=self._mode_map, value=kwargs[ATTR_EFFECT]))
return result
async def async_turn_off(self, **kwargs) -> None:
"""Turn the light off."""
# Dirty logic for lumi.gateway.mgl03 indicator light
value_on = False if self._prop_on.format_ == 'bool' else 0
value_on = False if self._prop_on.format_ == bool else 0
return await self.set_property_async(prop=self._prop_on, value=value_on)

View File

@ -720,7 +720,7 @@ class MIoTHttpClient:
prop_obj['fut'].set_result(None)
if props_req:
_LOGGER.info(
'get prop from cloud failed, %s, %s', len(key), props_req)
'get prop from cloud failed, %s', props_req)
if self._get_prop_list:
self._get_prop_timer = self._main_loop.call_later(

View File

@ -95,6 +95,7 @@ from .miot_spec import (
MIoTSpecInstance,
MIoTSpecProperty,
MIoTSpecService,
MIoTSpecValueList,
MIoTSpecValueRange
)
@ -512,7 +513,7 @@ class MIoTDevice:
if prop_access != (SPEC_PROP_TRANS_MAP[
'entities'][platform]['access']):
return None
if prop.format_ not in SPEC_PROP_TRANS_MAP[
if prop.format_.__name__ not in SPEC_PROP_TRANS_MAP[
'entities'][platform]['format']:
return None
if prop.unit:
@ -565,9 +566,9 @@ class MIoTDevice:
# general conversion
if not prop.platform:
if prop.writable:
if prop.format_ == 'str':
if prop.format_ == str:
prop.platform = 'text'
elif prop.format_ == 'bool':
elif prop.format_ == bool:
prop.platform = 'switch'
prop.device_class = SwitchDeviceClass.SWITCH
elif prop.value_list:
@ -837,18 +838,20 @@ class MIoTServiceEntity(Entity):
self.miot_device.unsub_event(
siid=event.service.iid, eiid=event.iid)
def get_map_description(self, map_: dict[int, Any], key: int) -> Any:
def get_map_value(
self, map_: dict[int, Any], key: int
) -> Any:
if map_ is None:
return None
return map_.get(key, None)
def get_map_value(
self, map_: dict[int, Any], description: Any
def get_map_key(
self, map_: dict[int, Any], value: Any
) -> Optional[int]:
if map_ is None:
return None
for key, value in map_.items():
if value == description:
for key, value_ in map_.items():
if value_ == value:
return key
return None
@ -1009,7 +1012,7 @@ class MIoTPropertyEntity(Entity):
_main_loop: asyncio.AbstractEventLoop
_value_range: Optional[MIoTSpecValueRange]
# {Any: Any}
_value_list: Optional[dict[Any, Any]]
_value_list: Optional[MIoTSpecValueList]
_value: Any
_pending_write_ha_state_timer: Optional[asyncio.TimerHandle]
@ -1022,11 +1025,7 @@ class MIoTPropertyEntity(Entity):
self.service = spec.service
self._main_loop = miot_device.miot_client.main_loop
self._value_range = spec.value_range
if spec.value_list:
self._value_list = {
item['value']: item['description'] for item in spec.value_list}
else:
self._value_list = None
self._value_list = spec.value_list
self._value = None
self._pending_write_ha_state_timer = None
# Gen entity_id
@ -1077,15 +1076,12 @@ class MIoTPropertyEntity(Entity):
def get_vlist_description(self, value: Any) -> Optional[str]:
if not self._value_list:
return None
return self._value_list.get(value, None)
return self._value_list.get_description_by_value(value)
def get_vlist_value(self, description: str) -> Any:
if not self._value_list:
return None
for key, value in self._value_list.items():
if value == description:
return key
return None
return self._value_list.get_value_by_description(description)
async def set_property_async(self, value: Any) -> bool:
if not self.spec.writable:

View File

@ -48,8 +48,9 @@ MIoT-Spec-V2 parser.
import asyncio
import platform
import time
from typing import Any, Optional, Union
from typing import Any, Optional, Type, Union
import logging
from slugify import slugify
# pylint: disable=relative-beyond-top-level
@ -75,6 +76,8 @@ class MIoTSpecValueRange:
self.load(value_range)
elif isinstance(value_range, list):
self.from_spec(value_range)
else:
raise MIoTSpecError('invalid value range format')
def load(self, value_range: dict) -> None:
if (
@ -105,15 +108,42 @@ class MIoTSpecValueRange:
return f'[{self.min_}, {self.max_}, {self.step}'
class _MIoTSpecValueListItem:
class MIoTSpecValueListItem:
"""MIoT SPEC value list item class."""
# All lower-case SPEC description.
# NOTICE: bool type without name
name: str
# Value
value: Any
# Descriptions after multilingual conversion.
description: str
def __init__(self, item: dict) -> None:
self.load(item)
def load(self, item: dict) -> None:
if 'value' not in item or 'description' not in item:
raise MIoTSpecError('invalid value list item, %s')
self.name = item.get('name', None)
self.value = item['value']
self.description = item['description']
@staticmethod
def from_spec(item: dict) -> 'MIoTSpecValueListItem':
if (
'name' not in item
or 'value' not in item
or 'description' not in item
):
raise MIoTSpecError('invalid value list item, %s')
# Slugify name and convert to lower-case.
cache = {
'name': slugify(text=item['name'], separator='_').lower(),
'value': item['value'],
'description': item['description']
}
return MIoTSpecValueListItem(cache)
def dump(self) -> dict:
return {
'name': self.name,
@ -121,14 +151,69 @@ class _MIoTSpecValueListItem:
'description': self.description
}
def __str__(self) -> str:
return f'{self.name}: {self.value} - {self.description}'
class _MIoTSpecValueList:
class MIoTSpecValueList:
"""MIoT SPEC value list class."""
items: list[_MIoTSpecValueListItem]
items: list[MIoTSpecValueListItem]
def __init__(self, value_list: list[dict]) -> None:
if not isinstance(value_list, list):
raise MIoTSpecError('invalid value list format')
self.items = []
self.load(value_list)
@property
def names(self) -> list[str]:
return [item.name for item in self.items]
@property
def values(self) -> list[Any]:
return [item.value for item in self.items]
@property
def descriptions(self) -> list[str]:
return [item.description for item in self.items]
@staticmethod
def from_spec(value_list: list[dict]) -> 'MIoTSpecValueList':
result = MIoTSpecValueList([])
dup_desc: dict[str, int] = {}
for item in value_list:
# Handle duplicate descriptions.
count = 0
if item['description'] in dup_desc:
count = dup_desc[item['description']]
count += 1
dup_desc[item['description']] = count
if count > 1:
item['name'] = f'{item["name"]}_{count}'
item['description'] = f'{item["description"]}_{count}'
result.items.append(MIoTSpecValueListItem.from_spec(item))
return result
def load(self, value_list: list[dict]) -> None:
for item in value_list:
self.items.append(MIoTSpecValueListItem(item))
def to_map(self) -> dict:
return {item.value: item.description for item in self.items}
def get_value_by_description(self, description: str) -> Any:
for item in self.items:
if item.description == description:
return item.value
return None
def get_description_by_value(self, value: Any) -> Optional[str]:
for item in self.items:
if item.value == value:
return item.description
return None
def dump(self) -> list:
return [item.dump() for item in self.items]
@ -421,13 +506,12 @@ class _MIoTSpecBase:
class MIoTSpecProperty(_MIoTSpecBase):
"""MIoT SPEC property class."""
format_: str
unit: Optional[str]
precision: int
_format_: Type
_value_range: Optional[MIoTSpecValueRange]
value_list: Optional[list[dict]]
_value_list: Optional[MIoTSpecValueList]
_access: list
_writable: bool
@ -459,6 +543,19 @@ class MIoTSpecProperty(_MIoTSpecBase):
self.spec_id = hash(
f'p.{self.name}.{self.service.iid}.{self.iid}')
@property
def format_(self) -> Type:
return self._format_
@format_.setter
def format_(self, value: str) -> None:
self._format_ = {
'string': str,
'str': str,
'bool': bool,
'float': float}.get(
value, int)
@property
def access(self) -> list:
return self._access
@ -498,15 +595,31 @@ class MIoTSpecProperty(_MIoTSpecBase):
self.precision = len(str(value[2]).split(
'.')[1].rstrip('0')) if '.' in str(value[2]) else 0
@property
def value_list(self) -> Optional[MIoTSpecValueList]:
return self._value_list
@value_list.setter
def value_list(
self, value: Union[list[dict], MIoTSpecValueList, None]
) -> None:
if not value:
self._value_list = None
return
if isinstance(value, list):
self._value_list = MIoTSpecValueList(value_list=value)
elif isinstance(value, MIoTSpecValueList):
self._value_list = value
def value_format(self, value: Any) -> Any:
if value is None:
return None
if self.format_ == 'int':
if self.format_ == int:
return int(value)
if self.format_ == 'float':
if self.format_ == float:
return round(value, self.precision)
if self.format_ == 'bool':
return bool(value in [True, 1, 'true', '1'])
if self.format_ == bool:
return bool(value in [True, 1, 'True', 'true', '1'])
return value
def dump(self) -> dict:
@ -518,12 +631,12 @@ class MIoTSpecProperty(_MIoTSpecBase):
'description_trans': self.description_trans,
'proprietary': self.proprietary,
'need_filter': self.need_filter,
'format': self.format_,
'format': self.format_.__name__,
'access': self._access,
'unit': self.unit,
'value_range': (
self.value_range.dump() if self.value_range else None),
'value_list': self.value_list,
self._value_range.dump() if self._value_range else None),
'value_list': self._value_list.dump() if self._value_list else None,
'precision': self.precision
}
@ -552,8 +665,8 @@ class MIoTSpecEvent(_MIoTSpecBase):
'description': self.description,
'description_trans': self.description_trans,
'proprietary': self.proprietary,
'need_filter': self.need_filter,
'argument': [prop.iid for prop in self.argument],
'need_filter': self.need_filter
}
@ -583,10 +696,10 @@ class MIoTSpecAction(_MIoTSpecBase):
'iid': self.iid,
'description': self.description,
'description_trans': self.description_trans,
'proprietary': self.proprietary,
'need_filter': self.need_filter,
'in': [prop.iid for prop in self.in_],
'out': [prop.iid for prop in self.out]
'out': [prop.iid for prop in self.out],
'proprietary': self.proprietary,
'need_filter': self.need_filter
}
@ -611,9 +724,9 @@ class MIoTSpecService(_MIoTSpecBase):
'description_trans': self.description_trans,
'proprietary': self.proprietary,
'properties': [prop.dump() for prop in self.properties],
'need_filter': self.need_filter,
'events': [event.dump() for event in self.events],
'actions': [action.dump() for action in self.actions],
'need_filter': self.need_filter
}
@ -903,11 +1016,11 @@ class MIoTSpecParser:
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:
@ -948,12 +1061,6 @@ class MIoTSpecParser:
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'|
# 'int8'|'int16'|'int32'|'int64'|'float'
return {'string': 'str', 'bool': 'bool', 'float': 'float'}.get(
format_, 'int')
async def __get_instance(self, urn: str) -> Optional[dict]:
return await MIoTHttp.get_json_async(
url='https://miot-spec.org/miot-spec-v2/instance',
@ -1024,7 +1131,7 @@ class MIoTSpecParser:
spec_prop: MIoTSpecProperty = MIoTSpecProperty(
spec=property_,
service=spec_service,
format_=self.__spec_format2dtype(property_['format']),
format_=property_['format'],
access=property_['access'],
unit=property_.get('unit', None))
spec_prop.name = p_type_strs[3]
@ -1055,14 +1162,14 @@ class MIoTSpecParser:
or self._std_lib.value_translate(
key=f'{type_strs[:5]}|{p_type_strs[3]}|'
f'{v["description"]}')
or v['name']
)
spec_prop.value_list = v_list
or v['name'])
spec_prop.value_list = MIoTSpecValueList.from_spec(v_list)
elif property_['format'] == 'bool':
v_tag = ':'.join(p_type_strs[:5])
v_descriptions: dict = (
v_descriptions = (
await self._bool_trans.translate_async(urn=v_tag))
if v_descriptions:
# bool without value-list.name
spec_prop.value_list = v_descriptions
spec_service.properties.append(spec_prop)
# Parse service event

View File

@ -90,7 +90,7 @@ class Notify(MIoTActionEntity, NotifyEntity):
super().__init__(miot_device=miot_device, spec=spec)
self._attr_extra_state_attributes = {}
action_in: str = ', '.join([
f'{prop.description_trans}({prop.format_})'
f'{prop.description_trans}({prop.format_.__name__})'
for prop in self.spec.in_])
self._attr_extra_state_attributes['action params'] = f'[{action_in}]'
@ -122,24 +122,24 @@ class Notify(MIoTActionEntity, NotifyEntity):
return
in_value: list[dict] = []
for index, prop in enumerate(self.spec.in_):
if prop.format_ == 'str':
if prop.format_ == str:
if isinstance(in_list[index], (bool, int, float, str)):
in_value.append(
{'piid': prop.iid, 'value': str(in_list[index])})
continue
elif prop.format_ == 'bool':
elif prop.format_ == bool:
if isinstance(in_list[index], (bool, int)):
# yes, no, on, off, true, false and other bool types
# will also be parsed as 0 and 1 of int.
in_value.append(
{'piid': prop.iid, 'value': bool(in_list[index])})
continue
elif prop.format_ == 'float':
elif prop.format_ == float:
if isinstance(in_list[index], (int, float)):
in_value.append(
{'piid': prop.iid, 'value': in_list[index]})
continue
elif prop.format_ == 'int':
elif prop.format_ == int:
if isinstance(in_list[index], int):
in_value.append(
{'piid': prop.iid, 'value': in_list[index]})

View File

@ -82,7 +82,8 @@ class Select(MIoTPropertyEntity, SelectEntity):
def __init__(self, miot_device: MIoTDevice, spec: MIoTSpecProperty) -> None:
"""Initialize the Select."""
super().__init__(miot_device=miot_device, spec=spec)
self._attr_options = list(self._value_list.values())
if self._value_list:
self._attr_options = self._value_list.descriptions
async def async_select_option(self, option: str) -> None:
"""Change the selected option."""

View File

@ -91,7 +91,7 @@ class Sensor(MIoTPropertyEntity, SensorEntity):
self._attr_device_class = SensorDeviceClass.ENUM
self._attr_icon = 'mdi:message-text'
self._attr_native_unit_of_measurement = None
self._attr_options = list(self._value_list.values())
self._attr_options = self._value_list.descriptions
else:
self._attr_device_class = spec.device_class
if spec.external_unit:
@ -122,7 +122,7 @@ class Sensor(MIoTPropertyEntity, SensorEntity):
'%s, data exception, out of range, %s, %s',
self.entity_id, self._value, self._value_range)
if self._value_list:
return self._value_list.get(self._value, None)
return self.get_vlist_description(self._value)
if isinstance(self._value, str):
return self._value[:255]
return self._value

View File

@ -111,7 +111,7 @@ class ActionText(MIoTActionEntity, TextEntity):
self._attr_extra_state_attributes = {}
self._attr_native_value = ''
action_in: str = ', '.join([
f'{prop.description_trans}({prop.format_})'
f'{prop.description_trans}({prop.format_.__name__})'
for prop in self.spec.in_])
self._attr_extra_state_attributes['action params'] = f'[{action_in}]'
# For action debug
@ -141,24 +141,24 @@ class ActionText(MIoTActionEntity, TextEntity):
f'invalid action params, {value}')
in_value: list[dict] = []
for index, prop in enumerate(self.spec.in_):
if prop.format_ == 'str':
if prop.format_ == str:
if isinstance(in_list[index], (bool, int, float, str)):
in_value.append(
{'piid': prop.iid, 'value': str(in_list[index])})
continue
elif prop.format_ == 'bool':
elif prop.format_ == bool:
if isinstance(in_list[index], (bool, int)):
# yes, no, on, off, true, false and other bool types
# will also be parsed as 0 and 1 of int.
in_value.append(
{'piid': prop.iid, 'value': bool(in_list[index])})
continue
elif prop.format_ == 'float':
elif prop.format_ == float:
if isinstance(in_list[index], (int, float)):
in_value.append(
{'piid': prop.iid, 'value': in_list[index]})
continue
elif prop.format_ == 'int':
elif prop.format_ == int:
if isinstance(in_list[index], int):
in_value.append(
{'piid': prop.iid, 'value': in_list[index]})

View File

@ -120,28 +120,18 @@ class Vacuum(MIoTServiceEntity, StateVacuumEntity):
# properties
for prop in entity_data.props:
if prop.name == 'status':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'invalid status value_list, %s', self.entity_id)
continue
self._status_map = {
item['value']: item['description']
for item in prop.value_list}
self._status_map = prop.value_list.to_map()
self._prop_status = prop
elif prop.name == 'fan-level':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'invalid fan-level value_list, %s', self.entity_id)
continue
self._fan_level_map = {
item['value']: item['description']
for item in prop.value_list}
self._fan_level_map = prop.value_list.to_map()
self._attr_fan_speed_list = list(self._fan_level_map.values())
self._attr_supported_features |= VacuumEntityFeature.FAN_SPEED
self._prop_fan_level = prop
@ -202,7 +192,7 @@ class Vacuum(MIoTServiceEntity, StateVacuumEntity):
@property
def state(self) -> Optional[str]:
"""Return the current state of the vacuum cleaner."""
return self.get_map_description(
return self.get_map_value(
map_=self._status_map,
key=self.get_prop_value(prop=self._prop_status))
@ -214,6 +204,6 @@ class Vacuum(MIoTServiceEntity, StateVacuumEntity):
@property
def fan_speed(self) -> Optional[str]:
"""Return the current fan speed of the vacuum cleaner."""
return self.get_map_description(
return self.get_map_value(
map_=self._fan_level_map,
key=self.get_prop_value(prop=self._prop_fan_level))

View File

@ -93,7 +93,7 @@ class WaterHeater(MIoTServiceEntity, WaterHeaterEntity):
_prop_target_temp: Optional[MIoTSpecProperty]
_prop_mode: Optional[MIoTSpecProperty]
_mode_list: Optional[dict[Any, Any]]
_mode_map: Optional[dict[Any, Any]]
def __init__(
self, miot_device: MIoTDevice, entity_data: MIoTEntityData
@ -106,7 +106,7 @@ class WaterHeater(MIoTServiceEntity, WaterHeaterEntity):
self._prop_temp = None
self._prop_target_temp = None
self._prop_mode = None
self._mode_list = None
self._mode_map = None
# properties
for prop in entity_data.props:
@ -143,17 +143,12 @@ class WaterHeater(MIoTServiceEntity, WaterHeaterEntity):
self._prop_target_temp = prop
# mode
if prop.name == 'mode':
if (
not isinstance(prop.value_list, list)
or not prop.value_list
):
if not prop.value_list:
_LOGGER.error(
'mode value_list is None, %s', self.entity_id)
continue
self._mode_list = {
item['value']: item['description']
for item in prop.value_list}
self._attr_operation_list = list(self._mode_list.values())
self._mode_map = prop.value_list.to_map()
self._attr_operation_list = list(self._mode_map.values())
self._attr_supported_features |= (
WaterHeaterEntityFeature.OPERATION_MODE)
self._prop_mode = prop
@ -189,7 +184,9 @@ class WaterHeater(MIoTServiceEntity, WaterHeaterEntity):
prop=self._prop_on, value=True, update=False)
await self.set_property_async(
prop=self._prop_mode,
value=self.__get_mode_value(description=operation_mode))
value=self.get_map_key(
map_=self._mode_map,
value=operation_mode))
async def async_turn_away_mode_on(self) -> None:
"""Set the water heater to away mode."""
@ -212,20 +209,6 @@ class WaterHeater(MIoTServiceEntity, WaterHeaterEntity):
return STATE_OFF
if not self._prop_mode and self.get_prop_value(prop=self._prop_on):
return STATE_ON
return self.__get_mode_description(
return self.get_map_value(
map_=self._mode_map,
key=self.get_prop_value(prop=self._prop_mode))
def __get_mode_description(self, key: int) -> Optional[str]:
"""Convert mode value to description."""
if self._mode_list is None:
return None
return self._mode_list.get(key, None)
def __get_mode_value(self, description: str) -> Optional[int]:
"""Convert mode description to value."""
if self._mode_list is None:
return None
for key, value in self._mode_list.items():
if value == description:
return key
return None