Skip to content

Commit

Permalink
strict-typed the lib
Browse files Browse the repository at this point in the history
  • Loading branch information
jafar-atili committed Sep 27, 2022
1 parent 67a97d7 commit f204450
Show file tree
Hide file tree
Showing 5 changed files with 106 additions and 63 deletions.
1 change: 1 addition & 0 deletions MANIFEST.in
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
include src/switchbee/py.typed
19 changes: 19 additions & 0 deletions mypy.ini
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
[mypy]
python_version = 3.9
show_error_codes = true
follow_imports = silent
ignore_missing_imports = true
strict_equality = true
warn_incomplete_stub = true
warn_redundant_casts = true
warn_unused_configs = true
warn_unused_ignores = true
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
no_implicit_optional = true
warn_return_any = true
warn_unreachable = true
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"

[project]
name = "pyswitchbee"
version = "1.5.2"
version = "1.5.3"
description = "SwitchBee Python Integration."
readme = "README.md"
authors = [{ name = "Jafar Atili", email = "[email protected]" }]
Expand Down
108 changes: 66 additions & 42 deletions src/switchbee/api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
from datetime import timedelta
from json import JSONDecodeError
from logging import getLogger
from typing import Any, List, Union
from typing import Any, List

from aiohttp import ClientSession

