From cc63ca8f6cea84554060f3c96b9a32d4beb06250 Mon Sep 17 00:00:00 2001 From: Luke Date: Fri, 8 Sep 2023 23:04:36 -0400 Subject: [PATCH] chore: add pyupgrade to ruff --- pyproject.toml | 1 + roborock/api.py | 19 +- roborock/cli.py | 8 +- roborock/cloud_api.py | 4 +- roborock/code_mappings.py | 13 +- roborock/command_cache.py | 8 +- roborock/containers.py | 412 +++++++++++++++++------------------ roborock/local_api.py | 9 +- roborock/protocol.py | 6 +- roborock/roborock_message.py | 5 +- roborock/roborock_typing.py | 19 +- roborock/util.py | 15 +- 12 files changed, 259 insertions(+), 260 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index f12b4e5..3eb9640 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,7 @@ build_command = "pip install poetry && poetry build" [tool.ruff] ignore = ["F403", "E741"] line-length = 120 +select=["E", "F", "UP"] [tool.black] line-length = 120 diff --git a/roborock/api.py b/roborock/api.py index de5aa1c..95ded74 100644 --- a/roborock/api.py +++ b/roborock/api.py @@ -13,7 +13,8 @@ import struct import time from random import randint -from typing import Any, Callable, Coroutine, Optional, Type, TypeVar, final +from typing import Any, TypeVar, final +from collections.abc import Callable, Coroutine import aiohttp @@ -89,7 +90,7 @@ def md5hex(message: str) -> str: class PreparedRequest: - def __init__(self, base_url: str, base_headers: Optional[dict] = None) -> None: + def __init__(self, base_url: str, base_headers: dict | None = None) -> None: self.base_url = base_url self.base_headers = base_headers or {} @@ -251,7 +252,7 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None: data_protocol = RoborockDataProtocol(int(data_point_number)) self._logger.debug(f"Got device update for {data_protocol.name}: {data_point}") if data_protocol in ROBOROCK_DATA_STATUS_PROTOCOL: - _cls: Type[Status] = ModelStatus.get( + _cls: type[Status] = ModelStatus.get( self.device_info.model, S7MaxVStatus ) # Default to S7 MAXV if we don't have the data if self.cache[CacheableAttribute.status].value is None: @@ -301,7 +302,7 @@ def on_message_received(self, messages: list[RoborockMessage]) -> None: except Exception as ex: self._logger.exception(ex) - def on_connection_lost(self, exc: Optional[Exception]) -> None: + def on_connection_lost(self, exc: Exception | None) -> None: self._last_disconnection = self.time_func() self._logger.info("Roborock client disconnected") if exc is not None: @@ -340,7 +341,7 @@ def _async_response( def _get_payload( self, method: RoborockCommand, - params: Optional[list | dict] = None, + params: list | dict | None = None, secured=False, ): timestamp = math.floor(time.time()) @@ -372,7 +373,7 @@ async def send_message(self, roborock_message: RoborockMessage): async def _send_command( self, method: RoborockCommand, - params: Optional[list | dict] = None, + params: list | dict | None = None, ): raise NotImplementedError @@ -380,8 +381,8 @@ async def _send_command( async def send_command( self, method: RoborockCommand, - params: Optional[list | dict] = None, - return_type: Optional[Type[RT]] = None, + params: list | dict | None = None, + return_type: type[RT] | None = None, ) -> RT: cacheable_attribute_result = find_cacheable_attribute(method) @@ -404,7 +405,7 @@ async def send_command( return response async def get_status(self) -> Status | None: - _cls: Type[Status] = ModelStatus.get( + _cls: type[Status] = ModelStatus.get( self.device_info.model, S7MaxVStatus ) # Default to S7 MAXV if we don't have the data return _cls.from_dict(await self.cache[CacheableAttribute.status].async_value()) diff --git a/roborock/cli.py b/roborock/cli.py index cbf9511..d9e9672 100644 --- a/roborock/cli.py +++ b/roborock/cli.py @@ -3,7 +3,7 @@ import json import logging from pathlib import Path -from typing import Any, Dict +from typing import Any import click from pyshark import FileCapture # type: ignore @@ -29,7 +29,7 @@ def __init__(self): def reload(self): if self.roborock_file.is_file(): - with open(self.roborock_file, "r") as f: + with open(self.roborock_file) as f: data = json.load(f) if data: self._login_data = LoginData.from_dict(data) @@ -54,7 +54,7 @@ def login_data(self): @click.group() @click.pass_context def cli(ctx, debug: int): - logging_config: Dict[str, Any] = {"level": logging.DEBUG if debug > 0 else logging.INFO} + logging_config: dict[str, Any] = {"level": logging.DEBUG if debug > 0 else logging.INFO} logging.basicConfig(**logging_config) # type: ignore ctx.obj = RoborockContext() @@ -153,7 +153,7 @@ async def parser(_, local_key, device_ip, file): else: _LOGGER.info("Listen for interface rvi0 since no file was provided") capture = LiveCapture(interface="rvi0") - buffer = {"data": bytes()} + buffer = {"data": b""} def on_package(packet: Packet): if hasattr(packet, "ip"): diff --git a/roborock/cloud_api.py b/roborock/cloud_api.py index 7b1d001..fbcc5ff 100644 --- a/roborock/cloud_api.py +++ b/roborock/cloud_api.py @@ -6,7 +6,7 @@ import threading import uuid from asyncio import Lock, Task -from typing import Any, Optional +from typing import Any from urllib.parse import urlparse import paho.mqtt.client as mqtt @@ -199,7 +199,7 @@ async def send_message(self, roborock_message: RoborockMessage): async def _send_command( self, method: RoborockCommand, - params: Optional[list | dict] = None, + params: list | dict | None = None, ): request_id, timestamp, payload = super()._get_payload(method, params, True) request_protocol = RoborockMessageProtocol.RPC_REQUEST diff --git a/roborock/code_mappings.py b/roborock/code_mappings.py index a33b643..75731e5 100644 --- a/roborock/code_mappings.py +++ b/roborock/code_mappings.py @@ -2,7 +2,6 @@ import logging from enum import IntEnum -from typing import Type _LOGGER = logging.getLogger(__name__) @@ -15,7 +14,7 @@ def name(self) -> str: return super().name.lower() @classmethod - def _missing_(cls: Type[RoborockEnum], key) -> RoborockEnum: + def _missing_(cls: type[RoborockEnum], key) -> RoborockEnum: if hasattr(cls, "unknown"): _LOGGER.warning(f"Missing {cls.__name__} code: {key} - defaulting to 'unknown'") return cls.unknown # type: ignore @@ -24,23 +23,23 @@ def _missing_(cls: Type[RoborockEnum], key) -> RoborockEnum: return default_value @classmethod - def as_dict(cls: Type[RoborockEnum]): + def as_dict(cls: type[RoborockEnum]): return {i.name: i.value for i in cls if i.name != "missing"} @classmethod - def as_enum_dict(cls: Type[RoborockEnum]): + def as_enum_dict(cls: type[RoborockEnum]): return {i.value: i for i in cls if i.name != "missing"} @classmethod - def values(cls: Type[RoborockEnum]) -> list[int]: + def values(cls: type[RoborockEnum]) -> list[int]: return list(cls.as_dict().values()) @classmethod - def keys(cls: Type[RoborockEnum]) -> list[str]: + def keys(cls: type[RoborockEnum]) -> list[str]: return list(cls.as_dict().keys()) @classmethod - def items(cls: Type[RoborockEnum]): + def items(cls: type[RoborockEnum]): return cls.as_dict().items() diff --git a/roborock/command_cache.py b/roborock/command_cache.py index 7d0a834..17f8ab2 100644 --- a/roborock/command_cache.py +++ b/roborock/command_cache.py @@ -2,7 +2,7 @@ from dataclasses import dataclass, field from enum import Enum -from typing import Mapping, Optional +from collections.abc import Mapping from roborock import RoborockCommand @@ -38,9 +38,9 @@ class CacheableAttribute(str, Enum): class RoborockAttribute: attribute: str get_command: RoborockCommand - add_command: Optional[RoborockCommand] = None - set_command: Optional[RoborockCommand] = None - close_command: Optional[RoborockCommand] = None + add_command: RoborockCommand | None = None + set_command: RoborockCommand | None = None + close_command: RoborockCommand | None = None additional_change_commands: list[RoborockCommand] = field(default_factory=list) diff --git a/roborock/containers.py b/roborock/containers.py index 67b9233..327f5ec 100644 --- a/roborock/containers.py +++ b/roborock/containers.py @@ -5,7 +5,7 @@ import re from dataclasses import asdict, dataclass from enum import Enum -from typing import Any, NamedTuple, Optional, Type +from typing import Any, NamedTuple from dacite import Config, from_dict @@ -67,10 +67,10 @@ def decamelize_obj(d: dict | list, ignore_keys: list[str]): if isinstance(d, RoborockBase): d = d.as_dict() if isinstance(d, list): - return [decamelize_obj(i, ignore_keys) if isinstance(i, (dict, list)) else i for i in d] + return [decamelize_obj(i, ignore_keys) if isinstance(i, dict | list) else i for i in d] return { (decamelize(a) if a not in ignore_keys else a): decamelize_obj(b, ignore_keys) - if isinstance(b, (dict, list)) + if isinstance(b, dict | list) else b for a, b in d.items() } @@ -100,13 +100,13 @@ def as_dict(self) -> dict: @dataclass class RoborockBaseTimer(RoborockBase): - start_hour: Optional[int] = None - start_minute: Optional[int] = None - end_hour: Optional[int] = None - end_minute: Optional[int] = None - enabled: Optional[int] = None - start_time: Optional[datetime.time] = None - end_time: Optional[datetime.time] = None + start_hour: int | None = None + start_minute: int | None = None + end_hour: int | None = None + end_minute: int | None = None + enabled: int | None = None + start_time: datetime.time | None = None + end_time: datetime.time | None = None def __post_init__(self) -> None: self.start_time = ( @@ -123,10 +123,10 @@ def __post_init__(self) -> None: @dataclass class Reference(RoborockBase): - r: Optional[str] = None - a: Optional[str] = None - m: Optional[str] = None - l: Optional[str] = None + r: str | None = None + a: str | None = None + m: str | None = None + l: str | None = None @dataclass @@ -140,41 +140,41 @@ class RRiot(RoborockBase): @dataclass class UserData(RoborockBase): - uid: Optional[int] = None - tokentype: Optional[str] = None - token: Optional[str] = None - rruid: Optional[str] = None - region: Optional[str] = None - countrycode: Optional[str] = None - country: Optional[str] = None - nickname: Optional[str] = None - rriot: Optional[RRiot] = None - tuya_device_state: Optional[int] = None - avatarurl: Optional[str] = None + uid: int | None = None + tokentype: str | None = None + token: str | None = None + rruid: str | None = None + region: str | None = None + countrycode: str | None = None + country: str | None = None + nickname: str | None = None + rriot: RRiot | None = None + tuya_device_state: int | None = None + avatarurl: str | None = None @dataclass class HomeDataProductSchema(RoborockBase): - id: Optional[Any] = None - name: Optional[Any] = None - code: Optional[Any] = None - mode: Optional[Any] = None - type: Optional[Any] = None - product_property: Optional[Any] = None - desc: Optional[Any] = None + id: Any | None = None + name: Any | None = None + code: Any | None = None + mode: Any | None = None + type: Any | None = None + product_property: Any | None = None + desc: Any | None = None @dataclass class HomeDataProduct(RoborockBase): - id: Optional[str] = None - name: Optional[str] = None - code: Optional[str] = None - model: Optional[str] = None - iconurl: Optional[str] = None - attribute: Optional[Any] = None - capability: Optional[int] = None - category: Optional[str] = None - schema: Optional[list[HomeDataProductSchema]] = None + id: str | None = None + name: str | None = None + code: str | None = None + model: str | None = None + iconurl: str | None = None + attribute: Any | None = None + capability: int | None = None + category: str | None = None + schema: list[HomeDataProductSchema] | None = None @dataclass @@ -183,46 +183,46 @@ class HomeDataDevice(RoborockBase): name: str local_key: str fv: str - attribute: Optional[Any] = None - active_time: Optional[int] = None - runtime_env: Optional[Any] = None - time_zone_id: Optional[str] = None - icon_url: Optional[str] = None - product_id: Optional[str] = None - lon: Optional[Any] = None - lat: Optional[Any] = None - share: Optional[Any] = None - share_time: Optional[Any] = None - online: Optional[bool] = None - pv: Optional[str] = None - room_id: Optional[Any] = None - tuya_uuid: Optional[Any] = None - tuya_migrated: Optional[bool] = None - extra: Optional[Any] = None - sn: Optional[str] = None - feature_set: Optional[str] = None - new_feature_set: Optional[str] = None - device_status: Optional[dict] = None - silent_ota_switch: Optional[bool] = None + attribute: Any | None = None + active_time: int | None = None + runtime_env: Any | None = None + time_zone_id: str | None = None + icon_url: str | None = None + product_id: str | None = None + lon: Any | None = None + lat: Any | None = None + share: Any | None = None + share_time: Any | None = None + online: bool | None = None + pv: str | None = None + room_id: Any | None = None + tuya_uuid: Any | None = None + tuya_migrated: bool | None = None + extra: Any | None = None + sn: str | None = None + feature_set: str | None = None + new_feature_set: str | None = None + device_status: dict | None = None + silent_ota_switch: bool | None = None @dataclass class HomeDataRoom(RoborockBase): - id: Optional[Any] = None - name: Optional[Any] = None + id: Any | None = None + name: Any | None = None @dataclass class HomeData(RoborockBase): - id: Optional[int] = None - name: Optional[str] = None - lon: Optional[Any] = None - lat: Optional[Any] = None - geo_name: Optional[Any] = None - products: Optional[list[HomeDataProduct]] = None - devices: Optional[list[HomeDataDevice]] = None - received_devices: Optional[list[HomeDataDevice]] = None - rooms: Optional[list[HomeDataRoom]] = None + id: int | None = None + name: str | None = None + lon: Any | None = None + lat: Any | None = None + geo_name: Any | None = None + products: list[HomeDataProduct] | None = None + devices: list[HomeDataDevice] | None = None + received_devices: list[HomeDataDevice] | None = None + rooms: list[HomeDataRoom] | None = None def get_all_devices(self) -> list[HomeDataDevice]: devices = [] @@ -237,64 +237,64 @@ def get_all_devices(self) -> list[HomeDataDevice]: class LoginData(RoborockBase): user_data: UserData email: str - home_data: Optional[HomeData] = None + home_data: HomeData | None = None @dataclass class Status(RoborockBase): - msg_ver: Optional[int] = None - msg_seq: Optional[int] = None - state: Optional[RoborockStateCode] = None - battery: Optional[int] = None - clean_time: Optional[int] = None - clean_area: Optional[int] = None - square_meter_clean_area: Optional[float] = None - error_code: Optional[RoborockErrorCode] = None - map_present: Optional[int] = None - in_cleaning: Optional[int] = None - in_returning: Optional[int] = None - in_fresh_state: Optional[int] = None - lab_status: Optional[int] = None - water_box_status: Optional[int] = None - back_type: Optional[int] = None - wash_phase: Optional[int] = None - wash_ready: Optional[int] = None - fan_power: Optional[RoborockFanPowerCode] = None - dnd_enabled: Optional[int] = None - map_status: Optional[int] = None - is_locating: Optional[int] = None - lock_status: Optional[int] = None - water_box_mode: Optional[RoborockMopIntensityCode] = None - water_box_carriage_status: Optional[int] = None - mop_forbidden_enable: Optional[int] = None - camera_status: Optional[int] = None - is_exploring: Optional[int] = None - home_sec_status: Optional[int] = None - home_sec_enable_password: Optional[int] = None - adbumper_status: Optional[list[int]] = None - water_shortage_status: Optional[int] = None - dock_type: Optional[RoborockDockTypeCode] = None - dust_collection_status: Optional[int] = None - auto_dust_collection: Optional[int] = None - avoid_count: Optional[int] = None - mop_mode: Optional[RoborockMopModeCode] = None - debug_mode: Optional[int] = None - collision_avoid_status: Optional[int] = None - switch_map_mode: Optional[int] = None - dock_error_status: Optional[RoborockDockErrorCode] = None - charge_status: Optional[int] = None - unsave_map_reason: Optional[int] = None - unsave_map_flag: Optional[int] = None - wash_status: Optional[int] = None - distance_off: Optional[int] = None - in_warmup: Optional[int] = None - dry_status: Optional[int] = None - rdt: Optional[int] = None - clean_percent: Optional[int] = None - rss: Optional[int] = None - dss: Optional[int] = None - common_status: Optional[int] = None - corner_clean_mode: Optional[int] = None + msg_ver: int | None = None + msg_seq: int | None = None + state: RoborockStateCode | None = None + battery: int | None = None + clean_time: int | None = None + clean_area: int | None = None + square_meter_clean_area: float | None = None + error_code: RoborockErrorCode | None = None + map_present: int | None = None + in_cleaning: int | None = None + in_returning: int | None = None + in_fresh_state: int | None = None + lab_status: int | None = None + water_box_status: int | None = None + back_type: int | None = None + wash_phase: int | None = None + wash_ready: int | None = None + fan_power: RoborockFanPowerCode | None = None + dnd_enabled: int | None = None + map_status: int | None = None + is_locating: int | None = None + lock_status: int | None = None + water_box_mode: RoborockMopIntensityCode | None = None + water_box_carriage_status: int | None = None + mop_forbidden_enable: int | None = None + camera_status: int | None = None + is_exploring: int | None = None + home_sec_status: int | None = None + home_sec_enable_password: int | None = None + adbumper_status: list[int] | None = None + water_shortage_status: int | None = None + dock_type: RoborockDockTypeCode | None = None + dust_collection_status: int | None = None + auto_dust_collection: int | None = None + avoid_count: int | None = None + mop_mode: RoborockMopModeCode | None = None + debug_mode: int | None = None + collision_avoid_status: int | None = None + switch_map_mode: int | None = None + dock_error_status: RoborockDockErrorCode | None = None + charge_status: int | None = None + unsave_map_reason: int | None = None + unsave_map_flag: int | None = None + wash_status: int | None = None + distance_off: int | None = None + in_warmup: int | None = None + dry_status: int | None = None + rdt: int | None = None + clean_percent: int | None = None + rss: int | None = None + dss: int | None = None + common_status: int | None = None + corner_clean_mode: int | None = None def __post_init__(self) -> None: self.square_meter_clean_area = round(self.clean_area / 1000000, 1) if self.clean_area is not None else None @@ -302,70 +302,70 @@ def __post_init__(self) -> None: @dataclass class S4MaxStatus(Status): - fan_power: Optional[RoborockFanSpeedS6Pure] = None - water_box_mode: Optional[RoborockMopIntensityS7] = None - mop_mode: Optional[RoborockMopModeS7] = None + fan_power: RoborockFanSpeedS6Pure | None = None + water_box_mode: RoborockMopIntensityS7 | None = None + mop_mode: RoborockMopModeS7 | None = None @dataclass class S5MaxStatus(Status): - fan_power: Optional[RoborockFanSpeedS6Pure] = None - water_box_mode: Optional[RoborockMopIntensityS5Max] = None + fan_power: RoborockFanSpeedS6Pure | None = None + water_box_mode: RoborockMopIntensityS5Max | None = None @dataclass class Q7MaxStatus(Status): - fan_power: Optional[RoborockFanSpeedQ7Max] = None - water_box_mode: Optional[RoborockMopIntensityV2] = None + fan_power: RoborockFanSpeedQ7Max | None = None + water_box_mode: RoborockMopIntensityV2 | None = None @dataclass class S6MaxVStatus(Status): - fan_power: Optional[RoborockFanSpeedS7MaxV] = None - water_box_mode: Optional[RoborockMopIntensityS7] = None + fan_power: RoborockFanSpeedS7MaxV | None = None + water_box_mode: RoborockMopIntensityS7 | None = None @dataclass class S6PureStatus(Status): - fan_power: Optional[RoborockFanSpeedS6Pure] = None + fan_power: RoborockFanSpeedS6Pure | None = None @dataclass class S7MaxVStatus(Status): - fan_power: Optional[RoborockFanSpeedS7MaxV] = None - water_box_mode: Optional[RoborockMopIntensityS7] = None - mop_mode: Optional[RoborockMopModeS7] = None + fan_power: RoborockFanSpeedS7MaxV | None = None + water_box_mode: RoborockMopIntensityS7 | None = None + mop_mode: RoborockMopModeS7 | None = None @dataclass class S7Status(Status): - fan_power: Optional[RoborockFanSpeedS7] = None - water_box_mode: Optional[RoborockMopIntensityS7] = None - mop_mode: Optional[RoborockMopModeS7] = None + fan_power: RoborockFanSpeedS7 | None = None + water_box_mode: RoborockMopIntensityS7 | None = None + mop_mode: RoborockMopModeS7 | None = None @dataclass class S8ProUltraStatus(Status): - fan_power: Optional[RoborockFanSpeedS7MaxV] = None - water_box_mode: Optional[RoborockMopIntensityS7] = None - mop_mode: Optional[RoborockMopModeS8ProUltra] = None + fan_power: RoborockFanSpeedS7MaxV | None = None + water_box_mode: RoborockMopIntensityS7 | None = None + mop_mode: RoborockMopModeS8ProUltra | None = None @dataclass class S8Status(Status): - fan_power: Optional[RoborockFanSpeedS7MaxV] = None - water_box_mode: Optional[RoborockMopIntensityS7] = None - mop_mode: Optional[RoborockMopModeS8ProUltra] = None + fan_power: RoborockFanSpeedS7MaxV | None = None + water_box_mode: RoborockMopIntensityS7 | None = None + mop_mode: RoborockMopModeS8ProUltra | None = None @dataclass class P10Status(Status): - fan_power: Optional[RoborockFanSpeedP10] = None - water_box_mode: Optional[RoborockMopIntensityV2] = None - mop_mode: Optional[RoborockMopModeS8ProUltra] = None + fan_power: RoborockFanSpeedP10 | None = None + water_box_mode: RoborockMopIntensityV2 | None = None + mop_mode: RoborockMopModeS8ProUltra | None = None -ModelStatus: dict[str, Type[Status]] = { +ModelStatus: dict[str, type[Status]] = { ROBOROCK_S4_MAX: S4MaxStatus, ROBOROCK_S5_MAX: S5MaxStatus, ROBOROCK_Q7_MAX: Q7MaxStatus, @@ -393,13 +393,13 @@ class ValleyElectricityTimer(RoborockBaseTimer): @dataclass class CleanSummary(RoborockBase): - clean_time: Optional[int] = None - clean_area: Optional[int] = None - square_meter_clean_area: Optional[float] = None - clean_count: Optional[int] = None - dust_collection_count: Optional[int] = None - records: Optional[list[int]] = None - last_clean_t: Optional[int] = None + clean_time: int | None = None + clean_area: int | None = None + square_meter_clean_area: float | None = None + clean_count: int | None = None + dust_collection_count: int | None = None + records: list[int] | None = None + last_clean_t: int | None = None def __post_init__(self) -> None: self.square_meter_clean_area = round(self.clean_area / 1000000, 1) if self.clean_area is not None else None @@ -407,20 +407,20 @@ def __post_init__(self) -> None: @dataclass class CleanRecord(RoborockBase): - begin: Optional[int] = None - end: Optional[int] = None - duration: Optional[int] = None - area: Optional[int] = None - square_meter_area: Optional[float] = None - error: Optional[int] = None - complete: Optional[int] = None - start_type: Optional[int] = None - clean_type: Optional[int] = None - finish_reason: Optional[int] = None - dust_collection_status: Optional[int] = None - avoid_count: Optional[int] = None - wash_count: Optional[int] = None - map_flag: Optional[int] = None + begin: int | None = None + end: int | None = None + duration: int | None = None + area: int | None = None + square_meter_area: float | None = None + error: int | None = None + complete: int | None = None + start_type: int | None = None + clean_type: int | None = None + finish_reason: int | None = None + dust_collection_status: int | None = None + avoid_count: int | None = None + wash_count: int | None = None + map_flag: int | None = None def __post_init__(self) -> None: self.square_meter_area = round(self.area / 1000000, 1) if self.area is not None else None @@ -428,18 +428,18 @@ def __post_init__(self) -> None: @dataclass class Consumable(RoborockBase): - main_brush_work_time: Optional[int] = None - side_brush_work_time: Optional[int] = None - filter_work_time: Optional[int] = None - filter_element_work_time: Optional[int] = None - sensor_dirty_time: Optional[int] = None - strainer_work_times: Optional[int] = None - dust_collection_work_times: Optional[int] = None - cleaning_brush_work_times: Optional[int] = None - main_brush_time_left: Optional[int] = None - side_brush_time_left: Optional[int] = None - filter_time_left: Optional[int] = None - sensor_time_left: Optional[int] = None + main_brush_work_time: int | None = None + side_brush_work_time: int | None = None + filter_work_time: int | None = None + filter_element_work_time: int | None = None + sensor_dirty_time: int | None = None + strainer_work_times: int | None = None + dust_collection_work_times: int | None = None + cleaning_brush_work_times: int | None = None + main_brush_time_left: int | None = None + side_brush_time_left: int | None = None + filter_time_left: int | None = None + sensor_time_left: int | None = None def __post_init__(self) -> None: self.main_brush_time_left = ( @@ -458,61 +458,61 @@ def __post_init__(self) -> None: @dataclass class MultiMapsListMapInfoBakMaps(RoborockBase): - mapflag: Optional[Any] = None - add_time: Optional[Any] = None + mapflag: Any | None = None + add_time: Any | None = None @dataclass class MultiMapsListMapInfo(RoborockBase): _ignore_keys = ["mapFlag"] - mapFlag: Optional[Any] = None - add_time: Optional[Any] = None - length: Optional[Any] = None - name: Optional[Any] = None - bak_maps: Optional[list[MultiMapsListMapInfoBakMaps]] = None + mapFlag: Any | None = None + add_time: Any | None = None + length: Any | None = None + name: Any | None = None + bak_maps: list[MultiMapsListMapInfoBakMaps] | None = None @dataclass class MultiMapsList(RoborockBase): _ignore_keys = ["mapFlag"] - max_multi_map: Optional[int] = None - max_bak_map: Optional[int] = None - multi_map_count: Optional[int] = None - map_info: Optional[list[MultiMapsListMapInfo]] = None + max_multi_map: int | None = None + max_bak_map: int | None = None + multi_map_count: int | None = None + map_info: list[MultiMapsListMapInfo] | None = None @dataclass class SmartWashParams(RoborockBase): - smart_wash: Optional[int] = None - wash_interval: Optional[int] = None + smart_wash: int | None = None + wash_interval: int | None = None @dataclass class DustCollectionMode(RoborockBase): - mode: Optional[RoborockDockDustCollectionModeCode] = None + mode: RoborockDockDustCollectionModeCode | None = None @dataclass class WashTowelMode(RoborockBase): - wash_mode: Optional[RoborockDockWashTowelModeCode] = None + wash_mode: RoborockDockWashTowelModeCode | None = None @dataclass class NetworkInfo(RoborockBase): ip: str - ssid: Optional[str] = None - mac: Optional[str] = None - bssid: Optional[str] = None - rssi: Optional[int] = None + ssid: str | None = None + mac: str | None = None + bssid: str | None = None + rssi: int | None = None @dataclass class DeviceData(RoborockBase): device: HomeDataDevice model: str - host: Optional[str] = None + host: str | None = None @dataclass diff --git a/roborock/local_api.py b/roborock/local_api.py index 3fe08af..32b9239 100644 --- a/roborock/local_api.py +++ b/roborock/local_api.py @@ -3,7 +3,6 @@ import asyncio import logging from asyncio import Lock, TimerHandle, Transport -from typing import Optional import async_timeout @@ -39,7 +38,7 @@ def data_received(self, message): parser_msg, self.remaining = MessageParser.parse(message, local_key=self.device_info.device.local_key) self.on_message_received(parser_msg) - def connection_lost(self, exc: Optional[Exception]): + def connection_lost(self, exc: Exception | None): self.sync_disconnect() self.on_connection_lost(exc) @@ -83,11 +82,11 @@ async def async_disconnect(self) -> None: async with self._mutex: self.sync_disconnect() - def build_roborock_message(self, method: RoborockCommand, params: Optional[list | dict] = None) -> RoborockMessage: + def build_roborock_message(self, method: RoborockCommand, params: list | dict | None = None) -> RoborockMessage: secured = True if method in COMMANDS_SECURED else False request_id, timestamp, payload = self._get_payload(method, params, secured) request_protocol = RoborockMessageProtocol.GENERAL_REQUEST - message_retry: Optional[MessageRetry] = None + message_retry: MessageRetry | None = None if method == RoborockCommand.RETRY_REQUEST and isinstance(params, dict): message_retry = MessageRetry(method=params["method"], retry_id=params["retry_id"]) return RoborockMessage( @@ -122,7 +121,7 @@ async def ping(self): async def _send_command( self, method: RoborockCommand, - params: Optional[list | dict] = None, + params: list | dict | None = None, ): roborock_message = self.build_roborock_message(method, params) return await self.send_message(roborock_message) diff --git a/roborock/protocol.py b/roborock/protocol.py index 8d78655..2d9e5ac 100644 --- a/roborock/protocol.py +++ b/roborock/protocol.py @@ -7,7 +7,7 @@ import json import logging from asyncio import BaseTransport, Lock -from typing import Callable +from collections.abc import Callable from construct import ( # type: ignore Bytes, @@ -35,8 +35,8 @@ from roborock.roborock_message import RoborockMessage _LOGGER = logging.getLogger(__name__) -SALT = "TXdfu$jyZ#TZHsg4".encode() -BROADCAST_TOKEN = "qWKYcdQWrbm9hPqe".encode() +SALT = b"TXdfu$jyZ#TZHsg4" +BROADCAST_TOKEN = b"qWKYcdQWrbm9hPqe" AP_CONFIG = 1 SOCK_DISCOVERY = 2 diff --git a/roborock/roborock_message.py b/roborock/roborock_message.py index b180270..238039d 100644 --- a/roborock/roborock_message.py +++ b/roborock/roborock_message.py @@ -5,7 +5,6 @@ import time from dataclasses import dataclass from random import randint -from typing import Optional from roborock import RoborockEnum @@ -64,12 +63,12 @@ class MessageRetry: @dataclass class RoborockMessage: protocol: RoborockMessageProtocol - payload: Optional[bytes] = None + payload: bytes | None = None seq: int = randint(100000, 999999) version: bytes = b"1.0" random: int = randint(10000, 99999) timestamp: int = math.floor(time.time()) - message_retry: Optional[MessageRetry] = None + message_retry: MessageRetry | None = None def get_request_id(self) -> int | None: if self.payload: diff --git a/roborock/roborock_typing.py b/roborock/roborock_typing.py index f9219d5..6ce35c7 100644 --- a/roborock/roborock_typing.py +++ b/roborock/roborock_typing.py @@ -2,7 +2,6 @@ from dataclasses import dataclass from enum import Enum -from typing import Optional from .containers import ( CleanRecord, @@ -133,7 +132,7 @@ class RoborockCommand(str, Enum): @dataclass class CommandInfo: - params: Optional[list | dict] = None + params: list | dict | None = None CommandInfoMap: dict[RoborockCommand | None, CommandInfo] = { @@ -309,18 +308,18 @@ class CommandInfo: @dataclass class DockSummary(RoborockBase): - dust_collection_mode: Optional[DustCollectionMode] = None - wash_towel_mode: Optional[WashTowelMode] = None - smart_wash_params: Optional[SmartWashParams] = None + dust_collection_mode: DustCollectionMode | None = None + wash_towel_mode: WashTowelMode | None = None + smart_wash_params: SmartWashParams | None = None @dataclass class DeviceProp(RoborockBase): - status: Optional[Status] = None - clean_summary: Optional[CleanSummary] = None - consumable: Optional[Consumable] = None - last_clean_record: Optional[CleanRecord] = None - dock_summary: Optional[DockSummary] = None + status: Status | None = None + clean_summary: CleanSummary | None = None + consumable: Consumable | None = None + last_clean_record: CleanRecord | None = None + dock_summary: DockSummary | None = None def update(self, device_prop: DeviceProp) -> None: if device_prop.status: diff --git a/roborock/util.py b/roborock/util.py index a882cce..daeadca 100644 --- a/roborock/util.py +++ b/roborock/util.py @@ -6,15 +6,16 @@ import logging from asyncio import AbstractEventLoop, TimerHandle from collections.abc import MutableMapping -from typing import Any, Callable, Coroutine, Optional, Tuple, TypeVar +from typing import Any, TypeVar +from collections.abc import Callable, Coroutine from roborock import RoborockException T = TypeVar("T") -DEFAULT_TIME_ZONE: Optional[datetime.tzinfo] = datetime.datetime.now().astimezone().tzinfo +DEFAULT_TIME_ZONE: datetime.tzinfo | None = datetime.datetime.now().astimezone().tzinfo -def unpack_list(value: list[T], size: int) -> list[Optional[T]]: +def unpack_list(value: list[T], size: int) -> list[T | None]: return (value + [None] * size)[:size] # type: ignore @@ -29,7 +30,7 @@ def get_running_loop_or_create_one() -> AbstractEventLoop: def parse_datetime_to_roborock_datetime( start_datetime: datetime.datetime, end_datetime: datetime.datetime -) -> Tuple[datetime.datetime, datetime.datetime]: +) -> tuple[datetime.datetime, datetime.datetime]: now = datetime.datetime.now(DEFAULT_TIME_ZONE) start_datetime = start_datetime.replace( year=now.year, month=now.month, day=now.day, second=0, microsecond=0, tzinfo=DEFAULT_TIME_ZONE @@ -48,7 +49,7 @@ def parse_datetime_to_roborock_datetime( def parse_time_to_datetime( start_time: datetime.time, end_time: datetime.time -) -> Tuple[datetime.datetime, datetime.datetime]: +) -> tuple[datetime.datetime, datetime.datetime]: """Help to handle time data.""" start_datetime = datetime.datetime.now(DEFAULT_TIME_ZONE).replace( hour=start_time.hour, minute=start_time.minute, second=0, microsecond=0 @@ -78,7 +79,7 @@ def __init__(self, loop: AbstractEventLoop, callback: Callable[[], Coroutine], i self.loop = loop self.callback = callback self.interval = interval - self._task: Optional[TimerHandle] = None + self._task: TimerHandle | None = None async def _run_task(self): response = None @@ -107,4 +108,4 @@ def __init__(self, prefix: str, logger: logging.Logger) -> None: self.prefix = prefix def process(self, msg: str, kwargs: MutableMapping[str, Any]) -> tuple[str, MutableMapping[str, Any]]: - return "[%s] %s" % (self.prefix, msg), kwargs + return f"[{self.prefix}] {msg}", kwargs