Skip to content

Commit

Permalink
Enable and convert to future annotations (#838)
Browse files Browse the repository at this point in the history
  • Loading branch information
sdb9696 committed Apr 17, 2024
1 parent 82d92ae commit 203bd79
Show file tree
Hide file tree
Showing 59 changed files with 562 additions and 462 deletions.
13 changes: 7 additions & 6 deletions devtools/dump_devinfo.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,8 @@
and finally execute a query to query all of them at once.
"""

from __future__ import annotations

import base64
import collections.abc
import json
Expand All @@ -17,7 +19,6 @@
from collections import defaultdict, namedtuple
from pathlib import Path
from pprint import pprint
from typing import Dict, List, Union

import asyncclick as click

Expand Down Expand Up @@ -143,7 +144,7 @@ def default_to_regular(d):
async def handle_device(basedir, autosave, device: Device, batch_size: int):
"""Create a fixture for a single device instance."""
if isinstance(device, SmartDevice):
fixture_results: List[FixtureResult] = await get_smart_fixtures(
fixture_results: list[FixtureResult] = await get_smart_fixtures(
device, batch_size
)
else:
Expand Down Expand Up @@ -344,12 +345,12 @@ def _echo_error(msg: str):

async def _make_requests_or_exit(
device: SmartDevice,
requests: List[SmartRequest],
requests: list[SmartRequest],
name: str,
batch_size: int,
*,
child_device_id: str,
) -> Dict[str, Dict]:
) -> dict[str, dict]:
final = {}
protocol = (
device.protocol
Expand All @@ -362,7 +363,7 @@ async def _make_requests_or_exit(
for i in range(0, end, step):
x = i
requests_step = requests[x : x + step]
request: Union[List[SmartRequest], SmartRequest] = (
request: list[SmartRequest] | SmartRequest = (
requests_step[0] if len(requests_step) == 1 else requests_step
)
responses = await protocol.query(SmartRequest._create_request_dict(request))
Expand Down Expand Up @@ -586,7 +587,7 @@ async def get_smart_fixtures(device: SmartDevice, batch_size: int):
finally:
await device.protocol.close()

device_requests: Dict[str, List[SmartRequest]] = {}
device_requests: dict[str, list[SmartRequest]] = {}
for success in successes:
device_request = device_requests.setdefault(success.child_device_id, [])
device_request.append(success.request)
Expand Down
91 changes: 46 additions & 45 deletions devtools/helpers/smartrequests.py
Original file line number Diff line number Diff line change
Expand Up @@ -25,17 +25,18 @@
"""

from __future__ import annotations

import logging
from dataclasses import asdict, dataclass
from typing import List, Optional, Union

_LOGGER = logging.getLogger(__name__)


class SmartRequest:
"""Class to represent a smart protocol request."""

def __init__(self, method_name: str, params: Optional["SmartRequestParams"] = None):
def __init__(self, method_name: str, params: SmartRequestParams | None = None):
self.method_name = method_name
if params:
self.params = params.to_dict()
Expand Down Expand Up @@ -93,7 +94,7 @@ class GetTriggerLogsParams(SmartRequestParams):
class LedStatusParams(SmartRequestParams):
"""LED Status params."""

led_rule: Optional[str] = None
led_rule: str | None = None

@staticmethod
def from_bool(state: bool):
Expand All @@ -105,42 +106,42 @@ def from_bool(state: bool):
class LightInfoParams(SmartRequestParams):
"""LightInfo params."""

brightness: Optional[int] = None
color_temp: Optional[int] = None
hue: Optional[int] = None
saturation: Optional[int] = None
brightness: int | None = None
color_temp: int | None = None
hue: int | None = None
saturation: int | None = None

@dataclass
class DynamicLightEffectParams(SmartRequestParams):
"""LightInfo params."""

enable: bool
id: Optional[str] = None
id: str | None = None

@staticmethod
def get_raw_request(
method: str, params: Optional[SmartRequestParams] = None
) -> "SmartRequest":
method: str, params: SmartRequestParams | None = None
) -> SmartRequest:
"""Send a raw request to the device."""
return SmartRequest(method, params)