Expand Down Expand Up @@ -78,7 +78,17 @@ def __init__(
self._last_conf_change: int = 0
self._devices_map: dict[
int,
SwitchBeeBaseDevice,
SwitchBeeSwitch
| SwitchBeeGroupSwitch
| SwitchBeeTimedSwitch
| SwitchBeeShutter
| SwitchBeeSomfy
| SwitchBeeDimmer
| SwitchBeeThermostat
| SwitchBeeScenario
| SwitchBeeRollingScenario
| SwitchBeeTimerSwitch
| SwitchBeeTwoWay,
] = {}

self._modules_map: dict[int, set] = {}
Expand All @@ -96,7 +106,22 @@ def mac(self) -> str | None:
return self._mac

@property
def devices(self) -> dict[int, SwitchBeeBaseDevice]:
def devices(
self,
) -> dict[
int,
SwitchBeeSwitch
| SwitchBeeGroupSwitch
| SwitchBeeTimedSwitch
| SwitchBeeShutter
| SwitchBeeSomfy
| SwitchBeeDimmer
| SwitchBeeThermostat
| SwitchBeeScenario
| SwitchBeeRollingScenario
| SwitchBeeTimerSwitch
| SwitchBeeTwoWay,
]:
return self._devices_map

@property
Expand All @@ -113,7 +138,7 @@ def devices_list(
def reconnect_count(self) -> int:
return self._login_count

def module_display(self, unit_id: int):
def module_display(self, unit_id: int) -> str:
return " and ".join(list(self._modules_map[unit_id]))

async def login_if_needed(self) -> None:
Expand All @@ -135,7 +160,7 @@ async def _post(self, body: dict) -> dict:
) as response:
if response.status == 200:
try:
json_result = await response.json(
json_result: dict = await response.json(
content_type=None, encoding="utf8"
)
if json_result[ApiAttribute.STATUS] != ApiStatus.OK:
Expand Down Expand Up @@ -207,36 +232,36 @@ async def _login(self) -> None:
# self._token_expiration = resp[ApiAttribute.DATA][ApiAttribute.EXPIRATION]
self._token_expiration = timestamp_now() + TOKEN_EXPIRATION

async def get_configuration(self):
async def get_configuration(self) -> dict:
await self.login_if_needed()
return await self._send_request(ApiCommand.GET_CONF)

async def get_multiple_states(self, ids: list):
async def get_multiple_states(self, ids: list) -> dict:
"""returns JSON {'status': 'OK', 'data': [{'id': 212, 'state': 'OFF'}, {'id': 343, 'state': 'OFF'}]}"""
await self.login_if_needed()
return await self._send_request(ApiCommand.GET_MULTI_STATES, ids)

async def get_state(self, id: int):
async def get_state(self, id: int) -> dict:
"""returns JSON {'status': 'OK', 'data': 'OFF'}"""
await self.login_if_needed()
return await self._send_request(ApiCommand.GET_STATE, id)

async def set_state(self, id: int, state):
async def set_state(self, id: int, state: str | int) -> dict:
"""returns JSON {'status': 'OK', 'data': 'OFF/ON'}"""
await self.login_if_needed()
return await self._send_request(
ApiCommand.OPERATE, {"directive": "SET", "itemId": id, "value": state}
)

async def get_stats(self):
async def get_stats(self) -> dict:
"""returns {'status': 'OK', 'data': {}} on my unit"""
await self.login_if_needed()
return await self._send_request(ApiCommand.STATS)

async def fetch_configuration(
self,
include: list[DeviceType] | None = [],
):
) -> None:
await self.login_if_needed()
data = await self.get_configuration()
if data[ApiAttribute.STATUS] != ApiStatus.OK:
Expand Down Expand Up @@ -406,7 +431,7 @@ async def fetch_configuration(

async def fetch_states(
self,
):
) -> None:

states = await self.get_multiple_states(
[
Expand All @@ -426,47 +451,46 @@ async def fetch_states(
]
)

for device in states[ApiAttribute.DATA]:
device_id = device[ApiAttribute.ID]
for device_state in states[ApiAttribute.DATA]:
device_id = device_state[ApiAttribute.ID]

if device_id not in self._devices_map:
continue

if self._devices_map[device_id].type == DeviceType.Dimmer:
self._devices_map[device_id].brightness = device[ApiAttribute.STATE]
elif self._devices_map[device_id].type == DeviceType.Shutter:
self._devices_map[device_id].position = device[ApiAttribute.STATE]
elif self._devices_map[device_id].type in [
DeviceType.Switch,
DeviceType.GroupSwitch,
DeviceType.TimedSwitch,
DeviceType.TimedPowerSwitch,
]:
self._devices_map[device_id].state = device[ApiAttribute.STATE]
elif self._devices_map[device_id].type == DeviceType.Thermostat:
device = self._devices_map[device_id]

if isinstance(device, SwitchBeeDimmer):
device.brightness = device_state[ApiAttribute.STATE]
elif isinstance(device, SwitchBeeShutter):
device.position = device_state[ApiAttribute.STATE]
elif isinstance(
device,
(
SwitchBeeSwitch,
SwitchBeeGroupSwitch,
SwitchBeeTimedSwitch,
SwitchBeeTimerSwitch,
),
):

device.state = device_state[ApiAttribute.STATE]
elif isinstance(device, SwitchBeeThermostat):
try:
self._devices_map[device_id].state = device[ApiAttribute.STATE][
ApiAttribute.POWER
]
device.state = device_state[ApiAttribute.STATE][ApiAttribute.POWER]
except TypeError:
logger.error(
"%s: Recieved invalid state from CU, keeping the old one: %s",
self._devices_map[device_id].name,
device,
"%s: Received invalid state from CU, keeping the old one: %s",
device.name,
device_state,
)
continue

self._devices_map[device_id].mode = device[ApiAttribute.STATE][
ApiAttribute.MODE
]
device.mode = device_state[ApiAttribute.STATE][ApiAttribute.MODE]
device.fan = device_state[ApiAttribute.STATE][ApiAttribute.FAN]

self._devices_map[device_id].fan = device[ApiAttribute.STATE][
ApiAttribute.FAN
device.target_temperature = device_state[ApiAttribute.STATE][
ApiAttribute.CONFIGURED_TEMPERATURE
]

self._devices_map[device_id].target_temperature = device[
ApiAttribute.STATE
][ApiAttribute.CONFIGURED_TEMPERATURE]
self._devices_map[device_id].temperature = device[ApiAttribute.STATE][
device.temperature = device_state[ApiAttribute.STATE][
ApiAttribute.ROOM_TEMPERATURE
]
39 changes: 19 additions & 20 deletions src/switchbee/device/__init__.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
from __future__ import annotations

from abc import ABC
from dataclasses import dataclass, field
from enum import Enum, unique
from typing import List, Union, final
from typing import List, final

from ..api.utils import timestamp_now
from ..const import ApiDeviceHardware, ApiDeviceType, ApiStateCommand
Expand All @@ -28,21 +27,21 @@ class DeviceType(Enum):
IrDevice = ApiDeviceType.IR_DEVICE, "Infra Red Device"
RollingScenario = ApiDeviceType.ROLLING_SCENARIO, "Rolling Scenario"

def __new__(cls, *args, **kwds):
def __new__(cls, *args, **kwds): # type: ignore
obj = object.__new__(cls)
obj._value_ = args[0]
return obj

# ignore the first param since it's already set by __new__
def __init__(self, _: str, display: str = None):
def __init__(self, _: str, display: str = "") -> None:
self._display = display

def __str__(self):
def __str__(self) -> str:
return self.display

# this makes sure that the description is read-only
@property
def display(self):
def display(self) -> str:
return self._display


Expand All @@ -58,26 +57,26 @@ class HardwareType(Enum):
RegularSwitch = ApiDeviceHardware.REGULAR_SWITCH, "Regular Switch"
Repeater = ApiDeviceHardware.REPEATER, "Repeater"

def __new__(cls, *args, **kwds):
def __new__(cls, *args, **kwds): # type: ignore
obj = object.__new__(cls)
obj._value_ = args[0]
return obj

# ignore the first param since it's already set by __new__
def __init__(self, _: str, display: str = None):
def __init__(self, _: str, display: str = "") -> None:
self._display = display

def __str__(self):
def __str__(self) -> str:
return self.display

# this makes sure that the description is read-only
@property
def display(self):
def display(self) -> str:
return self._display


@dataclass
class SwitchBeeBaseDevice(ABC):
class SwitchBeeBaseDevice:
id: int
name: str
zone: str
Expand All @@ -89,12 +88,12 @@ def __post_init__(self) -> None:
self.last_data_update = timestamp_now()
self.unit_id = self.id // 10

def __hash__(self):
def __hash__(self) -> int:
return self.id


@dataclass
class SwitchBeeBaseSwitch(ABC):
class SwitchBeeBaseSwitch:
_state: str | None = field(init=False, default=None)

@property
Expand All @@ -107,15 +106,15 @@ def state(self, value: str) -> None:


@dataclass
class SwitchBeeBaseShutter(ABC):
class SwitchBeeBaseShutter:
_position: int | None = field(init=False, repr=False, default=None)

@property
def position(self) -> int | None:
return self._position

@position.setter
def position(self, value: Union[str, int]) -> None:
def position(self, value: str | int) -> None:

if value == ApiStateCommand.OFF:
self._position = 0
Expand All @@ -126,7 +125,7 @@ def position(self, value: Union[str, int]) -> None:


@dataclass
class SwitchBeeBaseDimmer(ABC):
class SwitchBeeBaseDimmer:

_brightness: int = field(init=False)

Expand All @@ -135,7 +134,7 @@ def brightness(self) -> int:
return self._brightness

@brightness.setter
def brightness(self, value: Union[str, int]) -> None:
def brightness(self, value: str | int) -> None:
if value == ApiStateCommand.OFF:
self._brightness = 0
elif value == ApiStateCommand.ON:
Expand All @@ -145,7 +144,7 @@ def brightness(self, value: Union[str, int]) -> None:


@dataclass
class SwitchBeeBaseTimer(ABC):
class SwitchBeeBaseTimer:
_minutes_left: int = field(init=False)
_state: str | int = field(init=False)

Expand All @@ -169,9 +168,8 @@ def minutes_left(self) -> int:
return self._minutes_left



@dataclass
class SwitchBeeBaseThermostat(ABC):
class SwitchBeeBaseThermostat:

modes: List[str]
unit: str
Expand All @@ -183,6 +181,7 @@ class SwitchBeeBaseThermostat(ABC):
min_temperature: int = 16


@final
@dataclass
class SwitchBeeSwitch(SwitchBeeBaseSwitch, SwitchBeeBaseDevice):
def __post_init__(self) -> None:
Expand Down

0 comments on commit f204450

Please sign in to comment.