diff --git a/setup.cfg b/setup.cfg index 9d3b91f7a9..fd7844cee9 100644 --- a/setup.cfg +++ b/setup.cfg @@ -51,7 +51,7 @@ warn_unreachable = true implicit_reexport = true # partly typechecked modules (extra block for better overview) -[mypy-xknx.devices.device,xknx.devices.devices,xknx.devices.travelcalculator,xknx.remote_value.remote_value] +[mypy-xknx.devices.action,xknx.devices.binary_sensor,xknx.devices.climate,xknx.devices.climate_mode,xknx.devices.device,xknx.devices.devices,xknx.devices.travelcalculator,xknx.remote_value.remote_value,xknx.remote_value.remote_value_climate_mode] strict = true ignore_errors = false warn_unreachable = true diff --git a/test/config_tests/config_v1_test.py b/test/config_tests/config_v1_test.py index be7b10bc16..a850da1956 100644 --- a/test/config_tests/config_v1_test.py +++ b/test/config_tests/config_v1_test.py @@ -382,7 +382,9 @@ def test_config_climate_operation_mode(self): self.assertEqual( TestConfig.xknx.devices["Office.Climate"].mode, ClimateMode( - TestConfig.xknx, name=None, group_address_operation_mode="1/7/6" + TestConfig.xknx, + name="Office.Climate_mode", + group_address_operation_mode="1/7/6", ), ) @@ -392,7 +394,7 @@ def test_config_climate_operation_mode2(self): TestConfig.xknx.devices["Attic.Climate"].mode, ClimateMode( TestConfig.xknx, - name=None, + name="Attic.Climate_mode", group_address_operation_mode_protection="1/7/8", group_address_operation_mode_night="1/7/9", group_address_operation_mode_comfort="1/7/10", @@ -405,7 +407,7 @@ def test_config_climate_operation_mode_state(self): TestConfig.xknx.devices["Bath.Climate"].mode, ClimateMode( TestConfig.xknx, - name=None, + name="Bath.Climate_mode", group_address_operation_mode="1/7/6", group_address_operation_mode_state="1/7/7", ), @@ -417,7 +419,7 @@ def test_config_climate_controller_status_state(self): TestConfig.xknx.devices["Cellar.Climate"].mode, ClimateMode( TestConfig.xknx, - name=None, + name="Cellar.Climate_mode", group_address_controller_status="1/7/12", group_address_controller_status_state="1/7/13", ), diff --git a/test/devices_tests/climate_test.py b/test/devices_tests/climate_test.py index 19dcfeecb6..69c75e0a89 100644 --- a/test/devices_tests/climate_test.py +++ b/test/devices_tests/climate_test.py @@ -15,9 +15,8 @@ DPTHVACMode, DPTTemperature, DPTValue1Count, - HVACOperationMode, ) -from xknx.dpt.dpt_hvac_mode import HVACControllerMode +from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACOperationMode from xknx.exceptions import CouldNotParseTelegram, DeviceIllegalValue from xknx.telegram import GroupAddress, Telegram from xknx.telegram.apci import GroupValueRead, GroupValueWrite diff --git a/test/dpt_tests/dpt_hvac_mode_test.py b/test/dpt_tests/dpt_hvac_mode_test.py index 0c45cb8aa6..9608ed847c 100644 --- a/test/dpt_tests/dpt_hvac_mode_test.py +++ b/test/dpt_tests/dpt_hvac_mode_test.py @@ -1,7 +1,8 @@ """Unit test for KNX DPT HVAC Operation modes.""" import unittest -from xknx.dpt import DPTControllerStatus, DPTHVACMode, HVACOperationMode +from xknx.dpt import DPTControllerStatus, DPTHVACMode +from xknx.dpt.dpt_hvac_mode import HVACOperationMode from xknx.exceptions import ConversionError, CouldNotParseKNXIP diff --git a/test/remote_value_tests/remote_value_climate_mode_test.py b/test/remote_value_tests/remote_value_climate_mode_test.py index 4ce93b59ad..c34ccf0bb3 100644 --- a/test/remote_value_tests/remote_value_climate_mode_test.py +++ b/test/remote_value_tests/remote_value_climate_mode_test.py @@ -3,15 +3,14 @@ import unittest from xknx import XKNX -from xknx.dpt import DPTArray, DPTBinary, HVACOperationMode -from xknx.dpt.dpt_hvac_mode import HVACControllerMode +from xknx.dpt import DPTArray, DPTBinary +from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACOperationMode from xknx.exceptions import ConversionError, CouldNotParseTelegram from xknx.remote_value import ( RemoteValueBinaryHeatCool, RemoteValueBinaryOperationMode, RemoteValueClimateMode, ) -from xknx.remote_value.remote_value_climate_mode import _RemoteValueBinaryClimateMode from xknx.telegram import GroupAddress, Telegram from xknx.telegram.apci import GroupValueWrite @@ -137,12 +136,6 @@ def test_from_knx_unknown_operation_mode(self): with self.assertRaises(ConversionError): RemoteValueBinaryHeatCool(xknx, controller_mode=None) - def test_supported_operation_modes_not_implemented(self): - """Test from_knx function with unsupported operation.""" - xknx = XKNX() - with self.assertRaises(NotImplementedError): - _RemoteValueBinaryClimateMode.supported_operation_modes() - def test_to_knx_error_operation_mode(self): """Test to_knx function with wrong parameter.""" xknx = XKNX() diff --git a/test/remote_value_tests/remote_value_datetime_test.py b/test/remote_value_tests/remote_value_datetime_test.py index c1b18ce481..1b452326d7 100644 --- a/test/remote_value_tests/remote_value_datetime_test.py +++ b/test/remote_value_tests/remote_value_datetime_test.py @@ -4,16 +4,9 @@ import unittest from xknx import XKNX -from xknx.dpt import DPTArray, DPTBinary, HVACOperationMode -from xknx.exceptions import ConversionError, CouldNotParseTelegram -from xknx.remote_value import ( - RemoteValueBinaryHeatCool, - RemoteValueBinaryOperationMode, - RemoteValueClimateMode, - RemoteValueDateTime, -) -from xknx.remote_value.remote_value_climate_mode import _RemoteValueBinaryClimateMode -from xknx.telegram import GroupAddress, Telegram +from xknx.dpt import DPTArray +from xknx.exceptions import ConversionError +from xknx.remote_value import RemoteValueDateTime class TestRemoteValueDateTime(unittest.TestCase): diff --git a/test/remote_value_tests/remote_value_test.py b/test/remote_value_tests/remote_value_test.py index c68a97e119..3c44346c34 100644 --- a/test/remote_value_tests/remote_value_test.py +++ b/test/remote_value_tests/remote_value_test.py @@ -24,16 +24,6 @@ def tearDown(self): """Tear down test class.""" self.loop.close() - def test_warn_payload_valid(self): - """Test for warning if payload_valid is not implemented.""" - xknx = XKNX() - remote_value = RemoteValue(xknx) - with patch("logging.Logger.warning") as mock_warn: - remote_value.payload_valid(DPTBinary(0)) - mock_warn.assert_called_with( - "'payload_valid()' not implemented for %s", "RemoteValue" - ) - def test_info_set_uninitialized(self): """Test for info if RemoteValue is not initialized.""" xknx = XKNX() diff --git a/xknx/devices/action.py b/xknx/devices/action.py index abc0a50001..dffd72fad4 100644 --- a/xknx/devices/action.py +++ b/xknx/devices/action.py @@ -1,5 +1,9 @@ """Module for handling commands which may be attached to BinarySensor class.""" import logging +from typing import TYPE_CHECKING, Any, Awaitable, Callable, Optional + +if TYPE_CHECKING: + from xknx.xknx import XKNX logger = logging.getLogger("xknx.log") @@ -7,13 +11,13 @@ class ActionBase: """Base Class for handling commands.""" - def __init__(self, xknx, hook="on", counter=1): + def __init__(self, xknx: "XKNX", hook: str = "on", counter: Optional[int] = 1): """Initialize Action_Base class.""" self.xknx = xknx self.hook = hook self.counter = counter - def test_counter(self, counter): + def test_counter(self, counter: Optional[int]) -> bool: """Test if action filters for specific counter.""" if self.counter is None: # no specific counter_filter -> always true @@ -22,7 +26,7 @@ def test_counter(self, counter): return True return counter == self.counter - def test_if_applicable(self, state, counter=None): + def test_if_applicable(self, state: bool, counter: Optional[int] = None) -> bool: """Test if should be executed for this state and this counter number.""" if state and (self.hook == "on"): return self.test_counter(counter) @@ -30,15 +34,15 @@ def test_if_applicable(self, state, counter=None): return self.test_counter(counter) return False - async def execute(self): + async def execute(self) -> None: """Execute action. To be overwritten in derived classes.""" logger.info("Execute not implemented for %s", self.__class__.__name__) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return f'' - def __eq__(self, other): + def __eq__(self, other: object) -> bool: """Equal operator.""" return self.__dict__ == other.__dict__ @@ -46,7 +50,14 @@ def __eq__(self, other): class Action(ActionBase): """Class for handling commands.""" - def __init__(self, xknx, hook="on", target=None, method=None, counter=1): + def __init__( + self, + xknx: "XKNX", + hook: str = "on", + target: Optional[str] = None, + method: Optional[str] = None, + counter: int = 1, + ): """Initialize Action class.""" # pylint: disable=too-many-arguments super().__init__(xknx, hook, counter) @@ -54,7 +65,7 @@ def __init__(self, xknx, hook="on", target=None, method=None, counter=1): self.method = method @classmethod - def from_config(cls, xknx, config): + def from_config(cls, xknx: "XKNX", config: Any) -> "Action": """Initialize object from configuration structure.""" hook = config.get("hook", "on") target = config.get("target") @@ -62,15 +73,15 @@ def from_config(cls, xknx, config): counter = config.get("counter", 1) return cls(xknx, hook=hook, target=target, method=method, counter=counter) - async def execute(self): + async def execute(self) -> None: """Execute action.""" - if self.target is not None: + if self.target is not None and self.method is not None: if self.target not in self.xknx.devices: logger.warning("Unknown device %s witin action %s.", self.target, self) return await self.xknx.devices[self.target].do(self.method) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.target, self.method, super().__str__() @@ -80,17 +91,23 @@ def __str__(self): class ActionCallback(ActionBase): """Class for handling commands via callbacks.""" - def __init__(self, xknx, callback, hook="on", counter=1): + def __init__( + self, + xknx: "XKNX", + callback: Callable[[], Awaitable[None]], + hook: str = "on", + counter: int = 1, + ): """Initialize Action class.""" # pylint: disable=too-many-arguments super().__init__(xknx, hook, counter) self.callback = callback - async def execute(self): + async def execute(self) -> None: """Execute callback.""" await self.callback() - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.callback.__name__, super().__str__() diff --git a/xknx/devices/binary_sensor.py b/xknx/devices/binary_sensor.py index 2880ca2008..80b283522c 100644 --- a/xknx/devices/binary_sensor.py +++ b/xknx/devices/binary_sensor.py @@ -10,12 +10,17 @@ """ import asyncio import time -from typing import List, Optional +from typing import TYPE_CHECKING, Any, Iterator, List, Optional, cast from xknx.remote_value import RemoteValueSwitch from .action import Action -from .device import Device +from .device import Device, DeviceCallbackType + +if TYPE_CHECKING: + from xknx.telegram import Telegram + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX class BinarySensor(Device): @@ -24,17 +29,17 @@ class BinarySensor(Device): # pylint: disable=too-many-instance-attributes def __init__( self, - xknx, + xknx: "XKNX", name: str, - group_address_state=None, + group_address_state: "GroupAddressableType" = None, invert: Optional[bool] = False, sync_state: bool = True, ignore_internal_state: bool = False, - device_class: str = None, + device_class: Optional[str] = None, reset_after: Optional[float] = None, - actions: List[Action] = None, + actions: Optional[List[Action]] = None, context_timeout: Optional[float] = None, - device_updated_cb=None, + device_updated_cb: Optional[DeviceCallbackType] = None, ): """Initialize BinarySensor class.""" # pylint: disable=too-many-arguments @@ -46,14 +51,14 @@ def __init__( self.device_class = device_class self.ignore_internal_state = ignore_internal_state or bool(context_timeout) self.reset_after = reset_after - self.state = None + self.state: Optional[bool] = None self._context_timeout = context_timeout self._count_set_on = 0 self._count_set_off = 0 - self._last_set = None - self._reset_task = None - self._context_task = None + self._last_set: Optional[float] = None + self._reset_task: Optional[asyncio.Task[None]] = None + self._context_task: Optional[asyncio.Task[None]] = None # TODO: log a warning if reset_after and sync_state are true ? This could cause actions to self-fire. self.remote_value = RemoteValueSwitch( xknx, @@ -65,11 +70,11 @@ def __init__( after_update_cb=self._state_from_remote_value, ) - def _iter_remote_values(self): + def _iter_remote_values(self) -> Iterator[RemoteValueSwitch]: """Iterate the devices RemoteValue classes.""" yield self.remote_value - def __del__(self): + def __del__(self) -> None: """Destructor. Cleaning up if this was not done before.""" if self._reset_task: self._reset_task.cancel() @@ -78,7 +83,7 @@ def __del__(self): self._context_task.cancel() @classmethod - def from_config(cls, xknx, name, config): + def from_config(cls, xknx: "XKNX", name: str, config: Any) -> "BinarySensor": """Initialize object from configuration structure.""" group_address_state = config.get("group_address_state") invert = config.get("invert") @@ -106,11 +111,11 @@ def from_config(cls, xknx, name, config): actions=actions, ) - async def _state_from_remote_value(self): + async def _state_from_remote_value(self) -> None: """Update the internal state from RemoteValue (Callback).""" await self._set_internal_state(self.remote_value.value) - async def _set_internal_state(self, state: bool): + async def _set_internal_state(self, state: bool) -> None: """Set the internal state of the device. If state was changed after_update hooks and connected Actions are executed.""" if state != self.state or self.ignore_internal_state: self.state = state @@ -125,7 +130,7 @@ async def _set_internal_state(self, state: bool): else: await self._trigger_callbacks() - async def _counter_task(self, wait_seconds: float): + async def _counter_task(self, wait_seconds: float) -> None: """Trigger after 1 second to prevent double triggers.""" await asyncio.sleep(wait_seconds) await self._trigger_callbacks() @@ -135,16 +140,16 @@ async def _counter_task(self, wait_seconds: float): await self.after_update() - async def _trigger_callbacks(self): + async def _trigger_callbacks(self) -> None: """Trigger callbacks for device and execute actions if any.""" await self.after_update() for action in self.actions: - if action.test_if_applicable(self.state, self.counter): + if action.test_if_applicable(cast(bool, self.state), self.counter): await action.execute() @property - def counter(self): + def counter(self) -> int: """Return current counter for sensor.""" return self._count_set_on if self.state else self._count_set_off @@ -159,7 +164,7 @@ def within_same_context() -> bool: new_set_time = time.time() time_diff = new_set_time - self._last_set self._last_set = new_set_time - return time_diff < self._context_timeout + return time_diff < cast(float, self._context_timeout) if self._context_timeout and within_same_context(): if state: @@ -176,24 +181,24 @@ def within_same_context() -> bool: self._count_set_off = 1 return 1 - async def process_group_write(self, telegram): + async def process_group_write(self, telegram: "Telegram") -> None: """Process incoming and outgoing GROUP WRITE telegram.""" if await self.remote_value.process(telegram, always_callback=True): self._process_reset_after() - async def process_group_response(self, telegram): + async def process_group_response(self, telegram: "Telegram") -> None: """Process incoming GroupValueResponse telegrams.""" if await self.remote_value.process(telegram, always_callback=False): self._process_reset_after() - def _process_reset_after(self): + def _process_reset_after(self) -> None: """Create Task for resetting state if 'reset_after' is configured.""" if self.reset_after is not None and self.state: if self._reset_task: self._reset_task.cancel() self._reset_task = asyncio.create_task(self._reset_state(self.reset_after)) - async def _reset_state(self, wait_seconds: float): + async def _reset_state(self, wait_seconds: float) -> None: await asyncio.sleep(wait_seconds) await self._set_internal_state(False) @@ -205,7 +210,7 @@ def is_off(self) -> bool: """Return if binary sensor is 'off'.""" return not self.state - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.name, self.remote_value.group_addr_str(), self.state diff --git a/xknx/devices/climate.py b/xknx/devices/climate.py index e1e37278cd..28cc79e409 100644 --- a/xknx/devices/climate.py +++ b/xknx/devices/climate.py @@ -6,6 +6,7 @@ """ from enum import Enum import logging +from typing import TYPE_CHECKING, Any, Iterator, Optional, Union, cast from xknx.remote_value import ( RemoteValueSetpointShift, @@ -14,7 +15,13 @@ ) from .climate_mode import ClimateMode -from .device import Device +from .device import Device, DeviceCallbackType + +if TYPE_CHECKING: + from xknx.remote_value import RemoteValue + from xknx.telegram import Telegram + from xknx.telegram.address import GroupAddress, GroupAddressableType + from xknx.xknx import XKNX logger = logging.getLogger("xknx.log") @@ -38,24 +45,24 @@ class Climate(Device): # pylint: disable=too-many-instance-attributes,invalid-name def __init__( self, - xknx, - name, - group_address_temperature=None, - group_address_target_temperature=None, - group_address_target_temperature_state=None, - group_address_setpoint_shift=None, - group_address_setpoint_shift_state=None, - setpoint_shift_mode=DEFAULT_SETPOINT_SHIFT_MODE, - setpoint_shift_max=DEFAULT_SETPOINT_SHIFT_MAX, - setpoint_shift_min=DEFAULT_SETPOINT_SHIFT_MIN, - temperature_step=DEFAULT_TEMPERATURE_STEP, - group_address_on_off=None, - group_address_on_off_state=None, - on_off_invert=False, - min_temp=None, - max_temp=None, - mode=None, - device_updated_cb=None, + xknx: "XKNX", + name: str, + group_address_temperature: Optional["GroupAddressableType"] = None, + group_address_target_temperature: Optional["GroupAddressableType"] = None, + group_address_target_temperature_state: Optional["GroupAddressableType"] = None, + group_address_setpoint_shift: Optional["GroupAddressableType"] = None, + group_address_setpoint_shift_state: Optional["GroupAddressableType"] = None, + setpoint_shift_mode: SetpointShiftMode = DEFAULT_SETPOINT_SHIFT_MODE, + setpoint_shift_max: float = DEFAULT_SETPOINT_SHIFT_MAX, + setpoint_shift_min: float = DEFAULT_SETPOINT_SHIFT_MIN, + temperature_step: float = DEFAULT_TEMPERATURE_STEP, + group_address_on_off: Optional["GroupAddressableType"] = None, + group_address_on_off_state: Optional["GroupAddressableType"] = None, + on_off_invert: bool = False, + min_temp: Optional[float] = None, + max_temp: Optional[float] = None, + mode: Optional[ClimateMode] = None, + device_updated_cb: Optional[DeviceCallbackType] = None, ): """Initialize Climate class.""" # pylint: disable=too-many-arguments, too-many-locals, too-many-branches, too-many-statements @@ -84,6 +91,7 @@ def __init__( after_update_cb=self.after_update, ) + self._setpoint_shift: Union[RemoteValueTemp, RemoteValueSetpointShift] if setpoint_shift_mode == SetpointShiftMode.DPT9002: self._setpoint_shift = RemoteValueTemp( xknx, @@ -117,7 +125,7 @@ def __init__( self.mode = mode - def _iter_remote_values(self): + def _iter_remote_values(self) -> Iterator["RemoteValue"]: """Iterate the devices RemoteValue classes.""" yield from ( self.temperature, @@ -127,7 +135,7 @@ def _iter_remote_values(self): ) @classmethod - def from_config(cls, xknx, name, config): + def from_config(cls, xknx: "XKNX", name: str, config: Any) -> "Climate": """Initialize object from configuration structure.""" # pylint: disable=too-many-locals group_address_temperature = config.get("group_address_temperature") @@ -160,7 +168,7 @@ def from_config(cls, xknx, name, config): climate_mode = None if "mode" in config: climate_mode = ClimateMode.from_config( - xknx=xknx, name=None, config=config["mode"] + xknx=xknx, name=f"{name}_mode", config=config["mode"] ) return cls( @@ -183,28 +191,28 @@ def from_config(cls, xknx, name, config): mode=climate_mode, ) - def has_group_address(self, group_address): + def has_group_address(self, group_address: "GroupAddress") -> bool: """Test if device has given group address.""" if self.mode is not None and self.mode.has_group_address(group_address): return True return super().has_group_address(group_address) @property - def is_on(self): + def is_on(self) -> bool: """Return power status.""" # None will return False return bool(self.on.value) - async def turn_on(self): + async def turn_on(self) -> None: """Set power status to on.""" await self.on.on() - async def turn_off(self): + async def turn_off(self) -> None: """Set power status to off.""" await self.on.off() @property - def initialized_for_setpoint_shift_calculations(self): + def initialized_for_setpoint_shift_calculations(self) -> bool: """Test if object is initialized for setpoint shift calculations.""" if not self._setpoint_shift.initialized: return False @@ -216,9 +224,10 @@ def initialized_for_setpoint_shift_calculations(self): return False return True - async def set_target_temperature(self, target_temperature): + async def set_target_temperature(self, target_temperature: float) -> None: """Send new target temperature or setpoint_shift to KNX bus.""" - if self.initialized_for_setpoint_shift_calculations: + if self.base_temperature is not None: + # implies initialized_for_setpoint_shift_calculations temperature_delta = target_temperature - self.base_temperature await self.set_setpoint_shift(temperature_delta) else: @@ -228,24 +237,26 @@ async def set_target_temperature(self, target_temperature): await self.target_temperature.set(validated_temp) @property - def base_temperature(self): + def base_temperature(self) -> Optional[float]: """ - Return the base temperature. + Return the base temperature when setpoint_shift is initialized. Base temperature is the default temperature (setpoint-shift=0) for the active climate mode. As this value is usually not available via KNX, we have to derive this from the current target temperature and the current set point shift. """ if self.initialized_for_setpoint_shift_calculations: - return self.target_temperature.value - self.setpoint_shift + return cast(float, self.target_temperature.value - self.setpoint_shift) return None @property - def setpoint_shift(self): + def setpoint_shift(self) -> Optional[float]: """Return current offset from base temperature in Kelvin.""" - return self._setpoint_shift.value + return self._setpoint_shift.value # type: ignore - def validate_value(self, value, min_value, max_value): + def validate_value( + self, value: float, min_value: Optional[float], max_value: Optional[float] + ) -> float: """Check boundaries of temperature and return valid temperature value.""" if (min_value is not None) and (value < min_value): logger.warning("Min value exceeded at %s: %s", self.name, value) @@ -255,7 +266,7 @@ def validate_value(self, value, min_value, max_value): return max_value return value - async def set_setpoint_shift(self, offset): + async def set_setpoint_shift(self, offset: float) -> None: """Send new temperature offset to KNX bus.""" validated_offset = self.validate_value( offset, self.setpoint_shift_min, self.setpoint_shift_max @@ -267,37 +278,39 @@ async def set_setpoint_shift(self, offset): await self.target_temperature.set(base_temperature + validated_offset) @property - def target_temperature_max(self): + def target_temperature_max(self) -> Optional[float]: """Return the highest possible target temperature.""" if self.max_temp is not None: return self.max_temp - if self.initialized_for_setpoint_shift_calculations: + if self.base_temperature is not None: + # implies initialized_for_setpoint_shift_calculations return self.base_temperature + self.setpoint_shift_max return None @property - def target_temperature_min(self): + def target_temperature_min(self) -> Optional[float]: """Return the lowest possible target temperature.""" if self.min_temp is not None: return self.min_temp - if self.initialized_for_setpoint_shift_calculations: + if self.base_temperature is not None: + # implies initialized_for_setpoint_shift_calculations return self.base_temperature + self.setpoint_shift_min return None - async def process_group_write(self, telegram): + async def process_group_write(self, telegram: "Telegram") -> None: """Process incoming and outgoing GROUP WRITE telegram.""" for remote_value in self._iter_remote_values(): await remote_value.process(telegram) if self.mode is not None: await self.mode.process_group_write(telegram) - async def sync(self, wait_for_result=False): + async def sync(self, wait_for_result: bool = False) -> None: """Read states of device from KNX bus.""" await super().sync(wait_for_result=wait_for_result) if self.mode is not None: await self.mode.sync(wait_for_result=wait_for_result) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ( ' "ClimateMode": """Initialize object from configuration structure.""" # pylint: disable=too-many-locals group_address_operation_mode = config.get("group_address_operation_mode") @@ -216,7 +232,9 @@ def from_config(cls, xknx, name, config): group_address_heat_cool_state=group_address_heat_cool_state, ) - def _iter_remote_values(self): + def _iter_remote_values( + self, + ) -> Iterator["RemoteValue"]: """Iterate climate mode RemoteValue classes.""" return chain( self._iter_byte_operation_modes(), @@ -224,21 +242,25 @@ def _iter_remote_values(self): self._iter_binary_operation_modes(), ) - def _iter_byte_operation_modes(self): + def _iter_byte_operation_modes( + self, + ) -> Iterator[RemoteValueClimateMode[HVACOperationMode]]: """Iterate normal DPT 20.102 operation mode remote values.""" yield from ( self.remote_value_operation_mode, self.remote_value_controller_status, ) - def _iter_controller_remote_values(self): + def _iter_controller_remote_values( + self, + ) -> Iterator[RemoteValueClimateModeBase[HVACControllerMode]]: """Iterate DPT 20.105 controller remote values.""" yield from ( self.remote_value_controller_mode, self.remote_value_heat_cool, ) - def _iter_binary_operation_modes(self): + def _iter_binary_operation_modes(self) -> Iterator[RemoteValueBinaryOperationMode]: """Iterate DPT 1 binary operation modes.""" yield from ( self.remote_value_operation_mode_comfort, @@ -247,28 +269,33 @@ def _iter_binary_operation_modes(self): self.remote_value_operation_mode_standby, ) - async def _set_internal_operation_mode(self, operation_mode): + async def _set_internal_operation_mode( + self, operation_mode: HVACOperationMode + ) -> None: """Set internal value of operation mode. Call hooks if operation mode was changed.""" if operation_mode != self.operation_mode: self.operation_mode = operation_mode await self.after_update() - async def _set_internal_controller_mode(self, controller_mode): + async def _set_internal_controller_mode( + self, controller_mode: HVACControllerMode + ) -> None: """Set internal value of controller mode. Call hooks if controller mode was changed.""" if controller_mode != self.controller_mode: self.controller_mode = controller_mode await self.after_update() - async def set_operation_mode(self, operation_mode): + async def set_operation_mode(self, operation_mode: HVACOperationMode) -> None: """Set the operation mode of a thermostat. Send new operation_mode to BUS and update internal state.""" if ( not self.supports_operation_mode or operation_mode not in self._operation_modes ): raise DeviceIllegalValue( - "operation (preset) mode not supported", operation_mode + "operation (preset) mode not supported", str(operation_mode) ) + rv: RemoteValueClimateModeBase[HVACOperationMode] for rv in chain( self._iter_byte_operation_modes(), self._iter_binary_operation_modes() ): @@ -277,16 +304,17 @@ async def set_operation_mode(self, operation_mode): await self._set_internal_operation_mode(operation_mode) - async def set_controller_mode(self, controller_mode): + async def set_controller_mode(self, controller_mode: HVACControllerMode) -> None: """Set the controller mode of a thermostat. Send new controller mode to the bus and update internal state.""" if ( not self.supports_controller_mode or controller_mode not in self._controller_modes ): raise DeviceIllegalValue( - "controller (HVAC) mode not supported", controller_mode + "controller (HVAC) mode not supported", str(controller_mode) ) + rv: RemoteValueClimateModeBase[HVACControllerMode] for rv in self._iter_controller_remote_values(): if rv.writable and controller_mode in rv.supported_operation_modes(): await rv.set(controller_mode) @@ -294,42 +322,40 @@ async def set_controller_mode(self, controller_mode): await self._set_internal_controller_mode(controller_mode) @property - def operation_modes(self): + def operation_modes(self) -> List[HVACOperationMode]: """Return all configured operation modes.""" if not self.supports_operation_mode: return [] return self._operation_modes @property - def controller_modes(self): + def controller_modes(self) -> List[HVACControllerMode]: """Return all configured controller modes.""" if not self.supports_controller_mode: return [] return self._controller_modes - def gather_operation_modes(self): + def gather_operation_modes(self) -> List[HVACOperationMode]: """Gather operation modes from RemoteValues.""" - operation_modes = [] + operation_modes: List[HVACOperationMode] = [] for rv in chain( self._iter_binary_operation_modes(), self._iter_byte_operation_modes() ): if rv.writable: operation_modes.extend(rv.supported_operation_modes()) - # remove duplicates return list(set(operation_modes)) - def gather_controller_modes(self): + def gather_controller_modes(self) -> List[HVACControllerMode]: """Gather controller modes from RemoteValues.""" - operation_modes = [] + controller_modes: List[HVACControllerMode] = [] for rv in self._iter_controller_remote_values(): if rv.writable: - operation_modes.extend(rv.supported_operation_modes()) - + controller_modes.extend(rv.supported_operation_modes()) # remove duplicates - return list(set(operation_modes)) + return list(set(controller_modes)) - async def process_group_write(self, telegram): + async def process_group_write(self, telegram: "Telegram") -> None: """Process incoming and outgoing GROUP WRITE telegram.""" if self.supports_operation_mode: for rv in self._iter_remote_values(): @@ -345,7 +371,7 @@ async def process_group_write(self, telegram): await self._set_internal_controller_mode(rv.value) return - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ( ' str: """Return object as readable string.""" return ( ' str: """Return object as readable string.""" return ''.format( self.name, self._remote_value.group_addr_str(), self._broadcast_type diff --git a/xknx/devices/expose_sensor.py b/xknx/devices/expose_sensor.py index a651cafd22..a835a865ff 100644 --- a/xknx/devices/expose_sensor.py +++ b/xknx/devices/expose_sensor.py @@ -77,7 +77,7 @@ def resolve_state(self): """Return the current state of the sensor as a human readable string.""" return self.sensor_value.value - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.name, diff --git a/xknx/devices/fan.py b/xknx/devices/fan.py index a0344a9915..c511aef6e1 100644 --- a/xknx/devices/fan.py +++ b/xknx/devices/fan.py @@ -61,7 +61,7 @@ def from_config(cls, xknx, name, config): group_address_speed_state=group_address_speed_state, ) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.name, self.speed.group_addr_str() diff --git a/xknx/devices/light.py b/xknx/devices/light.py index 91be195a17..ca9c2c7253 100644 --- a/xknx/devices/light.py +++ b/xknx/devices/light.py @@ -379,7 +379,7 @@ def from_config(cls, xknx, name, config): max_kelvin=max_kelvin, ) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" str_brightness = ( "" diff --git a/xknx/devices/notification.py b/xknx/devices/notification.py index 9acf5e426a..4e06d829a4 100644 --- a/xknx/devices/notification.py +++ b/xknx/devices/notification.py @@ -72,7 +72,7 @@ async def do(self, action): "Could not understand action %s for device %s", action, self.get_name() ) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.name, self._message.group_addr_str() diff --git a/xknx/devices/scene.py b/xknx/devices/scene.py index cd90ce3cd6..161bebb278 100644 --- a/xknx/devices/scene.py +++ b/xknx/devices/scene.py @@ -41,7 +41,7 @@ def from_config(cls, xknx, name, config): xknx, name=name, group_address=group_address, scene_number=scene_number ) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.name, self.scene_value.group_addr_str(), self.scene_number diff --git a/xknx/devices/sensor.py b/xknx/devices/sensor.py index 22cdfa3939..0119e3c53a 100644 --- a/xknx/devices/sensor.py +++ b/xknx/devices/sensor.py @@ -93,7 +93,7 @@ def resolve_state(self): """Return the current state of the sensor as a human readable string.""" return self.sensor_value.value - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.name, diff --git a/xknx/devices/switch.py b/xknx/devices/switch.py index 0c58578e51..75fd54cada 100644 --- a/xknx/devices/switch.py +++ b/xknx/devices/switch.py @@ -110,7 +110,7 @@ async def _reset_state(self, wait_seconds: float): await asyncio.sleep(wait_seconds) await self.set_off() - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format( self.name, self.switch.group_addr_str() diff --git a/xknx/dpt/__init__.py b/xknx/dpt/__init__.py index bd8cb703ee..13fb10e372 100644 --- a/xknx/dpt/__init__.py +++ b/xknx/dpt/__init__.py @@ -169,13 +169,7 @@ ) from .dpt_date import DPTDate from .dpt_datetime import DPTDateTime -from .dpt_hvac_mode import ( - DPTControllerStatus, - DPTHVACContrMode, - DPTHVACMode, - HVACControllerMode, - HVACOperationMode, -) +from .dpt_hvac_mode import DPTControllerStatus, DPTHVACContrMode, DPTHVACMode from .dpt_scaling import DPTAngle, DPTScaling from .dpt_string import DPTString from .dpt_time import DPTTime diff --git a/xknx/dpt/dpt.py b/xknx/dpt/dpt.py index c6c42abfaf..4cc3c2b975 100644 --- a/xknx/dpt/dpt.py +++ b/xknx/dpt/dpt.py @@ -141,7 +141,7 @@ def __eq__(self, other): """Equal operator.""" return DPTComparator.compare(self, other) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return f'' @@ -167,7 +167,7 @@ def __eq__(self, other): """Equal operator.""" return DPTComparator.compare(self, other) - def __str__(self): + def __str__(self) -> str: """Return object as readable string.""" return ''.format(",".join(hex(b) for b in self.value)) diff --git a/xknx/dpt/dpt_4bit_control.py b/xknx/dpt/dpt_4bit_control.py index 512336883b..6b93f3fc7d 100644 --- a/xknx/dpt/dpt_4bit_control.py +++ b/xknx/dpt/dpt_4bit_control.py @@ -179,7 +179,7 @@ class TitleEnum(Enum): Ensures values are rendered nicely, e.g. in home assistant. """ - def __str__(self): + def __str__(self) -> str: """Return string representation.""" # pylint: disable=no-member return self.name.title() diff --git a/xknx/dpt/dpt_date.py b/xknx/dpt/dpt_date.py index b6d8ab28e8..4d387469db 100644 --- a/xknx/dpt/dpt_date.py +++ b/xknx/dpt/dpt_date.py @@ -52,7 +52,7 @@ def _knx_year(year): return (value.tm_mday, value.tm_mon, _knx_year(value.tm_year)) @staticmethod - def _test_range(day, month, year): + def _test_range(day: int, month: int, year: int) -> bool: """Test if the values are in the correct range.""" if day < 1 or day > 31: return False diff --git a/xknx/remote_value/remote_value.py b/xknx/remote_value/remote_value.py index 1fffbe9bf6..319f36152a 100644 --- a/xknx/remote_value/remote_value.py +++ b/xknx/remote_value/remote_value.py @@ -111,15 +111,11 @@ def _internal_addresses() -> Iterator[Optional[GroupAddress]]: return group_address in _internal_addresses() + @staticmethod @abstractmethod # TODO: typing - remove Optional - def payload_valid(self, payload: Optional["DPTPayload"]) -> bool: + def payload_valid(payload: Optional["DPTPayload"]) -> bool: """Test if telegram payload may be parsed - to be implemented in derived class..""" - # pylint: disable=unused-argument - logger.warning( - "'payload_valid()' not implemented for %s", self.__class__.__name__ - ) - return True @abstractmethod def from_knx(self, payload: "DPTPayload") -> Any: @@ -144,7 +140,7 @@ async def process(self, telegram: Telegram, always_callback: bool = False) -> bo ): raise CouldNotParseTelegram( "payload not a GroupValueWrite or GroupValueResponse", - payload=telegram.payload, + payload=str(telegram.payload), destination_address=str(telegram.destination_address), source_address=str(telegram.source_address), device_name=self.device_name, diff --git a/xknx/remote_value/remote_value_climate_mode.py b/xknx/remote_value/remote_value_climate_mode.py index 6b9868c385..017319d95d 100644 --- a/xknx/remote_value/remote_value_climate_mode.py +++ b/xknx/remote_value/remote_value_climate_mode.py @@ -3,8 +3,20 @@ DPT . """ +from abc import abstractmethod from enum import Enum -from typing import List +from typing import ( + TYPE_CHECKING, + Any, + Awaitable, + Callable, + Generic, + List, + Optional, + TypeVar, + Union, + cast, +) from xknx.dpt import ( DPTArray, @@ -12,15 +24,33 @@ DPTControllerStatus, DPTHVACContrMode, DPTHVACMode, - HVACOperationMode, ) +from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACOperationMode from xknx.exceptions import ConversionError, CouldNotParseTelegram -from ..dpt.dpt_hvac_mode import HVACControllerMode from .remote_value import RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX + + AsyncCallback = Callable[[], Awaitable[None]] + DPTPayload = Union[DPTArray, DPTBinary] + +HVACModeType = TypeVar("HVACModeType", "HVACControllerMode", "HVACOperationMode") + + +class RemoteValueClimateModeBase(RemoteValue, Generic[HVACModeType]): + """Base class for binary climate mode remote values.""" + + @abstractmethod + def supported_operation_modes( + self, + ) -> List["HVACModeType"]: + """Return a list of all supported operation modes.""" + -class RemoteValueClimateMode(RemoteValue): +class RemoteValueClimateMode(RemoteValueClimateModeBase[HVACModeType]): """Abstraction for remote value of KNX climate modes.""" class ClimateModeType(Enum): @@ -32,15 +62,15 @@ class ClimateModeType(Enum): def __init__( self, - xknx, - group_address=None, - group_address_state=None, - sync_state=True, - device_name=None, - feature_name="Climate Mode", - climate_mode_type=None, - after_update_cb=None, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + sync_state: bool = True, + device_name: Optional[str] = None, + feature_name: str = "Climate Mode", + climate_mode_type: Optional[ClimateModeType] = None, + after_update_cb: Optional["AsyncCallback"] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX climate mode.""" # pylint: disable=too-many-arguments @@ -57,57 +87,62 @@ def __init__( if not isinstance(climate_mode_type, self.ClimateModeType): raise ConversionError( "invalid climate mode type", - climate_mode_type=climate_mode_type, - device_name=device_name, + climate_mode_type=str(climate_mode_type), + device_name=str(device_name), feature_name=feature_name, ) self._climate_mode_transcoder = climate_mode_type.value - def supported_operation_modes(self): + def supported_operation_modes(self) -> List["HVACModeType"]: """Return a list of all supported operation modes.""" return list(self._climate_mode_transcoder.SUPPORTED_MODES.values()) - def payload_valid(self, payload): + @staticmethod + def payload_valid(payload: Optional["DPTPayload"]) -> bool: """Test if telegram payload may be parsed.""" return isinstance(payload, DPTArray) and len(payload.value) == 1 - def to_knx(self, value): + def to_knx(self, value: Any) -> "DPTPayload": """Convert value to payload.""" return DPTArray(self._climate_mode_transcoder.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: "DPTPayload") -> Optional[HVACModeType]: """Convert current payload to value.""" - return self._climate_mode_transcoder.from_knx(payload.value) + # TODO: typing - remove cast + return cast( + Optional[HVACModeType], + self._climate_mode_transcoder.from_knx(payload.value), + ) -class _RemoteValueBinaryClimateMode(RemoteValue): - """Base class for binary climate mode remote values.""" +class RemoteValueBinaryOperationMode(RemoteValueClimateModeBase[HVACOperationMode]): + """Abstraction for remote value of split up KNX climate modes.""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - sync_state=True, - device_name=None, - feature_name="Climate Mode Binary", - after_update_cb=None, - operation_mode=None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + sync_state: bool = True, + device_name: Optional[str] = None, + feature_name: str = "Climate Mode Binary", + after_update_cb: Optional["AsyncCallback"] = None, + operation_mode: Optional[HVACOperationMode] = None, ): """Initialize remote value of KNX DPT 1 representing a climate operation mode.""" # pylint: disable=too-many-arguments if not isinstance(operation_mode, HVACOperationMode): raise ConversionError( "Invalid operation mode type", - operation_mode=operation_mode, - device_name=device_name, + operation_mode=str(operation_mode), + device_name=str(device_name), feature_name=feature_name, ) if operation_mode not in self.supported_operation_modes(): raise ConversionError( "Operation mode not supported for binary mode object", - operation_mode=operation_mode, - device_name=device_name, + operation_mode=str(operation_mode), + device_name=str(device_name), feature_name=feature_name, ) self.operation_mode = operation_mode @@ -122,15 +157,11 @@ def __init__( ) @staticmethod - def supported_operation_modes(): - """Return a list of the configured operation mode.""" - raise NotImplementedError("supported_operation_modes has to return a list") - - def payload_valid(self, payload): + def payload_valid(payload: Optional["DPTPayload"]) -> bool: """Test if telegram payload may be parsed.""" return isinstance(payload, DPTBinary) - def to_knx(self, value): + def to_knx(self, value: Any) -> "DPTPayload": """Convert value to payload.""" if isinstance(value, HVACOperationMode): # foreign operation modes will set the RemoteValue to False @@ -142,12 +173,8 @@ def to_knx(self, value): feature_name=self.feature_name, ) - -class RemoteValueBinaryOperationMode(_RemoteValueBinaryClimateMode): - """Abstraction for remote value of split up KNX climate modes.""" - @staticmethod - def supported_operation_modes(): + def supported_operation_modes() -> List[HVACOperationMode]: """Return a list of the configured operation mode.""" return [ HVACOperationMode.COMFORT, @@ -156,7 +183,7 @@ def supported_operation_modes(): HVACOperationMode.STANDBY, ] - def from_knx(self, payload): + def from_knx(self, payload: "DPTPayload") -> Optional[HVACOperationMode]: """Convert current payload to value.""" if payload == DPTBinary(1): return self.operation_mode @@ -164,40 +191,40 @@ def from_knx(self, payload): return None raise CouldNotParseTelegram( "payload invalid", - payload=payload, + payload=str(payload), device_name=self.device_name, feature_name=self.feature_name, ) -class RemoteValueBinaryHeatCool(RemoteValue): +class RemoteValueBinaryHeatCool(RemoteValueClimateModeBase[HVACControllerMode]): """Abstraction for remote value of heat/cool controller mode.""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - sync_state=True, - device_name=None, - feature_name="Controller Mode Heat/Cool", - after_update_cb=None, - controller_mode=None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + sync_state: bool = True, + device_name: Optional[str] = None, + feature_name: str = "Controller Mode Heat/Cool", + after_update_cb: Optional["AsyncCallback"] = None, + controller_mode: Optional[HVACControllerMode] = None, ): """Initialize remote value of KNX DPT 1 representing a climate controller mode.""" # pylint: disable=too-many-arguments if not isinstance(controller_mode, HVACControllerMode): raise ConversionError( "Invalid controller mode type", - controller_mode=controller_mode, - device_name=device_name, + controller_mode=str(controller_mode), + device_name=str(device_name), feature_name=feature_name, ) if controller_mode not in self.supported_operation_modes(): raise ConversionError( "Controller mode not supported for binary mode object", - controller_mode=controller_mode, - device_name=device_name, + controller_mode=str(controller_mode), + device_name=str(device_name), feature_name=feature_name, ) self.controller_mode = controller_mode @@ -211,16 +238,17 @@ def __init__( after_update_cb=after_update_cb, ) - def payload_valid(self, payload): + @staticmethod + def payload_valid(payload: Optional["DPTPayload"]) -> bool: """Test if telegram payload may be parsed.""" return isinstance(payload, DPTBinary) @staticmethod - def supported_operation_modes(): + def supported_operation_modes() -> List[HVACControllerMode]: """Return a list of the configured operation mode.""" return [HVACControllerMode.HEAT, HVACControllerMode.COOL] - def to_knx(self, value): + def to_knx(self, value: Any) -> "DPTPayload": """Convert value to payload.""" if isinstance(value, HVACControllerMode): # foreign operation modes will set the RemoteValue to False @@ -232,7 +260,7 @@ def to_knx(self, value): feature_name=self.feature_name, ) - def from_knx(self, payload): + def from_knx(self, payload: "DPTPayload") -> Optional[HVACControllerMode]: """Convert current payload to value.""" if payload == DPTBinary(1): return self.controller_mode @@ -248,7 +276,7 @@ def from_knx(self, payload): ) raise CouldNotParseTelegram( "payload invalid", - payload=payload, + payload=str(payload), device_name=self.device_name, feature_name=self.feature_name, ) diff --git a/xknx/remote_value/remote_value_switch.py b/xknx/remote_value/remote_value_switch.py index 4351eff9bd..0fe284742a 100644 --- a/xknx/remote_value/remote_value_switch.py +++ b/xknx/remote_value/remote_value_switch.py @@ -68,11 +68,11 @@ def from_knx(self, payload): feature_name=self.feature_name, ) - async def off(self): + async def off(self) -> None: """Set value to OFF.""" await self.set(False) - async def on(self): + async def on(self) -> None: """Set value to ON.""" # pylint: disable=invalid-name await self.set(True) diff --git a/xknx/telegram/telegram.py b/xknx/telegram/telegram.py index b702c2333f..d6c44321ed 100644 --- a/xknx/telegram/telegram.py +++ b/xknx/telegram/telegram.py @@ -14,9 +14,10 @@ """ from enum import Enum -from typing import Any, Union +from typing import Optional, Union from .address import GroupAddress, IndividualAddress +from .apci import APCI class TelegramDirection(Enum): @@ -35,7 +36,7 @@ def __init__( self, destination_address: Union[GroupAddress, IndividualAddress] = GroupAddress(0), direction: TelegramDirection = TelegramDirection.OUTGOING, - payload: Any = None, + payload: Optional[APCI] = None, source_address: IndividualAddress = IndividualAddress(0), ) -> None: """Initialize Telegram class."""