@staticmethod
def component_nego() -> "SmartRequest":
def component_nego() -> SmartRequest:
"""Get quick setup component info."""
return SmartRequest("component_nego")

@staticmethod
def get_device_info() -> "SmartRequest":
def get_device_info() -> SmartRequest:
"""Get device info."""
return SmartRequest("get_device_info")

@staticmethod
def get_device_usage() -> "SmartRequest":
def get_device_usage() -> SmartRequest:
"""Get device usage."""
return SmartRequest("get_device_usage")

@staticmethod
def device_info_list(ver_code) -> List["SmartRequest"]:
def device_info_list(ver_code) -> list[SmartRequest]:
"""Get device info list."""
if ver_code == 1:
return [SmartRequest.get_device_info()]
Expand All @@ -151,126 +152,126 @@ def device_info_list(ver_code) -> List["SmartRequest"]:
]

@staticmethod
def get_auto_update_info() -> "SmartRequest":
def get_auto_update_info() -> SmartRequest:
"""Get auto update info."""
return SmartRequest("get_auto_update_info")

@staticmethod
def firmware_info_list() -> List["SmartRequest"]:
def firmware_info_list() -> list[SmartRequest]:
"""Get info list."""
return [
SmartRequest.get_raw_request("get_fw_download_state"),
SmartRequest.get_raw_request("get_latest_fw"),
]

@staticmethod
def qs_component_nego() -> "SmartRequest":
def qs_component_nego() -> SmartRequest:
"""Get quick setup component info."""
return SmartRequest("qs_component_nego")

@staticmethod
def get_device_time() -> "SmartRequest":
def get_device_time() -> SmartRequest:
"""Get device time."""
return SmartRequest("get_device_time")

@staticmethod
def get_child_device_list() -> "SmartRequest":
def get_child_device_list() -> SmartRequest:
"""Get child device list."""
return SmartRequest("get_child_device_list")

@staticmethod
def get_child_device_component_list() -> "SmartRequest":
def get_child_device_component_list() -> SmartRequest:
"""Get child device component list."""
return SmartRequest("get_child_device_component_list")

@staticmethod
def get_wireless_scan_info(
params: Optional[GetRulesParams] = None,
) -> "SmartRequest":
params: GetRulesParams | None = None,
) -> SmartRequest:
"""Get wireless scan info."""
return SmartRequest(
"get_wireless_scan_info", params or SmartRequest.GetRulesParams()
)

@staticmethod
def get_schedule_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_schedule_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get schedule rules."""
return SmartRequest(
"get_schedule_rules", params or SmartRequest.GetScheduleRulesParams()
)

@staticmethod
def get_next_event(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_next_event(params: GetRulesParams | None = None) -> SmartRequest:
"""Get next scheduled event."""
return SmartRequest("get_next_event", params or SmartRequest.GetRulesParams())

@staticmethod
def schedule_info_list() -> List["SmartRequest"]:
def schedule_info_list() -> list[SmartRequest]:
"""Get schedule info list."""
return [
SmartRequest.get_schedule_rules(),
SmartRequest.get_next_event(),
]

@staticmethod
def get_countdown_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_countdown_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get countdown rules."""
return SmartRequest(
"get_countdown_rules", params or SmartRequest.GetRulesParams()
)

