Upstream/aioesphomeapi
1
0
Fork 0
mirror of https://github.com/esphome/aioesphomeapi.git synced 2024-06-26 10:35:00 +02:00
Code Issues Projects Releases Wiki Activity

Switch from attrs to dataclasses (#36)

Browse Source
This commit is contained in:
Otto Winter 2021-06-29 15:36:14 +02:00 committed by GitHub
parent 61cefdb470
commit 872c643058
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 268 additions and 277 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
aioesphomeapi/client.py
View File

@ -1,7 +1,6 @@
import asyncio import asyncio
import logging import logging
from typing import ( from typing import (
TYPE_CHECKING,
Any, Any,
Awaitable, Awaitable,
Callable, Callable,
@ -9,11 +8,11 @@ from typing import (
List, List,
Optional, Optional,
Tuple, Tuple,
Type,
Union, Union,
cast, cast,
) )
import attr
import zeroconf import zeroconf
from google.protobuf import message from google.protobuf import message
@ -79,6 +78,7 @@ from aioesphomeapi.model import (
CoverState, CoverState,
DeviceInfo, DeviceInfo,
EntityInfo, EntityInfo,
EntityState,
FanDirection, FanDirection,
FanInfo, FanInfo,
FanSpeed, FanSpeed,
@ -87,6 +87,7 @@ from aioesphomeapi.model import (
LegacyCoverCommand, LegacyCoverCommand,
LightInfo, LightInfo,
LightState, LightState,
LogLevel,
NumberInfo, NumberInfo,
NumberState, NumberState,
SensorInfo, SensorInfo,
@ -96,13 +97,9 @@ from aioesphomeapi.model import (
TextSensorInfo, TextSensorInfo,
TextSensorState, TextSensorState,
UserService, UserService,
UserServiceArg,
UserServiceArgType, UserServiceArgType,
) )
if TYPE_CHECKING:
from aioesphomeapi.api_pb2 import LogLevel # type: ignore
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
ExecuteServiceDataType = Dict[ ExecuteServiceDataType = Dict[
@ -192,21 +189,13 @@ class APIClient:
resp = await self._connection.send_message_await_response( resp = await self._connection.send_message_await_response(
DeviceInfoRequest(), DeviceInfoResponse DeviceInfoRequest(), DeviceInfoResponse
) )
return DeviceInfo( return DeviceInfo.from_pb(resp)
uses_password=resp.uses_password,
name=resp.name,
mac_address=resp.mac_address,
esphome_version=resp.esphome_version,
compilation_time=resp.compilation_time,
model=resp.model,
has_deep_sleep=resp.has_deep_sleep,
)
async def list_entities_services( async def list_entities_services(
self, self,
) -> Tuple[List[EntityInfo], List[UserService]]: ) -> Tuple[List[EntityInfo], List[UserService]]:
self._check_authenticated() self._check_authenticated()
response_types = { response_types: Dict[Any, Optional[Type[EntityInfo]]] = {
ListEntitiesBinarySensorResponse: BinarySensorInfo, ListEntitiesBinarySensorResponse: BinarySensorInfo,
ListEntitiesCoverResponse: CoverInfo, ListEntitiesCoverResponse: CoverInfo,
ListEntitiesFanResponse: FanInfo, ListEntitiesFanResponse: FanInfo,
@ -234,21 +223,7 @@ class APIClient:
services: List[UserService] = [] services: List[UserService] = []
for msg in resp: for msg in resp:
if isinstance(msg, ListEntitiesServicesResponse): if isinstance(msg, ListEntitiesServicesResponse):
args = [] services.append(UserService.from_pb(msg))
for arg in msg.args:
args.append(
UserServiceArg(
name=arg.name,
type_=arg.type,
)
)
services.append(
UserService(
name=msg.name,
key=msg.key,
args=args, # type: ignore
)
)
continue continue
cls = None cls = None
for resp_type, cls in response_types.items(): for resp_type, cls in response_types.items():
@ -256,17 +231,14 @@ class APIClient:
break break
else: else:
continue continue
cls = cast(type, cls) assert cls is not None
kwargs = {} entities.append(cls.from_pb(msg))
for key, _ in attr.fields_dict(cls).items():
kwargs[key] = getattr(msg, key)
entities.append(cls(**kwargs))
return entities, services return entities, services
async def subscribe_states(self, on_state: Callable[[Any], None]) -> None: async def subscribe_states(self, on_state: Callable[[EntityState], None]) -> None:
self._check_authenticated() self._check_authenticated()
response_types = { response_types: Dict[Any, Type[EntityState]] = {
BinarySensorStateResponse: BinarySensorState, BinarySensorStateResponse: BinarySensorState,
CoverStateResponse: CoverState, CoverStateResponse: CoverState,
FanStateResponse: FanState, FanStateResponse: FanState,
@ -284,7 +256,7 @@ class APIClient:
if isinstance(msg, CameraImageResponse): if isinstance(msg, CameraImageResponse):
data = image_stream.pop(msg.key, bytes()) + msg.data data = image_stream.pop(msg.key, bytes()) + msg.data
if msg.done: if msg.done:
on_state(CameraState(key=msg.key, image=data)) on_state(CameraState.from_pb(msg))
else: else:
image_stream[msg.key] = data image_stream[msg.key] = data
return return
@ -295,11 +267,8 @@ class APIClient:
else: else:
return return
kwargs = {}
# pylint: disable=undefined-loop-variable # pylint: disable=undefined-loop-variable
for key, _ in attr.fields_dict(cls).items(): on_state(cls.from_pb(msg))
kwargs[key] = getattr(msg, key)
on_state(cls(**kwargs))
assert self._connection is not None assert self._connection is not None
await self._connection.send_message_callback_response( await self._connection.send_message_callback_response(
@ -309,7 +278,7 @@ class APIClient:
async def subscribe_logs( async def subscribe_logs(
self, self,
on_log: Callable[[SubscribeLogsResponse], None], on_log: Callable[[SubscribeLogsResponse], None],
log_level: Optional["LogLevel"] = None, log_level: Optional[LogLevel] = None,
) -> None: ) -> None:
self._check_authenticated() self._check_authenticated()
@ -330,10 +299,7 @@ class APIClient:
def on_msg(msg: message.Message) -> None: def on_msg(msg: message.Message) -> None:
if isinstance(msg, HomeassistantServiceResponse): if isinstance(msg, HomeassistantServiceResponse):
kwargs = {} on_service_call(HomeassistantServiceCall.from_pb(msg))
for key, _ in attr.fields_dict(HomeassistantServiceCall).items():
kwargs[key] = getattr(msg, key)
on_service_call(HomeassistantServiceCall(**kwargs))
assert self._connection is not None assert self._connection is not None
await self._connection.send_message_callback_response( await self._connection.send_message_callback_response(
@ -571,12 +537,12 @@ class APIClient:
UserServiceArgType.FLOAT_ARRAY: "float_array", UserServiceArgType.FLOAT_ARRAY: "float_array",
UserServiceArgType.STRING_ARRAY: "string_array", UserServiceArgType.STRING_ARRAY: "string_array",
} }
# pylint: disable=redefined-outer-name if arg_desc.type in map_array:
if arg_desc.type_ in map_array: attr = getattr(arg, map_array[arg_desc.type])
attr = getattr(arg, map_array[arg_desc.type_])
attr.extend(val) attr.extend(val)
else: else:
setattr(arg, map_single[arg_desc.type_], val) assert arg_desc.type in map_single
setattr(arg, map_single[arg_desc.type], val)
args.append(arg) args.append(arg)
# pylint: disable=no-member # pylint: disable=no-member

18
aioesphomeapi/connection.py
View File

@ -2,9 +2,9 @@ import asyncio
import logging import logging
import socket import socket
import time import time
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, List, Optional, cast from typing import Any, Awaitable, Callable, List, Optional, cast
import attr
import zeroconf import zeroconf
from google.protobuf import message from google.protobuf import message
@ -27,15 +27,15 @@ from aioesphomeapi.util import _bytes_to_varuint, _varuint_to_bytes, resolve_ip_
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@attr.s @dataclass
class ConnectionParams: class ConnectionParams:
eventloop = attr.ib(type=asyncio.events.AbstractEventLoop) eventloop: asyncio.events.AbstractEventLoop
address = attr.ib(type=str) address: str
port = attr.ib(type=int) port: int
password = attr.ib(type=Optional[str]) password: Optional[str]
client_info = attr.ib(type=str) client_info: str
keepalive = attr.ib(type=float) keepalive: float
zeroconf_instance = attr.ib(type=Optional[zeroconf.Zeroconf]) zeroconf_instance: Optional[zeroconf.Zeroconf]
class APIConnection: class APIConnection:

453
aioesphomeapi/model.py
View File

@ -1,7 +1,18 @@
import enum import enum
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Type, TypeVar from dataclasses import asdict, dataclass, field, fields
from typing import (
import attr TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
List,
Optional,
Type,
TypeVar,
Union,
cast,
)
if TYPE_CHECKING: if TYPE_CHECKING:
from .api_pb2 import HomeassistantServiceMap # type: ignore from .api_pb2 import HomeassistantServiceMap # type: ignore
@ -13,6 +24,7 @@ if TYPE_CHECKING:
# for a field (False, 0, empty string, enum with value 0, ...) # for a field (False, 0, empty string, enum with value 0, ...)
_T = TypeVar("_T", bound="APIIntEnum") _T = TypeVar("_T", bound="APIIntEnum")
_V = TypeVar("_V")
class APIIntEnum(enum.IntEnum): class APIIntEnum(enum.IntEnum):
@ -36,56 +48,93 @@ class APIIntEnum(enum.IntEnum):
return ret return ret
@attr.s @dataclass(frozen=True)
class APIVersion: class APIModelBase:
major = attr.ib(type=int, default=0) def __post_init__(self) -> None:
minor = attr.ib(type=int, default=0) for field_ in fields(type(self)):
convert = field_.metadata.get("converter")
if convert is None:
continue
val = getattr(self, field_.name)
# use this setattr to prevent FrozenInstanceError
super().__setattr__(field_.name, convert(val))
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(
cls: Type[_V], data: Dict[str, Any], *, ignore_missing: bool = True
) -> _V:
init_args = {
f.name: data[f.name]
for f in fields(cls)
if f.name in data or (not ignore_missing)
}
return cls(**init_args) # type: ignore
@classmethod
def from_pb(cls: Type[_V], data: Any) -> _V:
init_args = {f.name: getattr(data, f.name) for f in fields(cls)}
return cls(**init_args) # type: ignore
@attr.s def converter_field(*, converter: Callable[[Any], _V], **kwargs: Any) -> _V:
class DeviceInfo: metadata = kwargs.pop("metadata", {})
uses_password = attr.ib(type=bool, default=False) metadata["converter"] = converter
name = attr.ib(type=str, default="") return cast(_V, field(metadata=metadata, **kwargs))
mac_address = attr.ib(type=str, default="")
compilation_time = attr.ib(type=str, default="")
model = attr.ib(type=str, default="")
has_deep_sleep = attr.ib(type=bool, default=False)
esphome_version = attr.ib(type=str, default="")
@attr.s @dataclass(frozen=True, order=True)
class EntityInfo: class APIVersion(APIModelBase):
object_id = attr.ib(type=str, default="") major: int = 0
key = attr.ib(type=int, default=0) minor: int = 0
name = attr.ib(type=str, default="")
unique_id = attr.ib(type=str, default="")
@attr.s @dataclass(frozen=True)
class EntityState: class DeviceInfo(APIModelBase):
key = attr.ib(type=int, default=0) uses_password: bool = False
name: str = ""
mac_address: str = ""
compilation_time: str = ""
model: str = ""
has_deep_sleep: bool = False
esphome_version: str = ""
@dataclass(frozen=True)
class EntityInfo(APIModelBase):
object_id: str = ""
key: int = 0
name: str = ""
unique_id: str = ""
@dataclass(frozen=True)
class EntityState(APIModelBase):
key: int = 0
# ==================== BINARY SENSOR ==================== # ==================== BINARY SENSOR ====================
@attr.s @dataclass(frozen=True)
class BinarySensorInfo(EntityInfo): class BinarySensorInfo(EntityInfo):
device_class = attr.ib(type=str, default="") device_class: str = ""
is_status_binary_sensor = attr.ib(type=bool, default=False) is_status_binary_sensor: bool = False
@attr.s @dataclass(frozen=True)
class BinarySensorState(EntityState): class BinarySensorState(EntityState):
state = attr.ib(type=bool, default=False) state: bool = False
missing_state = attr.ib(type=bool, default=False) missing_state: bool = False
# ==================== COVER ==================== # ==================== COVER ====================
@attr.s @dataclass(frozen=True)
class CoverInfo(EntityInfo): class CoverInfo(EntityInfo):
assumed_state = attr.ib(type=bool, default=False) assumed_state: bool = False
supports_position = attr.ib(type=bool, default=False) supports_position: bool = False
supports_tilt = attr.ib(type=bool, default=False) supports_tilt: bool = False
device_class = attr.ib(type=str, default="") device_class: str = ""
class LegacyCoverState(APIIntEnum): class LegacyCoverState(APIIntEnum):
@ -105,19 +154,15 @@ class CoverOperation(APIIntEnum):
IS_CLOSING = 2 IS_CLOSING = 2
@attr.s @dataclass(frozen=True)
class CoverState(EntityState): class CoverState(EntityState):
legacy_state = attr.ib( legacy_state: Optional[LegacyCoverState] = converter_field(
type=LegacyCoverState, default=LegacyCoverState.OPEN, converter=LegacyCoverState.convert
converter=LegacyCoverState.convert, # type: ignore
default=LegacyCoverState.OPEN,
) )
position = attr.ib(type=float, default=0.0) position: float = 0.0
tilt = attr.ib(type=float, default=0.0) tilt: float = 0.0
current_operation = attr.ib( current_operation: Optional[CoverOperation] = converter_field(
type=CoverOperation, default=CoverOperation.IDLE, converter=CoverOperation.convert
converter=CoverOperation.convert, # type: ignore
default=CoverOperation.IDLE,
) )
def is_closed(self, api_version: APIVersion) -> bool: def is_closed(self, api_version: APIVersion) -> bool:
@ -127,12 +172,12 @@ class CoverState(EntityState):
# ==================== FAN ==================== # ==================== FAN ====================
@attr.s @dataclass(frozen=True)
class FanInfo(EntityInfo): class FanInfo(EntityInfo):
supports_oscillation = attr.ib(type=bool, default=False) supports_oscillation: bool = False
supports_speed = attr.ib(type=bool, default=False) supports_speed: bool = False
supports_direction = attr.ib(type=bool, default=False) supports_direction: bool = False
supported_speed_levels = attr.ib(type=int, default=0) supported_speed_levels: int = 0
class FanSpeed(APIIntEnum): class FanSpeed(APIIntEnum):
@ -146,45 +191,41 @@ class FanDirection(APIIntEnum):
REVERSE = 1 REVERSE = 1
@attr.s @dataclass(frozen=True)
class FanState(EntityState): class FanState(EntityState):
state = attr.ib(type=bool, default=False) state: bool = False
oscillating = attr.ib(type=bool, default=False) oscillating: bool = False
speed = attr.ib( speed: Optional[FanSpeed] = converter_field(
type=Optional[FanSpeed], default=FanSpeed.LOW, converter=FanSpeed.convert
converter=FanSpeed.convert, # type: ignore
default=FanSpeed.LOW,
) )
speed_level = attr.ib(type=int, default=0) speed_level: int = 0
direction = attr.ib( direction: Optional[FanDirection] = converter_field(
type=FanDirection, default=FanDirection.FORWARD, converter=FanDirection.convert
converter=FanDirection.convert, # type: ignore
default=FanDirection.FORWARD,
) )
# ==================== LIGHT ==================== # ==================== LIGHT ====================
@attr.s @dataclass(frozen=True)
class LightInfo(EntityInfo): class LightInfo(EntityInfo):
supports_brightness = attr.ib(type=bool, default=False) supports_brightness: bool = False
supports_rgb = attr.ib(type=bool, default=False) supports_rgb: bool = False
supports_white_value = attr.ib(type=bool, default=False) supports_white_value: bool = False
supports_color_temperature = attr.ib(type=bool, default=False) supports_color_temperature: bool = False
min_mireds = attr.ib(type=float, default=0.0) min_mireds: float = 0.0
max_mireds = attr.ib(type=float, default=0.0) max_mireds: float = 0.0
effects = attr.ib(type=List[str], converter=list, factory=list) effects: List[str] = converter_field(default_factory=list, converter=list)
@attr.s @dataclass(frozen=True)
class LightState(EntityState): class LightState(EntityState):
state = attr.ib(type=bool, default=False) state: bool = False
brightness = attr.ib(type=float, default=0.0) brightness: float = 0.0
red = attr.ib(type=float, default=0.0) red: float = 0.0
green = attr.ib(type=float, default=0.0) green: float = 0.0
blue = attr.ib(type=float, default=0.0) blue: float = 0.0
white = attr.ib(type=float, default=0.0) white: float = 0.0
color_temperature = attr.ib(type=float, default=0.0) color_temperature: float = 0.0
effect = attr.ib(type=str, default="") effect: str = ""
# ==================== SENSOR ==================== # ==================== SENSOR ====================
@ -193,59 +234,57 @@ class SensorStateClass(APIIntEnum):
MEASUREMENT = 1 MEASUREMENT = 1
@attr.s @dataclass(frozen=True)
class SensorInfo(EntityInfo): class SensorInfo(EntityInfo):
icon = attr.ib(type=str, default="") icon: str = ""
device_class = attr.ib(type=str, default="") device_class: str = ""
unit_of_measurement = attr.ib(type=str, default="") unit_of_measurement: str = ""
accuracy_decimals = attr.ib(type=int, default=0) accuracy_decimals: int = 0
force_update = attr.ib(type=bool, default=False) force_update: bool = False
state_class = attr.ib( state_class: Optional[SensorStateClass] = converter_field(
type=SensorStateClass, default=SensorStateClass.NONE, converter=SensorStateClass.convert
converter=SensorStateClass.convert, # type: ignore
default=SensorStateClass.NONE,
) )
@attr.s @dataclass(frozen=True)
class SensorState(EntityState): class SensorState(EntityState):
state = attr.ib(type=float, default=0.0) state: float = 0.0
missing_state = attr.ib(type=bool, default=False) missing_state: bool = False
# ==================== SWITCH ==================== # ==================== SWITCH ====================
@attr.s @dataclass(frozen=True)
class SwitchInfo(EntityInfo): class SwitchInfo(EntityInfo):
icon = attr.ib(type=str, default="") icon: str = ""
assumed_state = attr.ib(type=bool, default=False) assumed_state: bool = False
@attr.s @dataclass(frozen=True)
class SwitchState(EntityState): class SwitchState(EntityState):
state = attr.ib(type=bool, default=False) state: bool = False
# ==================== TEXT SENSOR ==================== # ==================== TEXT SENSOR ====================
@attr.s @dataclass(frozen=True)
class TextSensorInfo(EntityInfo): class TextSensorInfo(EntityInfo):
icon = attr.ib(type=str, default="") icon: str = ""
@attr.s @dataclass(frozen=True)
class TextSensorState(EntityState): class TextSensorState(EntityState):
state = attr.ib(type=str, default="") state: str = ""
missing_state = attr.ib(type=bool, default=False) missing_state: bool = False
# ==================== CAMERA ==================== # ==================== CAMERA ====================
@attr.s @dataclass(frozen=True)
class CameraInfo(EntityInfo): class CameraInfo(EntityInfo):
pass pass
@attr.s @dataclass(frozen=True)
class CameraState(EntityState): class CameraState(EntityState):
image = attr.ib(type=bytes, factory=bytes) image: bytes = field(default_factory=bytes)
# ==================== CLIMATE ==================== # ==================== CLIMATE ====================
@ -298,35 +337,33 @@ class ClimatePreset(APIIntEnum):
ACTIVITY = 7 ACTIVITY = 7
@attr.s @dataclass(frozen=True)
class ClimateInfo(EntityInfo): class ClimateInfo(EntityInfo):
supports_current_temperature = attr.ib(type=bool, default=False) supports_current_temperature: bool = False
supports_two_point_target_temperature = attr.ib(type=bool, default=False) supports_two_point_target_temperature: bool = False
supported_modes = attr.ib( supported_modes: List[ClimateMode] = converter_field(
type=List[ClimateMode], default_factory=list, converter=ClimateMode.convert_list
converter=ClimateMode.convert_list, # type: ignore
factory=list,
) )
visual_min_temperature = attr.ib(type=float, default=0.0) visual_min_temperature: float = 0.0
visual_max_temperature = attr.ib(type=float, default=0.0) visual_max_temperature: float = 0.0
visual_temperature_step = attr.ib(type=float, default=0.0) visual_temperature_step: float = 0.0
legacy_supports_away = attr.ib(type=bool, default=False) legacy_supports_away: bool = False
supports_action = attr.ib(type=bool, default=False) supports_action: bool = False
supported_fan_modes = attr.ib( supported_fan_modes: List[ClimateFanMode] = converter_field(
type=List[ClimateFanMode], default_factory=list, converter=ClimateFanMode.convert_list
converter=ClimateFanMode.convert_list, # type: ignore
factory=list,
) )
supported_swing_modes = attr.ib( supported_swing_modes: List[ClimateSwingMode] = converter_field(
type=List[ClimateSwingMode], default_factory=list, converter=ClimateSwingMode.convert_list
converter=ClimateSwingMode.convert_list, # type: ignore
factory=list,
) )
supported_custom_fan_modes = attr.ib(type=List[str], converter=list, factory=list) supported_custom_fan_modes: List[str] = converter_field(
supported_presets = attr.ib( default_factory=list, converter=list
type=List[ClimatePreset], converter=ClimatePreset.convert_list, factory=list # type: ignore )
supported_presets: List[ClimatePreset] = converter_field(
default_factory=list, converter=ClimatePreset.convert_list
)
supported_custom_presets: List[str] = converter_field(
default_factory=list, converter=list
) )
supported_custom_presets = attr.ib(type=List[str], converter=list, factory=list)
def supported_presets_compat(self, api_version: APIVersion) -> List[ClimatePreset]: def supported_presets_compat(self, api_version: APIVersion) -> List[ClimatePreset]:
if api_version < APIVersion(1, 5): if api_version < APIVersion(1, 5):
@ -338,40 +375,30 @@ class ClimateInfo(EntityInfo):
return self.supported_presets return self.supported_presets
@attr.s @dataclass(frozen=True)
class ClimateState(EntityState): class ClimateState(EntityState):
mode = attr.ib( mode: Optional[ClimateMode] = converter_field(
type=ClimateMode, default=ClimateMode.OFF, converter=ClimateMode.convert
converter=ClimateMode.convert, # type: ignore
default=ClimateMode.OFF,
) )
action = attr.ib( action: Optional[ClimateAction] = converter_field(
type=ClimateAction, default=ClimateAction.OFF, converter=ClimateAction.convert
converter=ClimateAction.convert, # type: ignore
default=ClimateAction.OFF,
) )
current_temperature = attr.ib(type=float, default=0.0) current_temperature: float = 0.0
target_temperature = attr.ib(type=float, default=0.0) target_temperature: float = 0.0
target_temperature_low = attr.ib(type=float, default=0.0) target_temperature_low: float = 0.0
target_temperature_high = attr.ib(type=float, default=0.0) target_temperature_high: float = 0.0
legacy_away = attr.ib(type=bool, default=False) legacy_away: bool = False
fan_mode = attr.ib( fan_mode: Optional[ClimateFanMode] = converter_field(
type=Optional[ClimateFanMode], default=ClimateFanMode.ON, converter=ClimateFanMode.convert
converter=ClimateFanMode.convert, # type: ignore
default=ClimateFanMode.ON,
) )
swing_mode = attr.ib( swing_mode: Optional[ClimateSwingMode] = converter_field(
type=Optional[ClimateSwingMode], default=ClimateSwingMode.OFF, converter=ClimateSwingMode.convert
converter=ClimateSwingMode.convert, # type: ignore
default=ClimateSwingMode.OFF,
) )
custom_fan_mode = attr.ib(type=str, default="") custom_fan_mode: str = ""
preset = attr.ib( preset: Optional[ClimatePreset] = converter_field(
type=Optional[ClimatePreset], default=ClimatePreset.HOME, converter=ClimatePreset.convert
converter=ClimatePreset.convert, # type: ignore
default=ClimatePreset.HOME,
) )
custom_preset = attr.ib(type=str, default="") custom_preset: str = ""
def preset_compat(self, api_version: APIVersion) -> Optional[ClimatePreset]: def preset_compat(self, api_version: APIVersion) -> Optional[ClimatePreset]:
if api_version < APIVersion(1, 5): if api_version < APIVersion(1, 5):
@ -380,18 +407,18 @@ class ClimateState(EntityState):
# ==================== NUMBER ==================== # ==================== NUMBER ====================
@attr.s @dataclass(frozen=True)
class NumberInfo(EntityInfo): class NumberInfo(EntityInfo):
icon = attr.ib(type=str, default="") icon: str = ""
min_value = attr.ib(type=float, default=0.0) min_value: float = 0.0
max_value = attr.ib(type=float, default=0.0) max_value: float = 0.0
step = attr.ib(type=float, default=0.0) step: float = 0.0
@attr.s @dataclass(frozen=True)
class NumberState(EntityState): class NumberState(EntityState):
state = attr.ib(type=float, default=0.0) state: float = 0.0
missing_state = attr.ib(type=bool, default=False) missing_state: bool = False
COMPONENT_TYPE_TO_INFO = { COMPONENT_TYPE_TO_INFO = {
@ -410,23 +437,26 @@ COMPONENT_TYPE_TO_INFO = {
# ==================== USER-DEFINED SERVICES ==================== # ==================== USER-DEFINED SERVICES ====================
def _convert_homeassistant_service_map( def _convert_homeassistant_service_map(
value: Iterable["HomeassistantServiceMap"], value: Union[Dict[str, str], Iterable["HomeassistantServiceMap"]],
) -> Dict[str, str]: ) -> Dict[str, str]:
return {v.key: v.value for v in value} if isinstance(value, dict):
# already a dict, don't convert
return value
return {v.key: v.value for v in value} # type: ignore
@attr.s @dataclass(frozen=True)
class HomeassistantServiceCall: class HomeassistantServiceCall(APIModelBase):
service = attr.ib(type=str, default="") service: str = ""
is_event = attr.ib(type=bool, default=False) is_event: bool = False
data = attr.ib( data: Dict[str, str] = converter_field(
type=Dict[str, str], converter=_convert_homeassistant_service_map, factory=dict default_factory=dict, converter=_convert_homeassistant_service_map
) )
data_template = attr.ib( data_template: Dict[str, str] = converter_field(
type=Dict[str, str], converter=_convert_homeassistant_service_map, factory=dict default_factory=dict, converter=_convert_homeassistant_service_map
) )
variables = attr.ib( variables: Dict[str, str] = converter_field(
type=Dict[str, str], converter=_convert_homeassistant_service_map, factory=dict default_factory=dict, converter=_convert_homeassistant_service_map
) )
@ -441,43 +471,38 @@ class UserServiceArgType(APIIntEnum):
STRING_ARRAY = 7 STRING_ARRAY = 7
_K = TypeVar("_K") @dataclass(frozen=True)
class UserServiceArg(APIModelBase):
name: str = ""
type: Optional[UserServiceArgType] = converter_field(
default=UserServiceArgType.BOOL, converter=UserServiceArgType.convert
)
@classmethod
def convert_list(cls, value: List[Any]) -> List["UserServiceArg"]:
ret = []
for x in value:
if isinstance(x, dict):
ret.append(UserServiceArg(x["name"], x["type"]))
else:
ret.append(UserServiceArg(x.name, x.type))
return ret
def _attr_obj_from_dict(cls: Type[_K], **kwargs: Any) -> _K: @dataclass(frozen=True)
return cls(**{key: kwargs[key] for key in attr.fields_dict(cls)}) # type: ignore class UserService(APIModelBase):
name: str = ""
key: int = 0
@attr.s args: List[UserServiceArg] = converter_field(
class UserServiceArg: default_factory=list, converter=UserServiceArg.convert_list
name = attr.ib(type=str, default="")
type_ = attr.ib(
type=UserServiceArgType,
converter=UserServiceArgType.convert, # type: ignore
default=UserServiceArgType.BOOL,
) )
@attr.s class LogLevel(APIIntEnum):
class UserService: LOG_LEVEL_NONE = 0
name = attr.ib(type=str, default="") LOG_LEVEL_ERROR = 1
key = attr.ib(type=int, default=0) LOG_LEVEL_WARN = 2
args = attr.ib(type=List[UserServiceArg], converter=list, factory=list) LOG_LEVEL_INFO = 3
LOG_LEVEL_DEBUG = 4
@classmethod LOG_LEVEL_VERBOSE = 5
def from_dict(cls, dict_: Dict[str, Any]) -> "UserService": LOG_LEVEL_VERY_VERBOSE = 6
args = []
for arg in dict_.get("args", []):
args.append(_attr_obj_from_dict(UserServiceArg, **arg))
return cls(
name=dict_.get("name", ""),
key=dict_.get("key", 0),
args=args, # type: ignore
)
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"key": self.key,
"args": [attr.asdict(arg) for arg in self.args],
}

1
pylintrc
View File

@ -14,4 +14,5 @@ disable=
unused-wildcard-import, unused-wildcard-import,
import-outside-toplevel, import-outside-toplevel,
raise-missing-from, raise-missing-from,
bad-mcs-classmethod-argument,
duplicate-code, duplicate-code,

1
requirements.txt
View File

@ -1,3 +1,2 @@
attrs>=19.3.0
protobuf>=3.12.2,<4.0 protobuf>=3.12.2,<4.0
zeroconf>=0.28.0,<1.0 zeroconf>=0.28.0,<1.0

2
setup.py
View File

@ -46,5 +46,5 @@ setup(
include_package_data=True, include_package_data=True,
zip_safe=False, zip_safe=False,
install_requires=REQUIRES, install_requires=REQUIRES,
python_requires='>=3.5.3', python_requires='>=3.7',
) )