@staticmethod
def get_antitheft_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_antitheft_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get antitheft rules."""
return SmartRequest(
"get_antitheft_rules", params or SmartRequest.GetRulesParams()
)

@staticmethod
def get_led_info(params: Optional[LedStatusParams] = None) -> "SmartRequest":
def get_led_info(params: LedStatusParams | None = None) -> SmartRequest:
"""Get led info."""
return SmartRequest("get_led_info", params or SmartRequest.LedStatusParams())

@staticmethod
def get_auto_off_config(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_auto_off_config(params: GetRulesParams | None = None) -> SmartRequest:
"""Get auto off config."""
return SmartRequest(
"get_auto_off_config", params or SmartRequest.GetRulesParams()
)

@staticmethod
def get_delay_action_info() -> "SmartRequest":
def get_delay_action_info() -> SmartRequest:
"""Get delay action info."""
return SmartRequest("get_delay_action_info")

@staticmethod
def auto_off_list() -> List["SmartRequest"]:
def auto_off_list() -> list[SmartRequest]:
"""Get energy usage."""
return [
SmartRequest.get_auto_off_config(),
SmartRequest.get_delay_action_info(), # May not live here
]

@staticmethod
def get_energy_usage() -> "SmartRequest":
def get_energy_usage() -> SmartRequest:
"""Get energy usage."""
return SmartRequest("get_energy_usage")

@staticmethod
def energy_monitoring_list() -> List["SmartRequest"]:
def energy_monitoring_list() -> list[SmartRequest]:
"""Get energy usage."""
return [
SmartRequest("get_energy_usage"),
SmartRequest.get_raw_request("get_electricity_price_config"),
]

@staticmethod
def get_current_power() -> "SmartRequest":
def get_current_power() -> SmartRequest:
"""Get current power."""
return SmartRequest("get_current_power")

@staticmethod
def power_protection_list() -> List["SmartRequest"]:
def power_protection_list() -> list[SmartRequest]:
"""Get power protection info list."""
return [
SmartRequest.get_current_power(),
Expand All @@ -279,45 +280,45 @@ def power_protection_list() -> List["SmartRequest"]:
]

@staticmethod
def get_preset_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest":
def get_preset_rules(params: GetRulesParams | None = None) -> SmartRequest:
"""Get preset rules."""
return SmartRequest("get_preset_rules", params or SmartRequest.GetRulesParams())

@staticmethod
def get_auto_light_info() -> "SmartRequest":
def get_auto_light_info() -> SmartRequest:
"""Get auto light info."""
return SmartRequest("get_auto_light_info")

@staticmethod
def get_dynamic_light_effect_rules(
params: Optional[GetRulesParams] = None,
) -> "SmartRequest":
params: GetRulesParams | None = None,
) -> SmartRequest:
"""Get dynamic light effect rules."""
return SmartRequest(
"get_dynamic_light_effect_rules", params or SmartRequest.GetRulesParams()
)

@staticmethod
def set_device_on(params: DeviceOnParams) -> "SmartRequest":
def set_device_on(params: DeviceOnParams) -> SmartRequest:
"""Set device on state."""
return SmartRequest("set_device_info", params)

@staticmethod
def set_light_info(params: LightInfoParams) -> "SmartRequest":
def set_light_info(params: LightInfoParams) -> SmartRequest:
"""Set color temperature."""
return SmartRequest("set_device_info", params)

@staticmethod
def set_dynamic_light_effect_rule_enable(
params: DynamicLightEffectParams,
) -> "SmartRequest":
) -> SmartRequest:
"""Enable dynamic light effect rule."""
return SmartRequest("set_dynamic_light_effect_rule_enable", params)

@staticmethod
def get_component_info_requests(component_nego_response) -> List["SmartRequest"]:
def get_component_info_requests(component_nego_response) -> list[SmartRequest]:
"""Get a list of requests based on the component info response."""
request_list: List["SmartRequest"] = []
request_list: list[SmartRequest] = []
for component in component_nego_response["component_list"]:
if (
requests := get_component_requests(
Expand All @@ -329,7 +330,7 @@ def get_component_info_requests(component_nego_response) -> List["SmartRequest"]

@staticmethod
def _create_request_dict(
smart_request: Union["SmartRequest", List["SmartRequest"]],
smart_request: SmartRequest | list[SmartRequest],
) -> dict:
"""Create request dict to be passed to SmartProtocol.query()."""
if isinstance(smart_request, list):
Expand Down
Loading

0 comments on commit 203bd79

Please sign in to comment.