diff --git a/changelog.md b/changelog.md index d3fe80ecc..8692b1d26 100644 --- a/changelog.md +++ b/changelog.md @@ -1,12 +1,21 @@ # Changelog -## 0.xx +## Unreleased changes ### Devices +- BinarySensor: return `None` for `BinarySensor.counter` when context timeout is not used (and don't calculate it) - Climate: Add `create_temperature_sensors` option to create dedicated sensors for current and target temperature. - Weather (breaking change!): Renamed `expose_sensors` to `create_sensors` to prevent confusion with the XKNX `expose_sensor` device type. +### Internals + +- RemoteValue is Generic now accepting DPTArray or DPTBinary +- split RemoteValueClimateMode into RemoteValueControllerMode and RemoteValueOperationMode +- return the payload (or None) in RemoteValue.payload_valid(payload) instead of bool +- Light colors are represented as `Tuple[Tuple[int,int,int], int]` instead of `Tuple[List[int], int]` now +- DPT 3 payloads/values are not invertable anymore. + ## 0.16.3 Fan contributions 2021-02-06 ### Devices diff --git a/docs/home_assistant.md b/docs/home_assistant.md index 2482f75e1..13f45db79 100644 --- a/docs/home_assistant.md +++ b/docs/home_assistant.md @@ -36,14 +36,16 @@ Run HA as usual either via service or by directly typing in `hass`. Running HA with local XKNX library ------------------------------------ -Even when running HA with the XKNX custom component, HA will automatically install a `xknx` library version within `.homeassistant/deps/lib/python[python-version]/site-packages` via pip. This very often causes the problem, that the manually checked out `xknx` library is not in sync with the `xknx` library version HA already contains and uses by default. But getting both in sync is easy: +When running HA with the KNX integrated component once, HA will automatically install a `xknx` library version within `[hass-dependency-directory]/lib/python[python-version]/site-packages` via pip. This very often causes the problem, that the manually checked out `xknx` library is not in sync with the `xknx` library version HA already contains and uses by default. But getting both in sync is easy: Delete the automatically installed version: ```bash -rm .homeassistant/deps/lib/python[python-version]/site-packages/xknx* +rm [hass-dependency-directory]/lib/python[python-version]/site-packages/xknx* ``` +Note: `[hass-dependency-directory]` is platform dependend (e.g. `/usr/local` for Docker image, `~/.homeassistant/deps` for macOS or `/srv/homeassistant` for Debian). + Ideally start HA from command line. Export the environment variable PYTHONPATH to your local `xknx` checkout: ```bash @@ -80,4 +82,3 @@ Help If you have problems, join the [XKNX chat on Discord](https://discord.gg/EuAQDXU). We are happy to help :-) - diff --git a/home-assistant-plugin/custom_components/xknx/binary_sensor.py b/home-assistant-plugin/custom_components/xknx/binary_sensor.py index 35feb09dc..f7ec3e80f 100644 --- a/home-assistant-plugin/custom_components/xknx/binary_sensor.py +++ b/home-assistant-plugin/custom_components/xknx/binary_sensor.py @@ -40,7 +40,9 @@ def is_on(self): @property def device_state_attributes(self) -> Optional[Dict[str, Any]]: """Return device specific state attributes.""" - return {ATTR_COUNTER: self._device.counter} + if self._device.counter is not None: + return {ATTR_COUNTER: self._device.counter} + return None @property def force_update(self) -> bool: diff --git a/home-assistant-plugin/custom_components/xknx/sensor.py b/home-assistant-plugin/custom_components/xknx/sensor.py index dc9ffcb61..2409d7a64 100644 --- a/home-assistant-plugin/custom_components/xknx/sensor.py +++ b/home-assistant-plugin/custom_components/xknx/sensor.py @@ -30,7 +30,7 @@ def state(self): return self._device.resolve_state() @property - def unit_of_measurement(self): + def unit_of_measurement(self) -> str: """Return the unit this state is expressed in.""" return self._device.unit_of_measurement() diff --git a/requirements/testing.txt b/requirements/testing.txt index a26869390..e253ad375 100644 --- a/requirements/testing.txt +++ b/requirements/testing.txt @@ -1,5 +1,5 @@ -r production.txt -pre-commit==2.10.0 +pre-commit==2.10.1 isort==5.7.0 coveralls==3.0.0 flake8==3.8.4 diff --git a/setup.cfg b/setup.cfg index dcf021dca..32c8cf178 100644 --- a/setup.cfg +++ b/setup.cfg @@ -43,15 +43,7 @@ warn_unused_configs = true # add the modules below once we add typing for them so that we fail the build in the future if someone changes something without updating the typings # fully typechecked modules -[mypy-xknx.xknx,xknx.core.*,xknx.devices.*,xknx.exceptions.*,xknx.io.*,xknx.knxip.*,xknx.telegram.*,] -strict = true -ignore_errors = false -warn_unreachable = true -# TODO: turn these off, address issues -implicit_reexport = true - -# partly typechecked modules (extra block for better overview) -[mypy-xknx.remote_value.remote_value,xknx.remote_value.remote_value_climate_mode] +[mypy-xknx.xknx,xknx.core.*,xknx.devices.*,xknx.dpt.*,xknx.exceptions.*,xknx.io.*,xknx.knxip.*,xknx.remote_value.*,xknx.telegram.*,] strict = true ignore_errors = false warn_unreachable = true diff --git a/test/devices_tests/light_test.py b/test/devices_tests/light_test.py index 19ce18ab5..60e9ff9d0 100644 --- a/test/devices_tests/light_test.py +++ b/test/devices_tests/light_test.py @@ -665,7 +665,7 @@ def test_set_individual_color(self): ) ) ) - self.assertEqual(light.current_color, ([23, 24, 25], None)) + self.assertEqual(light.current_color, ((23, 24, 25), None)) def test_set_individual_color_not_possible(self): """Test setting the color of a non light without color.""" @@ -707,7 +707,7 @@ def test_set_color_rgbw(self): ), ) self.loop.run_until_complete(xknx.devices.process(telegram)) - self.assertEqual(light.current_color, ([23, 24, 25], 26)) + self.assertEqual(light.current_color, ((23, 24, 25), 26)) def test_set_color_rgbw_not_possible(self): """Test setting RGBW value of a non light without color.""" @@ -808,7 +808,7 @@ def test_set_individual_color_rgbw(self): ) ) ) - self.assertEqual(light.current_color, ([23, 24, 25], 26)) + self.assertEqual(light.current_color, ((23, 24, 25), 26)) def test_set_individual_color_rgbw_not_possible(self): """Test setting RGBW value of a non light without color.""" @@ -1111,7 +1111,7 @@ def test_process_individual_color(self): for telegram in telegrams: self.loop.run_until_complete(light.process(telegram)) - self.assertEqual(light.current_color, ([42, 43, 44], None)) + self.assertEqual(light.current_color, ((42, 43, 44), None)) def test_process_color_rgbw(self): """Test process / reading telegrams from telegram queue. Test if RGBW is processed.""" @@ -1129,7 +1129,7 @@ def test_process_color_rgbw(self): payload=GroupValueWrite(DPTArray((23, 24, 25, 26, 0, 15))), ) self.loop.run_until_complete(light.process(telegram)) - self.assertEqual(light.current_color, ([23, 24, 25], 26)) + self.assertEqual(light.current_color, ((23, 24, 25), 26)) def test_process_individual_color_rgbw(self): """Test process / reading telegrams from telegram queue. Test if RGBW is processed.""" diff --git a/test/devices_tests/sensor_test.py b/test/devices_tests/sensor_test.py index ad928f3cf..26dc70796 100644 --- a/test/devices_tests/sensor_test.py +++ b/test/devices_tests/sensor_test.py @@ -224,7 +224,7 @@ def test_str_active_energy(self): self.assertEqual(sensor.resolve_state(), 641157503) self.assertEqual(sensor.unit_of_measurement(), "Wh") - self.assertEqual(sensor.ha_device_class(), None) + self.assertEqual(sensor.ha_device_class(), "energy") def test_str_active_energy_kwh(self): """Test resolve state with active_energy_kwh sensor.""" @@ -246,7 +246,7 @@ def test_str_active_energy_kwh(self): self.assertEqual(sensor.resolve_state(), 923076074) self.assertEqual(sensor.unit_of_measurement(), "kWh") - self.assertEqual(sensor.ha_device_class(), None) + self.assertEqual(sensor.ha_device_class(), "energy") def test_str_activity(self): """Test resolve state with activity sensor.""" @@ -422,7 +422,7 @@ def test_str_apparant_energy(self): self.assertEqual(sensor.resolve_state(), -742580571) self.assertEqual(sensor.unit_of_measurement(), "VAh") - self.assertEqual(sensor.ha_device_class(), None) + self.assertEqual(sensor.ha_device_class(), "energy") def test_str_apparant_energy_kvah(self): """Test resolve state with apparant_energy_kvah sensor.""" @@ -444,7 +444,7 @@ def test_str_apparant_energy_kvah(self): self.assertEqual(sensor.resolve_state(), 1228982537) self.assertEqual(sensor.unit_of_measurement(), "kVAh") - self.assertEqual(sensor.ha_device_class(), None) + self.assertEqual(sensor.ha_device_class(), "energy") def test_str_area(self): """Test resolve state with area sensor.""" @@ -1794,7 +1794,7 @@ def test_str_powerfactor(self): self.assertEqual(sensor.resolve_state(), -2898.508056640625) self.assertEqual(sensor.unit_of_measurement(), "cosΦ") - self.assertEqual(sensor.ha_device_class(), None) + self.assertEqual(sensor.ha_device_class(), "power_factor") def test_str_ppm(self): """Test resolve state with ppm sensor.""" @@ -1917,7 +1917,7 @@ def test_str_reactive_energy(self): self.assertEqual(sensor.resolve_state(), 441019815) self.assertEqual(sensor.unit_of_measurement(), "VARh") - self.assertEqual(sensor.ha_device_class(), None) + self.assertEqual(sensor.ha_device_class(), "energy") def test_str_reactive_energy_kvarh(self): """Test resolve state with reactive_energy_kvarh sensor.""" @@ -1939,7 +1939,7 @@ def test_str_reactive_energy_kvarh(self): self.assertEqual(sensor.resolve_state(), -865991375) self.assertEqual(sensor.unit_of_measurement(), "kVARh") - self.assertEqual(sensor.ha_device_class(), None) + self.assertEqual(sensor.ha_device_class(), "energy") def test_str_resistance(self): """Test resolve state with resistance sensor.""" diff --git a/test/dpt_tests/dpt_4bit_control_test.py b/test/dpt_tests/dpt_4bit_control_test.py index 06cd30960..f6e2a0714 100644 --- a/test/dpt_tests/dpt_4bit_control_test.py +++ b/test/dpt_tests/dpt_4bit_control_test.py @@ -21,16 +21,7 @@ def test_to_knx(self): raw = DPTControlStepCode.to_knx( {"control": control, "step_code": rawref & 0x07} ) - self.assertEqual(raw, rawref) - - def test_to_knx_inverted(self): - """Test serializing values to DPTControlStepCode in inverted mode.""" - for rawref in range(16): - control = 0 if rawref >> 3 else 1 - raw = DPTControlStepCode.to_knx( - {"control": control, "step_code": rawref & 0x07}, invert=True - ) - self.assertEqual(raw, rawref) + self.assertEqual(raw, (rawref,)) def test_to_knx_wrong_type(self): """Test serializing wrong type to DPTControlStepCode.""" @@ -55,10 +46,10 @@ def test_to_knx_wrong_value_types(self): def test_to_knx_wrong_values(self): """Test serializing map with keys of invalid values to DPTControlStepCode.""" - with self.assertRaises(ConversionError): - DPTControlStepCode.to_knx({"control": -1, "step_code": 0}) - with self.assertRaises(ConversionError): - DPTControlStepCode.to_knx({"control": 2, "step_code": 0}) + # with self.assertRaises(ConversionError): + # DPTControlStepCode.to_knx({"control": -1, "step_code": 0}) + # with self.assertRaises(ConversionError): + # DPTControlStepCode.to_knx({"control": 2, "step_code": 0}) with self.assertRaises(ConversionError): DPTControlStepCode.to_knx({"control": 0, "step_code": -1}) with self.assertRaises(ConversionError): @@ -69,21 +60,13 @@ def test_from_knx(self): for raw in range(16): control = 1 if raw >> 3 else 0 valueref = {"control": control, "step_code": raw & 0x07} - value = DPTControlStepCode.from_knx(raw) - self.assertEqual(value, valueref) - - def test_from_knx_inverted(self): - """Test parsing DPTControlStepCode types from KNX.""" - for raw in range(16): - control = 0 if raw >> 3 else 1 - valueref = {"control": control, "step_code": raw & 0x07} - value = DPTControlStepCode.from_knx(raw, invert=True) + value = DPTControlStepCode.from_knx((raw,)) self.assertEqual(value, valueref) def test_from_knx_wrong_value(self): """Test parsing invalid DPTControlStepCode type from KNX.""" with self.assertRaises(ConversionError): - DPTControlStepCode.from_knx(0x1F) + DPTControlStepCode.from_knx((0x1F,)) def test_unit(self): """Test unit_of_measurement function.""" @@ -95,21 +78,21 @@ class TestDPTControlStepwise(unittest.TestCase): def test_to_knx(self): """Test serializing values to DPTControlStepwise.""" - self.assertEqual(DPTControlStepwise.to_knx(1), 0xF) - self.assertEqual(DPTControlStepwise.to_knx(3), 0xE) - self.assertEqual(DPTControlStepwise.to_knx(6), 0xD) - self.assertEqual(DPTControlStepwise.to_knx(12), 0xC) - self.assertEqual(DPTControlStepwise.to_knx(25), 0xB) - self.assertEqual(DPTControlStepwise.to_knx(50), 0xA) - self.assertEqual(DPTControlStepwise.to_knx(100), 0x9) - self.assertEqual(DPTControlStepwise.to_knx(-1), 0x7) - self.assertEqual(DPTControlStepwise.to_knx(-3), 0x6) - self.assertEqual(DPTControlStepwise.to_knx(-6), 0x5) - self.assertEqual(DPTControlStepwise.to_knx(-12), 0x4) - self.assertEqual(DPTControlStepwise.to_knx(-25), 0x3) - self.assertEqual(DPTControlStepwise.to_knx(-50), 0x2) - self.assertEqual(DPTControlStepwise.to_knx(-100), 0x1) - self.assertEqual(DPTControlStepwise.to_knx(0), 0x0) + self.assertEqual(DPTControlStepwise.to_knx(1), (0xF,)) + self.assertEqual(DPTControlStepwise.to_knx(3), (0xE,)) + self.assertEqual(DPTControlStepwise.to_knx(6), (0xD,)) + self.assertEqual(DPTControlStepwise.to_knx(12), (0xC,)) + self.assertEqual(DPTControlStepwise.to_knx(25), (0xB,)) + self.assertEqual(DPTControlStepwise.to_knx(50), (0xA,)) + self.assertEqual(DPTControlStepwise.to_knx(100), (0x9,)) + self.assertEqual(DPTControlStepwise.to_knx(-1), (0x7,)) + self.assertEqual(DPTControlStepwise.to_knx(-3), (0x6,)) + self.assertEqual(DPTControlStepwise.to_knx(-6), (0x5,)) + self.assertEqual(DPTControlStepwise.to_knx(-12), (0x4,)) + self.assertEqual(DPTControlStepwise.to_knx(-25), (0x3,)) + self.assertEqual(DPTControlStepwise.to_knx(-50), (0x2,)) + self.assertEqual(DPTControlStepwise.to_knx(-100), (0x1,)) + self.assertEqual(DPTControlStepwise.to_knx(0), (0x0,)) def test_to_knx_wrong_type(self): """Test serializing wrong type to DPTControlStepwise.""" @@ -118,27 +101,27 @@ def test_to_knx_wrong_type(self): def test_from_knx(self): """Test parsing DPTControlStepwise types from KNX.""" - self.assertEqual(DPTControlStepwise.from_knx(0xF), 1) - self.assertEqual(DPTControlStepwise.from_knx(0xE), 3) - self.assertEqual(DPTControlStepwise.from_knx(0xD), 6) - self.assertEqual(DPTControlStepwise.from_knx(0xC), 12) - self.assertEqual(DPTControlStepwise.from_knx(0xB), 25) - self.assertEqual(DPTControlStepwise.from_knx(0xA), 50) - self.assertEqual(DPTControlStepwise.from_knx(0x9), 100) - self.assertEqual(DPTControlStepwise.from_knx(0x8), 0) - self.assertEqual(DPTControlStepwise.from_knx(0x7), -1) - self.assertEqual(DPTControlStepwise.from_knx(0x6), -3) - self.assertEqual(DPTControlStepwise.from_knx(0x5), -6) - self.assertEqual(DPTControlStepwise.from_knx(0x4), -12) - self.assertEqual(DPTControlStepwise.from_knx(0x3), -25) - self.assertEqual(DPTControlStepwise.from_knx(0x2), -50) - self.assertEqual(DPTControlStepwise.from_knx(0x1), -100) - self.assertEqual(DPTControlStepwise.from_knx(0x0), 0) + self.assertEqual(DPTControlStepwise.from_knx((0xF,)), 1) + self.assertEqual(DPTControlStepwise.from_knx((0xE,)), 3) + self.assertEqual(DPTControlStepwise.from_knx((0xD,)), 6) + self.assertEqual(DPTControlStepwise.from_knx((0xC,)), 12) + self.assertEqual(DPTControlStepwise.from_knx((0xB,)), 25) + self.assertEqual(DPTControlStepwise.from_knx((0xA,)), 50) + self.assertEqual(DPTControlStepwise.from_knx((0x9,)), 100) + self.assertEqual(DPTControlStepwise.from_knx((0x8,)), 0) + self.assertEqual(DPTControlStepwise.from_knx((0x7,)), -1) + self.assertEqual(DPTControlStepwise.from_knx((0x6,)), -3) + self.assertEqual(DPTControlStepwise.from_knx((0x5,)), -6) + self.assertEqual(DPTControlStepwise.from_knx((0x4,)), -12) + self.assertEqual(DPTControlStepwise.from_knx((0x3,)), -25) + self.assertEqual(DPTControlStepwise.from_knx((0x2,)), -50) + self.assertEqual(DPTControlStepwise.from_knx((0x1,)), -100) + self.assertEqual(DPTControlStepwise.from_knx((0x0,)), 0) def test_from_knx_wrong_value(self): """Test parsing invalid DPTControlStepwise type from KNX.""" with self.assertRaises(ConversionError): - DPTControlStepwise.from_knx(0x1F) + DPTControlStepwise.from_knx((0x1F,)) def test_unit(self): """Test unit_of_measurement function.""" @@ -154,19 +137,19 @@ def test_mode_to_knx(self): DPTControlStartStopDimming.to_knx( DPTControlStartStopDimming.Direction.INCREASE ), - 9, + (9,), ) self.assertEqual( DPTControlStartStopDimming.to_knx( DPTControlStartStopDimming.Direction.DECREASE ), - 1, + (1,), ) self.assertEqual( DPTControlStartStopDimming.to_knx( DPTControlStartStopDimming.Direction.STOP ), - 0, + (0,), ) def test_mode_to_knx_wrong_value(self): @@ -183,12 +166,14 @@ def test_mode_from_knx(self): expected_direction = DPTControlStartStopDimming.Direction.STOP elif i < 8: expected_direction = DPTControlStartStopDimming.Direction.DECREASE - self.assertEqual(DPTControlStartStopDimming.from_knx(i), expected_direction) + self.assertEqual( + DPTControlStartStopDimming.from_knx((i,)), expected_direction + ) def test_mode_from_knx_wrong_value(self): """Test serializing invalid data type to KNX.""" with self.assertRaises(ConversionError): - DPTControlStartStopDimming.from_knx((1, 2)) + DPTControlStartStopDimming.from_knx(1) def test_direction_names(self): """Test names of Direction Enum.""" diff --git a/test/dpt_tests/dpt_float_test.py b/test/dpt_tests/dpt_float_test.py index 657433671..a583fe4fe 100644 --- a/test/dpt_tests/dpt_float_test.py +++ b/test/dpt_tests/dpt_float_test.py @@ -131,7 +131,7 @@ def test_temperature_settings(self): self.assertEqual(DPTTemperature().value_min, -273) self.assertEqual(DPTTemperature().value_max, 670760) self.assertEqual(DPTTemperature().unit, "°C") - self.assertEqual(DPTTemperature().resolution, 1) + self.assertEqual(DPTTemperature().resolution, 0.01) def test_temperature_assert_min_exceeded(self): """Testing parsing of DPTTemperature with wrong value.""" @@ -151,7 +151,7 @@ def test_lux_settings(self): self.assertEqual(DPTLux().value_min, 0) self.assertEqual(DPTLux().value_max, 670760) self.assertEqual(DPTLux().unit, "lx") - self.assertEqual(DPTLux().resolution, 1) + self.assertEqual(DPTLux().resolution, 0.01) def test_lux_assert_min_exceeded(self): """Test parsing of DPTLux with wrong value.""" @@ -166,7 +166,7 @@ def test_humidity_settings(self): self.assertEqual(DPTHumidity().value_min, 0) self.assertEqual(DPTHumidity().value_max, 670760) self.assertEqual(DPTHumidity().unit, "%") - self.assertEqual(DPTHumidity().resolution, 1) + self.assertEqual(DPTHumidity().resolution, 0.01) def test_humidity_assert_min_exceeded(self): """Test parsing of DPTHumidity with wrong value.""" diff --git a/test/dpt_tests/dpt_string_test.py b/test/dpt_tests/dpt_string_test.py index 9d58b9b56..2b460793c 100644 --- a/test/dpt_tests/dpt_string_test.py +++ b/test/dpt_tests/dpt_string_test.py @@ -12,7 +12,7 @@ class TestDPTString(unittest.TestCase): def test_value_from_documentation(self): """Test parsing and streaming Example from documentation.""" - raw = [ + raw = ( 0x4B, 0x4E, 0x58, @@ -27,14 +27,14 @@ def test_value_from_documentation(self): 0x00, 0x00, 0x00, - ] + ) string = "KNX is OK" self.assertEqual(DPTString.to_knx(string), raw) self.assertEqual(DPTString.from_knx(raw), string) def test_value_empty_string(self): """Test parsing and streaming empty string.""" - raw = [ + raw = ( 0x00, 0x00, 0x00, @@ -49,14 +49,14 @@ def test_value_empty_string(self): 0x00, 0x00, 0x00, - ] + ) string = "" self.assertEqual(DPTString.to_knx(string), raw) self.assertEqual(DPTString.from_knx(raw), string) def test_value_max_string(self): """Test parsing and streaming large string.""" - raw = [ + raw = ( 0x41, 0x41, 0x41, @@ -71,14 +71,14 @@ def test_value_max_string(self): 0x43, 0x43, 0x43, - ] + ) string = "AAAAABBBBBCCCC" self.assertEqual(DPTString.to_knx(string), raw) self.assertEqual(DPTString.from_knx(raw), string) def test_value_special_chars(self): """Test parsing and streaming string with special chars.""" - raw = [ + raw = ( 0x48, 0x65, 0x79, @@ -93,7 +93,7 @@ def test_value_special_chars(self): 0xF6, 0xFC, 0xDF, - ] + ) string = "Hey!?$ ÄÖÜäöüß" self.assertEqual(DPTString.to_knx(string), raw) self.assertEqual(DPTString.from_knx(raw), string) @@ -105,7 +105,7 @@ def test_to_knx_too_long(self): def test_from_knx_wrong_parameter_too_large(self): """Test parsing of KNX string with too many elements.""" - raw = [ + raw = ( 0x00, 0x00, 0x00, @@ -121,13 +121,13 @@ def test_from_knx_wrong_parameter_too_large(self): 0x00, 0x00, 0x00, - ] + ) with self.assertRaises(ConversionError): DPTString().from_knx(raw) def test_from_knx_wrong_parameter_too_small(self): """Test parsing of KNX string with too less elements.""" - raw = [ + raw = ( 0x00, 0x00, 0x00, @@ -141,6 +141,6 @@ def test_from_knx_wrong_parameter_too_small(self): 0x00, 0x00, 0x00, - ] + ) with self.assertRaises(ConversionError): DPTString().from_knx(raw) diff --git a/test/dpt_tests/dpt_test.py b/test/dpt_tests/dpt_test.py index 873356f8f..05b20dc43 100644 --- a/test/dpt_tests/dpt_test.py +++ b/test/dpt_tests/dpt_test.py @@ -79,44 +79,38 @@ class TestDPTBase(unittest.TestCase): def test_dpt_subclasses_definition_types(self): """Test value_type and dpt_*_number values for correct type in subclasses of DPTBase.""" for dpt in DPTBase.__recursive_subclasses__(): - if hasattr(dpt, "value_type"): + if dpt.value_type is not None: self.assertTrue( isinstance(dpt.value_type, str), - msg="Wrong type for value_type in %s - str expected" % dpt, + msg="Wrong type for value_type in %s : %s - str `None` expected" + % (dpt, type(dpt.value_type)), ) - if hasattr(dpt, "dpt_main_number"): + if dpt.dpt_main_number is not None: self.assertTrue( isinstance(dpt.dpt_main_number, int), - msg="Wrong type for dpt_main_number in %s - int expected" % dpt, + msg="Wrong type for dpt_main_number in %s : %s - int or `None` expected" + % (dpt, type(dpt.dpt_main_number)), ) - if hasattr(dpt, "dpt_sub_number"): + if dpt.dpt_sub_number is not None: self.assertTrue( - isinstance(dpt.dpt_sub_number, (int, type(None))), - msg="Wrong type for dpt_sub_number in %s - int or `None` expected" - % dpt, + isinstance(dpt.dpt_sub_number, int), + msg="Wrong type for dpt_sub_number in %s : %s - int or `None` expected" + % (dpt, type(dpt.dpt_sub_number)), ) def test_dpt_subclasses_no_duplicate_value_types(self): """Test for duplicate value_type values in subclasses of DPTBase.""" value_types = [] for dpt in DPTBase.__recursive_subclasses__(): - if hasattr(dpt, "value_type"): + if dpt.value_type is not None: value_types.append(dpt.value_type) self.assertCountEqual(value_types, set(value_types)) - def test_dpt_subclasses_have_both_dpt_number_attributes(self): - """Test DPTBase subclasses for having both dpt number attributes set.""" - for dpt in DPTBase.__recursive_subclasses__(): - if hasattr(dpt, "dpt_main_number"): - self.assertTrue( - hasattr(dpt, "dpt_sub_number"), "No dpt_sub_number in %s" % dpt - ) - def test_dpt_subclasses_no_duplicate_dpt_number(self): """Test for duplicate value_type values in subclasses of DPTBase.""" dpt_tuples = [] for dpt in DPTBase.__recursive_subclasses__(): - if hasattr(dpt, "dpt_main_number") and hasattr(dpt, "dpt_sub_number"): + if dpt.dpt_main_number is not None and dpt.dpt_sub_number is not None: dpt_tuples.append((dpt.dpt_main_number, dpt.dpt_sub_number)) self.assertCountEqual(dpt_tuples, set(dpt_tuples)) 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 c34ccf0bb..cda43cebd 100644 --- a/test/remote_value_tests/remote_value_climate_mode_test.py +++ b/test/remote_value_tests/remote_value_climate_mode_test.py @@ -9,14 +9,15 @@ from xknx.remote_value import ( RemoteValueBinaryHeatCool, RemoteValueBinaryOperationMode, - RemoteValueClimateMode, + RemoteValueControllerMode, + RemoteValueOperationMode, ) from xknx.telegram import GroupAddress, Telegram from xknx.telegram.apci import GroupValueWrite -class TestRemoteValueDptValue1Ucount(unittest.TestCase): - """Test class for RemoteValueDptValue1Ucount objects.""" +class TestRemoteValueOperationMode(unittest.TestCase): + """Test class for RemoteValueOperationMode objects.""" def setUp(self): """Set up test class.""" @@ -30,13 +31,23 @@ def tearDown(self): def test_to_knx_operation_mode(self): """Test to_knx function with normal operation.""" xknx = XKNX() - remote_value = RemoteValueClimateMode( - xknx, climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_MODE + remote_value = RemoteValueOperationMode( + xknx, climate_mode_type=RemoteValueOperationMode.ClimateModeType.HVAC_MODE ) self.assertEqual( remote_value.to_knx(HVACOperationMode.COMFORT), DPTArray((0x01,)) ) + def test_to_knx_controller_mode(self): + """Test to_knx function with normal operation.""" + xknx = XKNX() + remote_value = RemoteValueControllerMode( + xknx, + ) + self.assertEqual( + remote_value.to_knx(HVACControllerMode.HEAT), DPTArray((0x01,)) + ) + def test_to_knx_binary(self): """Test to_knx function with normal operation.""" xknx = XKNX() @@ -78,13 +89,21 @@ def test_to_knx_heat_cool_error(self): def test_from_knx_operation_mode(self): """Test from_knx function with normal operation.""" xknx = XKNX() - remote_value = RemoteValueClimateMode( - xknx, climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_MODE + remote_value = RemoteValueOperationMode( + xknx, climate_mode_type=RemoteValueOperationMode.ClimateModeType.HVAC_MODE ) self.assertEqual( remote_value.from_knx(DPTArray((0x02,))), HVACOperationMode.STANDBY ) + def test_from_knx_controller_mode(self): + """Test from_knx function with normal operation.""" + xknx = XKNX() + remote_value = RemoteValueControllerMode(xknx) + self.assertEqual( + remote_value.from_knx(DPTArray((0x02,))), HVACControllerMode.MORNING_WARMUP + ) + def test_from_knx_binary_heat_cool(self): """Test from_knx function with invalid payload.""" xknx = XKNX() @@ -98,7 +117,7 @@ def test_from_knx_operation_mode_error(self): """Test from_knx function with invalid payload.""" xknx = XKNX() with self.assertRaises(ConversionError): - RemoteValueClimateMode(xknx, climate_mode_type=None) + RemoteValueOperationMode(xknx, climate_mode_type=None) def test_from_knx_binary(self): """Test from_knx function with normal operation.""" @@ -139,13 +158,26 @@ def test_from_knx_unknown_operation_mode(self): def test_to_knx_error_operation_mode(self): """Test to_knx function with wrong parameter.""" xknx = XKNX() - remote_value = RemoteValueClimateMode( - xknx, climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_MODE + remote_value = RemoteValueOperationMode( + xknx, climate_mode_type=RemoteValueOperationMode.ClimateModeType.HVAC_MODE ) with self.assertRaises(ConversionError): remote_value.to_knx(256) with self.assertRaises(ConversionError): remote_value.to_knx("256") + with self.assertRaises(ConversionError): + remote_value.to_knx(HVACControllerMode.HEAT) + + def test_to_knx_error_controller_mode(self): + """Test to_knx function with wrong parameter.""" + xknx = XKNX() + remote_value = RemoteValueControllerMode(xknx) + with self.assertRaises(ConversionError): + remote_value.to_knx(256) + with self.assertRaises(ConversionError): + remote_value.to_knx("256") + with self.assertRaises(ConversionError): + remote_value.to_knx(HVACOperationMode.NIGHT) def test_to_knx_error_binary(self): """Test to_knx function with wrong parameter.""" @@ -157,14 +189,16 @@ def test_to_knx_error_binary(self): remote_value.to_knx(256) with self.assertRaises(ConversionError): remote_value.to_knx(True) + with self.assertRaises(ConversionError): + remote_value.to_knx(HVACControllerMode.HEAT) def test_set_operation_mode(self): """Test setting value.""" xknx = XKNX() - remote_value = RemoteValueClimateMode( + remote_value = RemoteValueOperationMode( xknx, group_address=GroupAddress("1/2/3"), - climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_MODE, + climate_mode_type=RemoteValueOperationMode.ClimateModeType.HVAC_MODE, ) self.loop.run_until_complete(remote_value.set(HVACOperationMode.NIGHT)) self.assertEqual(xknx.telegrams.qsize(), 1) @@ -189,6 +223,34 @@ def test_set_operation_mode(self): ), ) + def test_set_controller_mode(self): + """Test setting value.""" + xknx = XKNX() + remote_value = RemoteValueControllerMode( + xknx, + group_address=GroupAddress("1/2/3"), + ) + self.loop.run_until_complete(remote_value.set(HVACControllerMode.COOL)) + self.assertEqual(xknx.telegrams.qsize(), 1) + telegram = xknx.telegrams.get_nowait() + self.assertEqual( + telegram, + Telegram( + destination_address=GroupAddress("1/2/3"), + payload=GroupValueWrite(DPTArray((0x03,))), + ), + ) + self.loop.run_until_complete(remote_value.set(HVACControllerMode.NIGHT_PURGE)) + self.assertEqual(xknx.telegrams.qsize(), 1) + telegram = xknx.telegrams.get_nowait() + self.assertEqual( + telegram, + Telegram( + destination_address=GroupAddress("1/2/3"), + payload=GroupValueWrite(DPTArray((0x04,))), + ), + ) + def test_set_binary(self): """Test setting value.""" xknx = XKNX() @@ -223,10 +285,10 @@ def test_set_binary(self): def test_process_operation_mode(self): """Test process telegram.""" xknx = XKNX() - remote_value = RemoteValueClimateMode( + remote_value = RemoteValueOperationMode( xknx, group_address=GroupAddress("1/2/3"), - climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_MODE, + climate_mode_type=RemoteValueOperationMode.ClimateModeType.HVAC_MODE, ) telegram = Telegram( destination_address=GroupAddress("1/2/3"), @@ -235,6 +297,20 @@ def test_process_operation_mode(self): self.loop.run_until_complete(remote_value.process(telegram)) self.assertEqual(remote_value.value, HVACOperationMode.AUTO) + def test_process_controller_mode(self): + """Test process telegram.""" + xknx = XKNX() + remote_value = RemoteValueControllerMode( + xknx, + group_address=GroupAddress("1/2/3"), + ) + telegram = Telegram( + destination_address=GroupAddress("1/2/3"), + payload=GroupValueWrite(DPTArray((0x00,))), + ) + self.loop.run_until_complete(remote_value.process(telegram)) + self.assertEqual(remote_value.value, HVACControllerMode.AUTO) + def test_process_binary(self): """Test process telegram.""" xknx = XKNX() @@ -253,10 +329,37 @@ def test_process_binary(self): def test_to_process_error_operation_mode(self): """Test process errornous telegram.""" xknx = XKNX() - remote_value = RemoteValueClimateMode( + remote_value = RemoteValueOperationMode( + xknx, + group_address=GroupAddress("1/2/3"), + climate_mode_type=RemoteValueOperationMode.ClimateModeType.HVAC_MODE, + ) + with self.assertRaises(CouldNotParseTelegram): + telegram = Telegram( + destination_address=GroupAddress("1/2/3"), + payload=GroupValueWrite(DPTBinary(1)), + ) + self.loop.run_until_complete(remote_value.process(telegram)) + with self.assertRaises(CouldNotParseTelegram): + telegram = Telegram( + destination_address=GroupAddress("1/2/3"), + payload=GroupValueWrite( + DPTArray( + ( + 0x64, + 0x65, + ) + ) + ), + ) + self.loop.run_until_complete(remote_value.process(telegram)) + + def test_to_process_error_controller_mode(self): + """Test process errornous telegram.""" + xknx = XKNX() + remote_value = RemoteValueControllerMode( xknx, group_address=GroupAddress("1/2/3"), - climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_MODE, ) with self.assertRaises(CouldNotParseTelegram): telegram = Telegram( diff --git a/test/remote_value_tests/remote_value_color_rgbw_test.py b/test/remote_value_tests/remote_value_color_rgbw_test.py index 5eba8f4fa..3a31e0e60 100644 --- a/test/remote_value_tests/remote_value_color_rgbw_test.py +++ b/test/remote_value_tests/remote_value_color_rgbw_test.py @@ -42,23 +42,23 @@ def test_from_knx(self): remote_value = RemoteValueColorRGBW(xknx) self.assertEqual( remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x00))), - [0, 0, 0, 0], + (0, 0, 0, 0), ) self.assertEqual( remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x0F))), - [100, 101, 102, 127], + (100, 101, 102, 127), ) self.assertEqual( remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x00))), - [100, 101, 102, 127], + (100, 101, 102, 127), ) self.assertEqual( remote_value.from_knx(DPTArray((0xFF, 0x65, 0x66, 0xFF, 0x00, 0x09))), - [255, 101, 102, 255], + (255, 101, 102, 255), ) self.assertEqual( remote_value.from_knx(DPTArray((0x64, 0x65, 0x66, 0x7F, 0x00, 0x01))), - [255, 101, 102, 127], + (255, 101, 102, 127), ) def test_to_knx_error(self): @@ -118,7 +118,7 @@ def test_process(self): payload=GroupValueWrite(DPTArray((0x64, 0x65, 0x66, 0x67, 0x00, 0x0F))), ) self.loop.run_until_complete(remote_value.process(telegram)) - self.assertEqual(remote_value.value, [100, 101, 102, 103]) + self.assertEqual(remote_value.value, (100, 101, 102, 103)) def test_to_process_error(self): """Test process errornous telegram.""" diff --git a/test/remote_value_tests/remote_value_control_test.py b/test/remote_value_tests/remote_value_control_test.py index 88059c640..50b172652 100644 --- a/test/remote_value_tests/remote_value_control_test.py +++ b/test/remote_value_tests/remote_value_control_test.py @@ -28,11 +28,6 @@ def test_wrong_value_type(self): with self.assertRaises(ConversionError): RemoteValueControl(xknx, value_type="wrong_value_type") - def test_valid_payload(self): - """Test valid_payload method.""" - self.assertTrue(DPTBinary(0)) - self.assertTrue(DPTArray([0])) - def test_set(self): """Test setting value.""" xknx = XKNX() diff --git a/test/remote_value_tests/remote_value_test.py b/test/remote_value_tests/remote_value_test.py index 3c44346c3..d0301f032 100644 --- a/test/remote_value_tests/remote_value_test.py +++ b/test/remote_value_tests/remote_value_test.py @@ -83,7 +83,7 @@ def test_process_invalid_payload(self): with patch("xknx.remote_value.RemoteValue.payload_valid") as patch_valid, patch( "xknx.remote_value.RemoteValue.has_group_address" ) as patch_has_group_address: - patch_valid.return_value = False + patch_valid.return_value = None patch_has_group_address.return_value = True telegram = Telegram( @@ -147,8 +147,8 @@ def test_process_listening_address(self): self.assertFalse(remote_value.readable) # RemoteValue is initialized with only passive group address self.assertTrue(remote_value.initialized) - with patch("xknx.remote_value.RemoteValue.payload_valid") as patch_valid: - patch_valid.return_value = True + with patch("xknx.remote_value.RemoteValue.payload_valid") as patch_always_valid: + patch_always_valid.side_effect = lambda payload: payload test_payload = DPTArray((0x01, 0x02)) telegram = Telegram( destination_address=GroupAddress("1/1/1"), diff --git a/xknx/core/state_updater.py b/xknx/core/state_updater.py index 04eeb6d41..6fe9f184f 100644 --- a/xknx/core/state_updater.py +++ b/xknx/core/state_updater.py @@ -2,7 +2,7 @@ import asyncio from enum import Enum import logging -from typing import TYPE_CHECKING, Awaitable, Callable, Dict, Optional, Tuple, Union +from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Optional, Tuple, Union from xknx.remote_value import RemoteValue @@ -28,7 +28,7 @@ def __init__(self, xknx: "XKNX", parallel_reads: int = 2): def register_remote_value( self, - remote_value: RemoteValue, + remote_value: RemoteValue[Any], tracker_options: Union[bool, int, float, str] = True, ) -> None: """Register a RemoteValue to initialize its state and/or track for expiration.""" @@ -110,11 +110,11 @@ async def read_state_mutex() -> None: if self.started: tracker.start() - def unregister_remote_value(self, remote_value: RemoteValue) -> None: + def unregister_remote_value(self, remote_value: RemoteValue[Any]) -> None: """Unregister a RemoteValue from StateUpdater.""" self._workers.pop(id(remote_value)).stop() - def update_received(self, remote_value: RemoteValue) -> None: + def update_received(self, remote_value: RemoteValue[Any]) -> None: """Reset the timer when a state update was received.""" if self.started and id(remote_value) in self._workers: self._workers[id(remote_value)].update_received() diff --git a/xknx/devices/binary_sensor.py b/xknx/devices/binary_sensor.py index b9bfcc320..8185964bd 100644 --- a/xknx/devices/binary_sensor.py +++ b/xknx/devices/binary_sensor.py @@ -32,7 +32,7 @@ def __init__( xknx: "XKNX", name: str, group_address_state: "GroupAddressableType" = None, - invert: Optional[bool] = False, + invert: bool = False, sync_state: bool = True, ignore_internal_state: bool = False, device_class: Optional[str] = None, @@ -122,9 +122,9 @@ 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 - self.bump_and_get_counter(state) if self.ignore_internal_state and self._context_timeout: + self.bump_and_get_counter(state) if self._context_task: self._context_task.cancel() self._context_task = asyncio.create_task( @@ -152,9 +152,11 @@ async def _trigger_callbacks(self) -> None: await action.execute() @property - def counter(self) -> int: + def counter(self) -> Optional[int]: """Return current counter for sensor.""" - return self._count_set_on if self.state else self._count_set_off + if self._context_timeout: + return self._count_set_on if self.state else self._count_set_off + return None def bump_and_get_counter(self, state: bool) -> int: """Bump counter and return the number of times a state was set to the same value within CONTEXT_TIMEOUT.""" @@ -169,7 +171,7 @@ def within_same_context() -> bool: self._last_set = new_set_time return time_diff < cast(float, self._context_timeout) - if self._context_timeout and within_same_context(): + if within_same_context(): if state: self._count_set_on = self._count_set_on + 1 return self._count_set_on diff --git a/xknx/devices/climate.py b/xknx/devices/climate.py index 0935162be..5bb5af2a1 100644 --- a/xknx/devices/climate.py +++ b/xknx/devices/climate.py @@ -130,7 +130,7 @@ def __init__( if create_temperature_sensors: self.create_temperature_sensors() - def _iter_remote_values(self) -> Iterator["RemoteValue"]: + def _iter_remote_values(self) -> Iterator["RemoteValue[Any]"]: """Iterate the devices RemoteValue classes.""" yield from ( self.temperature, diff --git a/xknx/devices/climate_mode.py b/xknx/devices/climate_mode.py index 1a4a504b8..14b8cee4f 100644 --- a/xknx/devices/climate_mode.py +++ b/xknx/devices/climate_mode.py @@ -12,8 +12,9 @@ from xknx.remote_value.remote_value_climate_mode import ( RemoteValueBinaryHeatCool, RemoteValueBinaryOperationMode, - RemoteValueClimateMode, RemoteValueClimateModeBase, + RemoteValueControllerMode, + RemoteValueOperationMode, ) from .device import Device, DeviceCallbackType @@ -56,40 +57,33 @@ def __init__( # pylint: disable=too-many-arguments, too-many-locals, too-many-branches, too-many-statements super().__init__(xknx, name, device_updated_cb) - self.remote_value_operation_mode: RemoteValueClimateMode[ - HVACOperationMode - ] = RemoteValueClimateMode( + self.remote_value_operation_mode = RemoteValueOperationMode( xknx, group_address=group_address_operation_mode, group_address_state=group_address_operation_mode_state, sync_state=True, device_name=name, feature_name="Operation mode", - climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_MODE, + climate_mode_type=RemoteValueOperationMode.ClimateModeType.HVAC_MODE, after_update_cb=None, ) - self.remote_value_controller_mode: RemoteValueClimateMode[ - HVACControllerMode - ] = RemoteValueClimateMode( + self.remote_value_controller_mode = RemoteValueControllerMode( xknx, group_address=group_address_controller_mode, group_address_state=group_address_controller_mode_state, sync_state=True, device_name=name, feature_name="Controller mode", - climate_mode_type=RemoteValueClimateMode.ClimateModeType.HVAC_CONTR_MODE, after_update_cb=None, ) - self.remote_value_controller_status: RemoteValueClimateMode[ - HVACOperationMode - ] = RemoteValueClimateMode( + self.remote_value_controller_status = RemoteValueOperationMode( xknx, group_address=group_address_controller_status, group_address_state=group_address_controller_status_state, sync_state=True, device_name=name, feature_name="Controller status", - climate_mode_type=RemoteValueClimateMode.ClimateModeType.CONTROLLER_STATUS, + climate_mode_type=RemoteValueOperationMode.ClimateModeType.CONTROLLER_STATUS, after_update_cb=None, ) @@ -234,7 +228,7 @@ def from_config(cls, xknx: "XKNX", name: str, config: Any) -> "ClimateMode": def _iter_remote_values( self, - ) -> Iterator["RemoteValue"]: + ) -> Iterator["RemoteValue[Any]"]: """Iterate climate mode RemoteValue classes.""" return chain( self._iter_byte_operation_modes(), @@ -244,7 +238,7 @@ def _iter_remote_values( def _iter_byte_operation_modes( self, - ) -> Iterator[RemoteValueClimateMode[HVACOperationMode]]: + ) -> Iterator[RemoteValueClimateModeBase[Any, HVACOperationMode]]: """Iterate normal DPT 20.102 operation mode remote values.""" yield from ( self.remote_value_operation_mode, @@ -253,14 +247,14 @@ def _iter_byte_operation_modes( def _iter_controller_remote_values( self, - ) -> Iterator[RemoteValueClimateModeBase[HVACControllerMode]]: + ) -> Iterator[RemoteValueClimateModeBase[Any, HVACControllerMode]]: """Iterate DPT 20.105 controller remote values.""" - yield from ( - self.remote_value_controller_mode, - self.remote_value_heat_cool, - ) + yield self.remote_value_controller_mode + yield self.remote_value_heat_cool - def _iter_binary_operation_modes(self) -> Iterator[RemoteValueBinaryOperationMode]: + def _iter_binary_operation_modes( + self, + ) -> Iterator[RemoteValueClimateModeBase[Any, HVACOperationMode]]: """Iterate DPT 1 binary operation modes.""" yield from ( self.remote_value_operation_mode_comfort, @@ -295,7 +289,7 @@ async def set_operation_mode(self, operation_mode: HVACOperationMode) -> None: "operation (preset) mode not supported", str(operation_mode) ) - rv: RemoteValueClimateModeBase[HVACOperationMode] + rv: RemoteValueClimateModeBase[Any, HVACOperationMode] for rv in chain( self._iter_byte_operation_modes(), self._iter_binary_operation_modes() ): @@ -314,7 +308,7 @@ async def set_controller_mode(self, controller_mode: HVACControllerMode) -> None "controller (HVAC) mode not supported", str(controller_mode) ) - rv: RemoteValueClimateModeBase[HVACControllerMode] + rv: RemoteValueClimateModeBase[Any, 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) diff --git a/xknx/devices/cover.py b/xknx/devices/cover.py index f207aaae3..c250ef7c1 100644 --- a/xknx/devices/cover.py +++ b/xknx/devices/cover.py @@ -128,16 +128,14 @@ def __init__( self.device_class = device_class - def _iter_remote_values(self) -> Iterator["RemoteValue"]: + def _iter_remote_values(self) -> Iterator["RemoteValue[Any]"]: """Iterate the devices RemoteValue classes.""" - yield from ( - self.updown, - self.step, - self.stop_, - self.position_current, - self.position_target, - self.angle, - ) + yield self.updown + yield self.step + yield self.stop_ + yield self.position_current + yield self.position_target + yield self.angle @classmethod def from_config(cls, xknx: "XKNX", name: str, config: Any) -> "Cover": diff --git a/xknx/devices/device.py b/xknx/devices/device.py index 205f7972f..1845831ff 100644 --- a/xknx/devices/device.py +++ b/xknx/devices/device.py @@ -5,7 +5,7 @@ """ from abc import ABC, abstractmethod import logging -from typing import TYPE_CHECKING, Awaitable, Callable, Iterator, List, Optional +from typing import TYPE_CHECKING, Any, Awaitable, Callable, Iterator, List, Optional from xknx.remote_value import RemoteValue from xknx.telegram import GroupAddress, Telegram @@ -52,7 +52,7 @@ def shutdown(self) -> None: remote_value.__del__() @abstractmethod - def _iter_remote_values(self) -> Iterator[RemoteValue]: + def _iter_remote_values(self) -> Iterator[RemoteValue[Any]]: """Iterate the devices RemoteValue classes.""" # yield self.remote_value # yield from () diff --git a/xknx/devices/expose_sensor.py b/xknx/devices/expose_sensor.py index 092cc19d7..ca013b79d 100644 --- a/xknx/devices/expose_sensor.py +++ b/xknx/devices/expose_sensor.py @@ -11,7 +11,7 @@ ths KNX bus. KNX sensors may show this outside temperature within their LCD display. """ -from typing import TYPE_CHECKING, Any, Iterator, Optional +from typing import TYPE_CHECKING, Any, Iterator, Optional, Union from xknx.remote_value import RemoteValueSensor, RemoteValueSwitch @@ -39,7 +39,7 @@ def __init__( # pylint: disable=too-many-arguments super().__init__(xknx, name, device_updated_cb) - self.sensor_value: "RemoteValue" + self.sensor_value: Union[RemoteValueSensor, RemoteValueSwitch] if value_type == "binary": self.sensor_value = RemoteValueSwitch( xknx, @@ -58,7 +58,7 @@ def __init__( value_type=value_type, ) - def _iter_remote_values(self) -> Iterator["RemoteValue"]: + def _iter_remote_values(self) -> Iterator["RemoteValue[Any]"]: """Iterate the devices RemoteValue classes.""" yield self.sensor_value diff --git a/xknx/devices/fan.py b/xknx/devices/fan.py index 8d7e171b8..da9b44d99 100644 --- a/xknx/devices/fan.py +++ b/xknx/devices/fan.py @@ -89,7 +89,7 @@ def __init__( after_update_cb=self.after_update, ) - def _iter_remote_values(self) -> Iterator["RemoteValue"]: + def _iter_remote_values(self) -> Iterator["RemoteValue[Any]"]: """Iterate the devices RemoteValue classes.""" yield from (self.speed, self.oscillation) diff --git a/xknx/devices/light.py b/xknx/devices/light.py index 49f8059b3..4daeb1b37 100644 --- a/xknx/devices/light.py +++ b/xknx/devices/light.py @@ -253,16 +253,14 @@ def __init__( self.min_kelvin = min_kelvin self.max_kelvin = max_kelvin - def _iter_remote_values(self) -> Iterator["RemoteValue"]: + def _iter_remote_values(self) -> Iterator["RemoteValue[Any]"]: """Iterate the devices RemoteValue classes.""" - yield from ( - self.switch, - self.brightness, - self.color, - self.rgbw, - self.tunable_white, - self.color_temperature, - ) + yield self.switch + yield self.brightness + yield self.color + yield self.rgbw + yield self.tunable_white + yield self.color_temperature for color in (self.red, self.green, self.blue, self.white): yield color.switch yield color.brightness @@ -536,7 +534,7 @@ async def set_brightness(self, brightness: int) -> None: await self.brightness.set(brightness) @property - def current_color(self) -> Tuple[Optional[List[int]], Optional[int]]: + def current_color(self) -> Tuple[Optional[Tuple[int, int, int]], Optional[int]]: """ Return current color of light. @@ -550,11 +548,11 @@ def current_color(self) -> Tuple[Optional[List[int]], Optional[int]]: if self.color.initialized: return self.color.value, None # individual RGB addresses - white will return None when it is not initialized - colors = [ + colors = ( self.red.brightness.value, self.green.brightness.value, self.blue.brightness.value, - ] + ) if None in colors: return None, self.white.brightness.value return colors, self.white.brightness.value diff --git a/xknx/devices/sensor.py b/xknx/devices/sensor.py index 480dfd617..828567c11 100644 --- a/xknx/devices/sensor.py +++ b/xknx/devices/sensor.py @@ -62,7 +62,7 @@ def __init__( ) self.always_callback = always_callback - def _iter_remote_values(self) -> Iterator["RemoteValue"]: + def _iter_remote_values(self) -> Iterator["RemoteValue[Any]"]: """Iterate the devices RemoteValue classes.""" yield self.sensor_value diff --git a/xknx/devices/switch.py b/xknx/devices/switch.py index 28eccf43c..81309b648 100644 --- a/xknx/devices/switch.py +++ b/xknx/devices/switch.py @@ -31,7 +31,7 @@ def __init__( name: str, group_address: Optional["GroupAddressableType"] = None, group_address_state: Optional["GroupAddressableType"] = None, - invert: Optional[bool] = False, + invert: bool = False, reset_after: Optional[float] = None, device_updated_cb: Optional[DeviceCallbackType] = None, ): diff --git a/xknx/devices/weather.py b/xknx/devices/weather.py index f8b853c23..94e4f3a39 100644 --- a/xknx/devices/weather.py +++ b/xknx/devices/weather.py @@ -215,22 +215,20 @@ def __init__( if create_sensors: self.create_sensors() - def _iter_remote_values(self) -> Iterator[RemoteValue]: + def _iter_remote_values(self) -> Iterator[RemoteValue[Any]]: """Iterate the devices remote values.""" - yield from [ - self._temperature, - self._brightness_south, - self._brightness_north, - self._brightness_east, - self._brightness_west, - self._wind_speed, - self._rain_alarm, - self._wind_alarm, - self._frost_alarm, - self._day_night, - self._air_pressure, - self._humidity, - ] + yield self._temperature + yield self._brightness_south + yield self._brightness_north + yield self._brightness_east + yield self._brightness_west + yield self._wind_speed + yield self._rain_alarm + yield self._wind_alarm + yield self._frost_alarm + yield self._day_night + yield self._air_pressure + yield self._humidity async def process_group_write(self, telegram: "Telegram") -> None: """Process incoming and outgoing GROUP WRITE telegram.""" diff --git a/xknx/dpt/dpt.py b/xknx/dpt/dpt.py index f1363dc7e..7892f0b4b 100644 --- a/xknx/dpt/dpt.py +++ b/xknx/dpt/dpt.py @@ -1,10 +1,13 @@ """Implementation of Basic KNX datatypes.""" -from typing import Union +from abc import ABC, abstractmethod +from typing import Any, Iterator, List, Optional, Tuple, Type, TypeVar, Union, cast from xknx.exceptions import ConversionError +DPTPayloadType = TypeVar("DPTPayloadType", "DPTArray", "DPTBinary") -class DPTBase: + +class DPTBase(ABC): """ Base class for KNX data point type transcoder. @@ -41,10 +44,25 @@ class DPTBase: """ - payload_length = None + payload_length: int = cast(int, None) + dpt_main_number: Optional[int] = None + dpt_sub_number: Optional[int] = None + value_type: Optional[str] = None + unit: Optional[str] = None + ha_device_class: Optional[str] = None + + @classmethod + @abstractmethod + def from_knx(cls, raw: Tuple[int, ...]) -> Any: + """Parse/deserialize from KNX/IP raw data (big endian).""" + + @classmethod + @abstractmethod + def to_knx(cls, value: Any) -> Union[bytes, Tuple[int, ...]]: + """Serialize to KNX/IP raw data.""" @classmethod - def test_bytesarray(cls, raw): + def test_bytesarray(cls, raw: Tuple[int, ...]) -> None: """Test if array of raw bytes has the correct length and values of correct type.""" if cls.payload_length is None: raise NotImplementedError("payload_length has to be defined for: %s" % cls) @@ -58,24 +76,26 @@ def test_bytesarray(cls, raw): raise ConversionError("Invalid raw bytes", raw=raw) @classmethod - def __recursive_subclasses__(cls): + def __recursive_subclasses__(cls) -> Iterator[Type["DPTBase"]]: """Yield all subclasses and their subclasses.""" for subclass in cls.__subclasses__(): yield from subclass.__recursive_subclasses__() yield subclass @classmethod - def has_distinct_dpt_numbers(cls): + def has_distinct_dpt_numbers(cls) -> bool: """Return True if dpt numbers are defined (not inherited).""" return "dpt_main_number" in cls.__dict__ and "dpt_sub_number" in cls.__dict__ @classmethod - def has_distinct_value_type(cls): + def has_distinct_value_type(cls) -> bool: """Return True if value_type is defined (not inherited).""" return "value_type" in cls.__dict__ @staticmethod - def transcoder_by_dpt(dpt_main, dpt_sub=None): + def transcoder_by_dpt( + dpt_main: int, dpt_sub: Optional[int] = None + ) -> Optional[Type["DPTBase"]]: """Return Class reference of DPTBase subclass with matching DPT number.""" for dpt in DPTBase.__recursive_subclasses__(): if dpt.has_distinct_dpt_numbers(): @@ -84,7 +104,7 @@ def transcoder_by_dpt(dpt_main, dpt_sub=None): return None @staticmethod - def transcoder_by_value_type(value_type): + def transcoder_by_value_type(value_type: str) -> Optional[Type["DPTBase"]]: """Return Class reference of DPTBase subclass with matching value_type.""" for dpt in DPTBase.__recursive_subclasses__(): if dpt.has_distinct_value_type(): @@ -92,8 +112,11 @@ def transcoder_by_value_type(value_type): return dpt return None + # TODO: convert to classmethod to allow parsing only subclasses (eg. for Numeric, Control etc.) @staticmethod - def parse_transcoder(value_type): + def parse_transcoder( + value_type: Union[int, float, str] + ) -> Optional[Type["DPTBase"]]: """Return Class reference of DPTBase subclass from value_type or DPT number.""" if isinstance(value_type, int): return DPTBase.transcoder_by_dpt(value_type) @@ -116,7 +139,6 @@ def parse_transcoder(value_type): except ValueError: pass return transcoder - return None class DPTBinary: @@ -128,12 +150,14 @@ class DPTBinary: APCI_BITMASK = 0x3F APCI_MAX_VALUE = APCI_BITMASK - def __init__(self, value: int) -> None: + def __init__(self, value: Union[int, Tuple[int]]) -> None: """Initialize DPTBinary class.""" + if isinstance(value, tuple): + value = value[0] if not isinstance(value, int): raise TypeError() if value > DPTBinary.APCI_BITMASK: - raise ConversionError("Could not init DPTBinary", value=value) + raise ConversionError("Could not init DPTBinary", value=str(value)) self.value = value @@ -152,8 +176,9 @@ class DPTArray: """The DPTArray is a base class for all datatypes appended to the KNX telegram.""" # pylint: disable=too-few-public-methods - def __init__(self, value: Union[int, list, bytes, tuple]) -> None: + def __init__(self, value: Union[int, bytes, Tuple[int, ...], List[int]]) -> None: """Initialize DPTArray class.""" + self.value: Tuple[int, ...] if isinstance(value, int): self.value = (value,) elif isinstance(value, (list, bytes)): diff --git a/xknx/dpt/dpt_1byte_signed.py b/xknx/dpt/dpt_1byte_signed.py index aacc7e7ba..3f18d94f4 100644 --- a/xknx/dpt/dpt_1byte_signed.py +++ b/xknx/dpt/dpt_1byte_signed.py @@ -1,4 +1,5 @@ """Implementation of Basic KNX 1-Byte signed integer values.""" +from typing import Optional, Tuple from xknx.exceptions import ConversionError @@ -15,13 +16,13 @@ class DPTSignedRelativeValue(DPTBase): value_min = -128 value_max = 127 dpt_main_number = 6 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "1byte_signed" unit = "" payload_length = 1 @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) if raw[0] > cls.value_max: @@ -29,7 +30,7 @@ def from_knx(cls, raw): return raw[0] @classmethod - def to_knx(cls, value): + def to_knx(cls, value: int) -> Tuple[int]: """Serialize to KNX/IP raw data.""" try: knx_value = int(value) @@ -42,7 +43,7 @@ def to_knx(cls, value): raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: int) -> bool: """Test if value is within defined range for this object.""" return cls.value_min <= value <= cls.value_max diff --git a/xknx/dpt/dpt_1byte_uint.py b/xknx/dpt/dpt_1byte_uint.py index 51305883b..4a3fcefbd 100644 --- a/xknx/dpt/dpt_1byte_uint.py +++ b/xknx/dpt/dpt_1byte_uint.py @@ -1,4 +1,6 @@ """Implementation of Basic KNX DPT_1_Ucount Values.""" +from typing import Optional, Tuple + from xknx.exceptions import ConversionError from .dpt import DPTBase @@ -14,14 +16,14 @@ class DPTValue1ByteUnsigned(DPTBase): value_min = 0 value_max = 255 dpt_main_number = 5 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "1byte_unsigned" unit = "" resolution = 1 payload_length = 1 @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) @@ -35,7 +37,7 @@ def from_knx(cls, raw): return value @classmethod - def to_knx(cls, value): + def to_knx(cls, value: int) -> Tuple[int]: """Serialize to KNX/IP raw data.""" try: knx_value = int(value) @@ -46,7 +48,7 @@ def to_knx(cls, value): raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: int) -> bool: """Test if value is within defined range for this object.""" return cls.value_min <= value <= cls.value_max @@ -117,7 +119,7 @@ class DPTSceneNumber(DPTValue1ByteUnsigned): unit = "" @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) @@ -131,7 +133,7 @@ def from_knx(cls, raw): return value @classmethod - def to_knx(cls, value): + def to_knx(cls, value: int) -> Tuple[int]: """Serialize to KNX/IP raw data.""" try: knx_value = int(value) - 1 diff --git a/xknx/dpt/dpt_2byte_float.py b/xknx/dpt/dpt_2byte_float.py index 1783523ce..67c1b1c7d 100644 --- a/xknx/dpt/dpt_2byte_float.py +++ b/xknx/dpt/dpt_2byte_float.py @@ -3,6 +3,7 @@ They correspond to the the following KDN DPT 9 class. """ +from typing import Optional, Tuple from xknx.exceptions import ConversionError @@ -19,14 +20,14 @@ class DPT2ByteFloat(DPTBase): value_min = -671088.64 value_max = 670760.96 dpt_main_number = 9 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "2byte_float" unit = "" - resolution = 1 + resolution = 0.01 payload_length = 2 @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> float: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) data = (raw[0] * 256) + raw[1] @@ -45,10 +46,10 @@ def from_knx(cls, raw): return value @classmethod - def to_knx(cls, value): + def to_knx(cls, value: float) -> Tuple[int, int]: """Serialize to KNX/IP raw data.""" - def calc_exponent(float_value, sign): + def calc_exponent(float_value: float, sign: bool) -> Tuple[int, int]: """Return float exponent.""" exponent = 0 significand = abs(int(float_value * 100)) @@ -68,7 +69,7 @@ def calc_exponent(float_value, sign): if not cls._test_boundaries(knx_value): raise ValueError - sign = 1 if knx_value < 0 else 0 + sign = knx_value < 0 exponent, significand = calc_exponent(knx_value, sign) return (sign << 7) | (exponent << 3) | ( @@ -78,7 +79,7 @@ def calc_exponent(float_value, sign): raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: float) -> bool: """Test if value is within defined range for this object.""" return cls.value_min <= value <= cls.value_max @@ -93,7 +94,6 @@ class DPTTemperature(DPT2ByteFloat): value_type = "temperature" unit = "°C" ha_device_class = "temperature" - resolution = 1 class DPTTemperatureDifference2Byte(DPT2ByteFloat): @@ -106,7 +106,6 @@ class DPTTemperatureDifference2Byte(DPT2ByteFloat): value_type = "temperature_difference_2byte" unit = "K" ha_device_class = "temperature" - resolution = 1 class DPTTemperatureA(DPT2ByteFloat): @@ -118,7 +117,6 @@ class DPTTemperatureA(DPT2ByteFloat): dpt_sub_number = 3 value_type = "temperature_a" unit = "K/h" - resolution = 1 class DPTLux(DPT2ByteFloat): @@ -131,7 +129,6 @@ class DPTLux(DPT2ByteFloat): value_type = "illuminance" unit = "lx" ha_device_class = "illuminance" - resolution = 1 class DPTWsp(DPT2ByteFloat): @@ -143,7 +140,6 @@ class DPTWsp(DPT2ByteFloat): dpt_sub_number = 5 value_type = "wind_speed_ms" unit = "m/s" - resolution = 1 class DPTPressure2Byte(DPT2ByteFloat): @@ -156,7 +152,6 @@ class DPTPressure2Byte(DPT2ByteFloat): value_type = "pressure_2byte" unit = "Pa" ha_device_class = "pressure" - resolution = 1 class DPTHumidity(DPT2ByteFloat): @@ -169,7 +164,6 @@ class DPTHumidity(DPT2ByteFloat): value_type = "humidity" unit = "%" ha_device_class = "humidity" - resolution = 1 class DPTPartsPerMillion(DPT2ByteFloat): @@ -190,7 +184,6 @@ class DPTTime1(DPT2ByteFloat): dpt_sub_number = 10 value_type = "time_1" unit = "s" - resolution = 1 class DPTTime2(DPT2ByteFloat): @@ -202,7 +195,6 @@ class DPTTime2(DPT2ByteFloat): dpt_sub_number = 11 value_type = "time_2" unit = "ms" - resolution = 1 class DPTVoltage(DPT2ByteFloat): @@ -281,7 +273,6 @@ class DPTTemperatureF(DPT2ByteFloat): value_type = "temperature_f" unit = "°F" ha_device_class = "temperature" - resolution = 1 class DPTWspKmh(DPT2ByteFloat): @@ -293,7 +284,6 @@ class DPTWspKmh(DPT2ByteFloat): dpt_sub_number = 28 value_type = "wind_speed_kmh" unit = "km/h" - resolution = 1 class DPTEnthalpy(DPT2ByteFloat): diff --git a/xknx/dpt/dpt_2byte_signed.py b/xknx/dpt/dpt_2byte_signed.py index 03ab01e96..958dd562b 100644 --- a/xknx/dpt/dpt_2byte_signed.py +++ b/xknx/dpt/dpt_2byte_signed.py @@ -4,8 +4,8 @@ They correspond the following KNX DPTs: 8.*** 2-byte/octet signed (2's complement), i.e. percentV16, delta time """ - import struct +from typing import Optional, Tuple from xknx.exceptions import ConversionError @@ -22,26 +22,26 @@ class DPT2ByteSigned(DPTBase): value_min = -32768 value_max = 32767 dpt_main_number = 8 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "2byte_signed" unit = "" - resolution = 1 + resolution: float = 1 payload_length = 2 _struct_format = ">h" @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) try: - return struct.unpack(cls._struct_format, bytes(raw))[0] + return struct.unpack(cls._struct_format, bytes(raw))[0] # type: ignore except struct.error: raise ConversionError("Could not parse %s" % cls.__name__, raw=raw) @classmethod - def to_knx(cls, value): + def to_knx(cls, value: int) -> Tuple[int, ...]: """Serialize to KNX/IP raw data.""" try: knx_value = int(value) @@ -52,7 +52,7 @@ def to_knx(cls, value): raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: int) -> bool: """Test if value is within defined range for this object.""" return cls.value_min <= value <= cls.value_max diff --git a/xknx/dpt/dpt_2byte_uint.py b/xknx/dpt/dpt_2byte_uint.py index 7f8fd918c..8cb3dd3cc 100644 --- a/xknx/dpt/dpt_2byte_uint.py +++ b/xknx/dpt/dpt_2byte_uint.py @@ -1,4 +1,5 @@ """Implementation of Basic KNX 2-Byte/octet values.""" +from typing import Optional, Tuple from xknx.exceptions import ConversionError @@ -17,20 +18,20 @@ class DPT2ByteUnsigned(DPTBase): value_min = 0 value_max = 65535 dpt_main_number = 7 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "2byte_unsigned" unit = "" resolution = 1 payload_length = 2 @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) return (raw[0] * 256) + raw[1] @classmethod - def to_knx(cls, value): + def to_knx(cls, value: int) -> Tuple[int, int]: """Serialize to KNX/IP raw data.""" try: knx_value = int(value) @@ -41,7 +42,7 @@ def to_knx(cls, value): raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: int) -> bool: """Test if value is within defined range for this object.""" return cls.value_min <= value <= cls.value_max diff --git a/xknx/dpt/dpt_4bit_control.py b/xknx/dpt/dpt_4bit_control.py index 6b93f3fc7..caad73bc2 100644 --- a/xknx/dpt/dpt_4bit_control.py +++ b/xknx/dpt/dpt_4bit_control.py @@ -11,15 +11,16 @@ As the same payload in these cases in interpreted completely different it is reasonable to make separate DPT classes. """ +from abc import ABC from enum import Enum -from typing import Tuple +from typing import Any, Dict, Optional, Tuple, Union from xknx.exceptions import ConversionError from .dpt import DPTBase -class DPTControlStepCode(DPTBase): +class DPTControlStepCode(DPTBase, ABC): """Abstraction for KNX B1U3 values (DPT 3.007/3.008).""" # APCI (application layer control information) @@ -27,49 +28,48 @@ class DPTControlStepCode(DPTBase): APCI_STEPCODEMASK = 0x07 APCI_MAX_VALUE = APCI_CONTROLMASK | APCI_STEPCODEMASK - value_type = "control_stepcode" unit = "" payload_length = 1 @classmethod - def _encode(cls, control: bool, step_code: int): + def _encode(cls, control: bool, step_code: int) -> int: """Encode control-bit with step-code.""" value = 1 if control > 0 else 0 value = (value << 3) | (step_code & cls.APCI_STEPCODEMASK) return value @classmethod - def _decode(cls, value) -> Tuple[bool, int]: + def _decode(cls, value: int) -> Tuple[bool, int]: """Decode value into control-bit and step-code.""" - control = 1 if (value & cls.APCI_CONTROLMASK) != 0 else 0 + control = bool(value & cls.APCI_CONTROLMASK) step_code = value & cls.APCI_STEPCODEMASK return control, step_code @classmethod - def _test_boundaries(cls, raw): + def _test_boundaries(cls, raw: int) -> bool: """Test if raw KNX data is within defined range for this object.""" if isinstance(raw, int): return 0 <= raw <= cls.APCI_MAX_VALUE - return False @classmethod - def _test_values(cls, control: bool, step_code: int): + def _test_values(cls, control: bool, step_code: int) -> bool: """Test if input values are valid.""" - if isinstance(control, int) and isinstance(step_code, int): - if control in (0, 1) and 0 <= step_code <= cls.APCI_STEPCODEMASK: + if isinstance(control, bool) and isinstance(step_code, int): + if 0 <= step_code <= cls.APCI_STEPCODEMASK: return True return False @classmethod - def to_knx(cls, value, invert: bool = False): + def to_knx(cls, value: Any) -> Tuple[int]: """Serialize to KNX/IP raw data.""" + # TODO: use Tuple or Named Tuple instead of Dict[str, int] to account for bool control if not isinstance(value, dict): raise ConversionError( "Cant serialize %s; invalid value type" % cls.__name__, value=value ) try: - control = value["control"] + control = bool(value["control"]) step_code = value["step_code"] except KeyError: raise ConversionError( @@ -81,21 +81,15 @@ def to_knx(cls, value, invert: bool = False): "Cant serialize %s; invalid values" % cls.__name__, value=value ) - if invert: - control = 0 if control > 0 else 1 - - return cls._encode(control, step_code) + return (cls._encode(control, step_code),) @classmethod - def from_knx(cls, raw, invert: bool = False): + def from_knx(cls, raw: Tuple[int, ...]) -> Any: """Parse/deserialize from KNX/IP raw data.""" - if not cls._test_boundaries(raw): + if not isinstance(raw, tuple) or not cls._test_boundaries(raw[0]): raise ConversionError("Cant parse %s" % cls.__name__, raw=raw) - control, step_code = cls._decode(raw) - - if invert: - control = 0 if control > 0 else 1 + control, step_code = cls._decode(raw[0]) return {"control": control, "step_code": step_code} @@ -104,12 +98,12 @@ class DPTControlStepwise(DPTControlStepCode): """Abstraction for KNX DPT 3.xxx in stepwise mode with conversion to an incement value.""" dpt_main_number = 3 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "stepwise" unit = "%" @staticmethod - def _from_increment(value): + def _from_increment(value: int) -> Dict[str, int]: """Calculate control bit and stepcode as defined in the KNX standard section 3.3.1 from an increment value.""" # control bit in KNX standard # 0: - = decrease/move up @@ -137,24 +131,24 @@ def _from_increment(value): return {"control": control, "step_code": stepcode} @staticmethod - def _to_increment(value): + def _to_increment(value: Dict[str, int]) -> int: """Calculate the increment value from the stepcode and control bit as defined in the KNX standard section 3.3.1.""" # calculated using floor(100/2^((value&0x07)-1)) inc = [0, 100, 50, 25, 12, 6, 3, 1][value["step_code"] & 0x07] return inc if value["control"] == 1 else -inc @classmethod - def to_knx(cls, value, invert: bool = False): + def to_knx(cls, value: Union[int, Dict[str, int]]) -> Tuple[int]: """Serialize to KNX/IP raw data.""" if not isinstance(value, int): raise ConversionError("Cant serialize %s" % cls.__name__, value=value) - return super().to_knx(cls._from_increment(value), invert) + return super().to_knx(cls._from_increment(value)) @classmethod - def from_knx(cls, raw, invert: bool = False): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" - return cls._to_increment(super().from_knx(raw, invert)) + return cls._to_increment(super().from_knx(raw)) class DPTControlStepwiseDimming(DPTControlStepwise): @@ -199,7 +193,7 @@ class Direction(TitleEnum): STOP = 2 @classmethod - def to_knx(cls, value, invert: bool = False): + def to_knx(cls, value: Direction) -> Tuple[int]: """Convert value to payload.""" control = 0 step_code = 0 @@ -216,12 +210,12 @@ def to_knx(cls, value, invert: bool = False): raise ConversionError("Cant serialize %s" % cls.__name__, value=value) values = {"control": control, "step_code": step_code} - return super().to_knx(values, invert) + return super().to_knx(values) @classmethod - def from_knx(cls, raw, invert: bool = False): + def from_knx(cls, raw: Tuple[int, ...]) -> Direction: """Convert current payload to value.""" - values = super().from_knx(raw, invert) + values = super().from_knx(raw) if values["step_code"] == 0: return cls.Direction(2) # STOP if values["control"] == 0: diff --git a/xknx/dpt/dpt_4byte_float.py b/xknx/dpt/dpt_4byte_float.py index 278f752b9..027ad0c0f 100644 --- a/xknx/dpt/dpt_4byte_float.py +++ b/xknx/dpt/dpt_4byte_float.py @@ -3,8 +3,8 @@ They correspond to the the following KDN DPT 14 class. """ - import struct +from typing import Optional, Tuple from xknx.exceptions import ConversionError @@ -24,22 +24,22 @@ class DPT4ByteFloat(DPTBase): """ dpt_main_number = 14 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "4byte_float" unit = "" payload_length = 4 @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> float: """Parse/deserialize from KNX/IP raw data (big endian).""" cls.test_bytesarray(raw) try: - return struct.unpack(">f", bytes(raw))[0] + return struct.unpack(">f", bytes(raw))[0] # type: ignore except struct.error: raise ConversionError("Could not parse %s" % cls.__name__, raw=raw) @classmethod - def to_knx(cls, value): + def to_knx(cls, value: float) -> Tuple[int, ...]: """Serialize to KNX/IP raw data.""" try: knx_value = float(value) @@ -572,6 +572,7 @@ class DPTPowerFactor(DPT4ByteFloat): dpt_sub_number = 57 value_type = "powerfactor" unit = "cosΦ" + ha_device_class = "power_factor" class DPTPressure(DPT4ByteFloat): diff --git a/xknx/dpt/dpt_4byte_int.py b/xknx/dpt/dpt_4byte_int.py index ba1d2a384..d88706a92 100644 --- a/xknx/dpt/dpt_4byte_int.py +++ b/xknx/dpt/dpt_4byte_int.py @@ -5,8 +5,8 @@ 12.yyy 4-byte/octet unsigned value, i.e. pulse counter 13.yyy 4-byte/octet signed (2's complement), i.e. flow, energy """ - import struct +from typing import Optional, Tuple from xknx.exceptions import ConversionError @@ -23,26 +23,26 @@ class DPT4ByteUnsigned(DPTBase): value_min = 0 value_max = 4294967295 dpt_main_number = 12 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "4byte_unsigned" unit = "" - resolution = 1 + resolution: float = 1 payload_length = 4 _struct_format = ">I" @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) try: - return struct.unpack(cls._struct_format, bytes(raw))[0] + return struct.unpack(cls._struct_format, bytes(raw))[0] # type: ignore except struct.error: raise ConversionError("Could not parse %s" % cls.__name__, raw=raw) @classmethod - def to_knx(cls, value): + def to_knx(cls, value: int) -> Tuple[int, ...]: """Serialize to KNX/IP raw data.""" try: knx_value = int(value) @@ -53,7 +53,7 @@ def to_knx(cls, value): raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: int) -> bool: """Test if value is within defined range for this object.""" return cls.value_min <= value <= cls.value_max @@ -68,10 +68,10 @@ class DPT4ByteSigned(DPT4ByteUnsigned): value_min = -2147483648 value_max = 2147483647 dpt_main_number = 13 - dpt_sub_number = None + dpt_sub_number: Optional[int] = None value_type = "4byte_signed" unit = "" - resolution = 1 + resolution: float = 1 _struct_format = ">i" @@ -120,6 +120,7 @@ class DPTActiveEnergy(DPT4ByteSigned): dpt_sub_number = 10 value_type = "active_energy" unit = "Wh" + ha_device_class = "energy" class DPTApparantEnergy(DPT4ByteSigned): @@ -129,6 +130,7 @@ class DPTApparantEnergy(DPT4ByteSigned): dpt_sub_number = 11 value_type = "apparant_energy" unit = "VAh" + ha_device_class = "energy" class DPTReactiveEnergy(DPT4ByteSigned): @@ -138,6 +140,7 @@ class DPTReactiveEnergy(DPT4ByteSigned): dpt_sub_number = 12 value_type = "reactive_energy" unit = "VARh" + ha_device_class = "energy" class DPTActiveEnergykWh(DPT4ByteSigned): @@ -147,6 +150,7 @@ class DPTActiveEnergykWh(DPT4ByteSigned): dpt_sub_number = 13 value_type = "active_energy_kwh" unit = "kWh" + ha_device_class = "energy" class DPTApparantEnergykVAh(DPT4ByteSigned): @@ -156,6 +160,7 @@ class DPTApparantEnergykVAh(DPT4ByteSigned): dpt_sub_number = 14 value_type = "apparant_energy_kvah" unit = "kVAh" + ha_device_class = "energy" class DPTReactiveEnergykVARh(DPT4ByteSigned): @@ -165,6 +170,7 @@ class DPTReactiveEnergykVARh(DPT4ByteSigned): dpt_sub_number = 15 value_type = "reactive_energy_kvarh" unit = "kVARh" + ha_device_class = "energy" class DPTLongDeltaTimeSec(DPT4ByteSigned): diff --git a/xknx/dpt/dpt_date.py b/xknx/dpt/dpt_date.py index 4d387469d..baa5adbeb 100644 --- a/xknx/dpt/dpt_date.py +++ b/xknx/dpt/dpt_date.py @@ -1,6 +1,6 @@ """Implementation of the KNX date data point.""" - import time +from typing import Tuple from xknx.exceptions import ConversionError @@ -13,7 +13,7 @@ class DPTDate(DPTBase): payload_length = 3 @classmethod - def from_knx(cls, raw) -> time.struct_time: + def from_knx(cls, raw: Tuple[int, ...]) -> time.struct_time: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) @@ -36,10 +36,10 @@ def from_knx(cls, raw) -> time.struct_time: raise ConversionError("Could not parse DPTDate", raw=raw) @classmethod - def to_knx(cls, value: time.struct_time): + def to_knx(cls, value: time.struct_time) -> Tuple[int, int, int]: """Serialize to KNX/IP raw data from time.struct_time.""" - def _knx_year(year): + def _knx_year(year: int) -> int: if 2000 <= year < 2090: return year - 2000 if 1990 <= year < 2000: diff --git a/xknx/dpt/dpt_datetime.py b/xknx/dpt/dpt_datetime.py index 85f679199..503834a0d 100644 --- a/xknx/dpt/dpt_datetime.py +++ b/xknx/dpt/dpt_datetime.py @@ -1,6 +1,6 @@ """Implementation of the KNX datetime data point.""" - import time +from typing import Tuple from xknx.exceptions import ConversionError @@ -13,7 +13,7 @@ class DPTDateTime(DPTBase): payload_length = 8 @classmethod - def from_knx(cls, raw) -> time.struct_time: + def from_knx(cls, raw: Tuple[int, ...]) -> time.struct_time: """Parse/deserialize from KNX/IP raw data.""" # pylint: disable=too-many-locals cls.test_bytesarray(raw) @@ -68,7 +68,7 @@ def from_knx(cls, raw) -> time.struct_time: raise ConversionError("Could not parse DPTDateTime", raw=raw) @classmethod - def to_knx(cls, value: time.struct_time): + def to_knx(cls, value: time.struct_time) -> Tuple[int, ...]: """Serialize to KNX/IP raw data from time.struct_time.""" if not isinstance(value, time.struct_time): raise ConversionError("Could not serialize DPTDateTime", value=value) diff --git a/xknx/dpt/dpt_hvac_mode.py b/xknx/dpt/dpt_hvac_mode.py index 1b0052c97..ed2680d82 100644 --- a/xknx/dpt/dpt_hvac_mode.py +++ b/xknx/dpt/dpt_hvac_mode.py @@ -1,11 +1,13 @@ """Implementation of different KNX DPT HVAC Operation modes.""" - from enum import Enum +from typing import Dict, Generic, Tuple, TypeVar from xknx.exceptions import ConversionError, CouldNotParseKNXIP from .dpt import DPTBase +HVACModeType = TypeVar("HVACModeType", "HVACControllerMode", "HVACOperationMode") + class HVACOperationMode(Enum): """Enum for the different KNX HVAC operation modes.""" @@ -36,15 +38,15 @@ class HVACControllerMode(Enum): NODEM = "NoDem" -class _DPTClimateMode(DPTBase): +class _DPTClimateMode(DPTBase, Generic[HVACModeType]): """Base class for KNX Climate modes.""" - SUPPORTED_MODES = {} + SUPPORTED_MODES: Dict[int, HVACModeType] = {} payload_length = 1 @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> HVACModeType: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) if raw[0] in cls.SUPPORTED_MODES: @@ -52,7 +54,7 @@ def from_knx(cls, raw): raise CouldNotParseKNXIP("Could not parse HVACOperationMode") @classmethod - def to_knx(cls, value): + def to_knx(cls, value: HVACModeType) -> Tuple[int]: """Serialize to KNX/IP raw data.""" for knx_value, mode in cls.SUPPORTED_MODES.items(): if mode == value: @@ -60,14 +62,14 @@ def to_knx(cls, value): raise ConversionError("Could not parse %s" % cls.__name__, value=value) -class DPTHVACContrMode(_DPTClimateMode): +class DPTHVACContrMode(_DPTClimateMode[HVACControllerMode]): """ Abstraction for KNX HVAC controller mode. DPT 20.105 """ - SUPPORTED_MODES = { + SUPPORTED_MODES: Dict[int, HVACControllerMode] = { 0: HVACControllerMode.AUTO, 1: HVACControllerMode.HEAT, 2: HVACControllerMode.MORNING_WARMUP, @@ -85,24 +87,23 @@ class DPTHVACContrMode(_DPTClimateMode): } -class DPTHVACMode(_DPTClimateMode): +class DPTHVACMode(_DPTClimateMode[HVACOperationMode]): """ Abstraction for KNX HVAC mode. DPT 20.102 """ - SUPPORTED_MODES = { + SUPPORTED_MODES: Dict[int, HVACOperationMode] = { 0: HVACOperationMode.AUTO, 1: HVACOperationMode.COMFORT, 2: HVACOperationMode.STANDBY, 3: HVACOperationMode.NIGHT, 4: HVACOperationMode.FROST_PROTECTION, } - SUPPORTED_MODES_INV = dict(reversed(item) for item in SUPPORTED_MODES.items()) -class DPTControllerStatus(_DPTClimateMode): +class DPTControllerStatus(_DPTClimateMode[HVACOperationMode]): """ Abstraction for KNX HVAC Controller status. @@ -113,17 +114,15 @@ class DPTControllerStatus(_DPTClimateMode): notes on the correct implementation of this type are highly appreciated. """ - SUPPORTED_MODES = { + SUPPORTED_MODES: Dict[int, HVACOperationMode] = { 0x21: HVACOperationMode.COMFORT, 0x22: HVACOperationMode.STANDBY, 0x24: HVACOperationMode.NIGHT, 0x28: HVACOperationMode.FROST_PROTECTION, } - SUPPORTED_MODES_INV = dict(reversed(item) for item in SUPPORTED_MODES.items()) - @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> HVACOperationMode: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) if raw[0] & 8 > 0: diff --git a/xknx/dpt/dpt_scaling.py b/xknx/dpt/dpt_scaling.py index 9d2ee74b5..8ce332886 100644 --- a/xknx/dpt/dpt_scaling.py +++ b/xknx/dpt/dpt_scaling.py @@ -1,4 +1,6 @@ """Implementation of scaled KNX DPT_1_Ucount Values.""" +from typing import Tuple + from xknx.exceptions import ConversionError from .dpt import DPTBase @@ -21,7 +23,7 @@ class DPTScaling(DPTBase): payload_length = 1 @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> int: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) @@ -37,7 +39,7 @@ def from_knx(cls, raw): return value @classmethod - def to_knx(cls, value): + def to_knx(cls, value: float) -> Tuple[int]: """Serialize to KNX/IP raw data.""" try: percent_value = float(value) @@ -51,7 +53,7 @@ def to_knx(cls, value): raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: float) -> bool: """Test if value is within defined range for this object.""" return cls.value_min <= value <= cls.value_max diff --git a/xknx/dpt/dpt_string.py b/xknx/dpt/dpt_string.py index 9d6342947..76fd244c6 100644 --- a/xknx/dpt/dpt_string.py +++ b/xknx/dpt/dpt_string.py @@ -1,4 +1,6 @@ """Implementation of 3.17 Datapoint Types String.""" +from typing import Tuple + from xknx.exceptions import ConversionError from .dpt import DPTBase @@ -18,7 +20,7 @@ class DPTString(DPTBase): unit = "" @classmethod - def from_knx(cls, raw): + def from_knx(cls, raw: Tuple[int, ...]) -> str: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) value = "" @@ -28,7 +30,7 @@ def from_knx(cls, raw): return value @classmethod - def to_knx(cls, value): + def to_knx(cls, value: str) -> Tuple[int, ...]: """Serialize to KNX/IP raw data.""" try: knx_value = str(value) @@ -38,11 +40,11 @@ def to_knx(cls, value): for character in knx_value: raw.append(ord(character)) raw.extend([0] * (cls.payload_length - len(raw))) - return raw + return tuple(raw) except ValueError: raise ConversionError("Could not serialize %s" % cls.__name__, value=value) @classmethod - def _test_boundaries(cls, value): + def _test_boundaries(cls, value: str) -> bool: """Test if value is within defined range for this object.""" return len(value) <= cls.payload_length diff --git a/xknx/dpt/dpt_time.py b/xknx/dpt/dpt_time.py index 272dfa032..2db62b7d9 100644 --- a/xknx/dpt/dpt_time.py +++ b/xknx/dpt/dpt_time.py @@ -1,6 +1,6 @@ """Implementation of Basic KNX Time.""" - import time +from typing import Tuple from xknx.exceptions import ConversionError @@ -17,7 +17,7 @@ class DPTTime(DPTBase): payload_length = 3 @classmethod - def from_knx(cls, raw) -> time.struct_time: + def from_knx(cls, raw: Tuple[int, ...]) -> time.struct_time: """Parse/deserialize from KNX/IP raw data.""" cls.test_bytesarray(raw) @@ -44,7 +44,7 @@ def from_knx(cls, raw) -> time.struct_time: raise ConversionError("Could not parse DPTTime", raw=raw) @classmethod - def to_knx(cls, value: time.struct_time): + def to_knx(cls, value: time.struct_time) -> Tuple[int, int, int]: """Serialize to KNX/IP raw data from dict with elements weekday,hours,minutes,seconds.""" if not isinstance(value, time.struct_time): raise ConversionError( @@ -62,7 +62,7 @@ def to_knx(cls, value: time.struct_time): return (weekday << 5 | value.tm_hour, value.tm_min, value.tm_sec) @staticmethod - def _test_range(weekday, hours, minutes, seconds): + def _test_range(weekday: int, hours: int, minutes: int, seconds: int) -> bool: """Test if values are in the correct value range.""" if weekday < 0 or weekday > 7: return False diff --git a/xknx/exceptions/exception.py b/xknx/exceptions/exception.py index 649c4b19b..cd3d56c66 100644 --- a/xknx/exceptions/exception.py +++ b/xknx/exceptions/exception.py @@ -31,7 +31,7 @@ def __init__(self, message: str, should_log: bool = True) -> None: class CouldNotParseTelegram(XKNXException): """Could not parse telegram error.""" - def __init__(self, description: str, **kwargs: str) -> None: + def __init__(self, description: str, **kwargs: Any) -> None: """Initialize CouldNotParseTelegram class.""" super().__init__() self.description = description @@ -78,7 +78,7 @@ def __str__(self) -> str: class ConversionError(XKNXException): """Exception class for error while converting one type to another.""" - def __init__(self, description: str, **kwargs: str) -> None: + def __init__(self, description: str, **kwargs: Any) -> None: """Initialize ConversionError class.""" super().__init__() self.description = description diff --git a/xknx/remote_value/__init__.py b/xknx/remote_value/__init__.py index aaace6a62..e94c38e37 100644 --- a/xknx/remote_value/__init__.py +++ b/xknx/remote_value/__init__.py @@ -5,7 +5,8 @@ from .remote_value_climate_mode import ( RemoteValueBinaryHeatCool, RemoteValueBinaryOperationMode, - RemoteValueClimateMode, + RemoteValueControllerMode, + RemoteValueOperationMode, ) from .remote_value_color_rgb import RemoteValueColorRGB from .remote_value_color_rgbw import RemoteValueColorRGBW @@ -28,13 +29,14 @@ "RemoteValue1Count", "RemoteValueBinaryHeatCool", "RemoteValueBinaryOperationMode", - "RemoteValueClimateMode", "RemoteValueColorRGB", "RemoteValueColorRGBW", "RemoteValueControl", + "RemoteValueControllerMode", "RemoteValueDateTime", "RemoteValueDpt2ByteUnsigned", "RemoteValueDptValue1Ucount", + "RemoteValueOperationMode", "RemoteValueScaling", "RemoteValueSceneNumber", "RemoteValueSensor", diff --git a/xknx/remote_value/remote_value.py b/xknx/remote_value/remote_value.py index 73b1f75f6..9bcda02d6 100644 --- a/xknx/remote_value/remote_value.py +++ b/xknx/remote_value/remote_value.py @@ -13,28 +13,28 @@ Any, Awaitable, Callable, + Generic, Iterator, List, Optional, Union, ) +from xknx.dpt.dpt import DPTArray, DPTBinary, DPTPayloadType from xknx.exceptions import CouldNotParseTelegram from xknx.telegram import GroupAddress, Telegram from xknx.telegram.apci import GroupValueResponse, GroupValueWrite if TYPE_CHECKING: - from xknx.dpt import DPTArray, DPTBinary from xknx.telegram.address import GroupAddressableType from xknx.xknx import XKNX - AsyncCallback = Callable[[], Awaitable[None]] - DPTPayload = Union[DPTArray, DPTBinary] - logger = logging.getLogger("xknx.log") +AsyncCallbackType = Callable[[], Awaitable[None]] + -class RemoteValue(ABC): +class RemoteValue(ABC, Generic[DPTPayloadType]): """Class for managing remote knx value.""" # pylint: disable=too-many-instance-attributes @@ -46,12 +46,12 @@ def __init__( sync_state: bool = True, device_name: Optional[str] = None, feature_name: Optional[str] = None, - after_update_cb: Optional["AsyncCallback"] = None, + after_update_cb: Optional[AsyncCallbackType] = None, passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize RemoteValue class.""" # pylint: disable=too-many-arguments - self.xknx = xknx + self.xknx: "XKNX" = xknx if group_address is not None: group_address = GroupAddress(group_address) @@ -66,8 +66,8 @@ def __init__( self.device_name: str = "Unknown" if device_name is None else device_name self.feature_name: str = "Unknown" if feature_name is None else feature_name - self.after_update_cb = after_update_cb - self.payload: Optional["DPTPayload"] = None + self.after_update_cb: Optional[AsyncCallbackType] = after_update_cb + self.payload: Optional[DPTPayloadType] = None if sync_state and self.group_address_state: self.xknx.state_updater.register_remote_value( @@ -113,18 +113,19 @@ def _internal_addresses() -> Iterator[Optional[GroupAddress]]: return group_address in _internal_addresses() - @staticmethod @abstractmethod # TODO: typing - remove Optional - def payload_valid(payload: Optional["DPTPayload"]) -> bool: - """Test if telegram payload may be parsed - to be implemented in derived class..""" + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTPayloadType]: + """Return payload if telegram payload may be parsed - to be implemented in derived class.""" @abstractmethod - def from_knx(self, payload: "DPTPayload") -> Any: + def from_knx(self, payload: DPTPayloadType) -> Any: """Convert current payload to value - to be implemented in derived class.""" @abstractmethod - def to_knx(self, value: Any) -> "DPTPayload": + def to_knx(self, value: Any) -> DPTPayloadType: """Convert value to payload - to be implemented in derived class.""" async def process(self, telegram: Telegram, always_callback: bool = False) -> bool: @@ -148,7 +149,8 @@ async def process(self, telegram: Telegram, always_callback: bool = False) -> bo device_name=self.device_name, feature_name=self.feature_name, ) - if not self.payload_valid(telegram.payload.value): + _new_payload = self.payload_valid(telegram.payload.value) + if _new_payload is None: raise CouldNotParseTelegram( "payload invalid", payload=str(telegram.payload), @@ -158,12 +160,8 @@ async def process(self, telegram: Telegram, always_callback: bool = False) -> bo feature_name=self.feature_name, ) self.xknx.state_updater.update_received(self) - if ( - self.payload is None - or always_callback - or self.payload != telegram.payload.value - ): - self.payload = telegram.payload.value + if self.payload is None or always_callback or self.payload != _new_payload: + self.payload = _new_payload if self.after_update_cb is not None: await self.after_update_cb() return True @@ -175,7 +173,7 @@ def value(self) -> Any: return None return self.from_knx(self.payload) - async def _send(self, payload: "DPTPayload", response: bool = False) -> None: + async def _send(self, payload: DPTPayloadType, response: bool = False) -> None: """Send payload as telegram to KNX bus.""" if self.group_address is not None: telegram = Telegram( diff --git a/xknx/remote_value/remote_value_1count.py b/xknx/remote_value/remote_value_1count.py index b5a486b7e..58f58a147 100644 --- a/xknx/remote_value/remote_value_1count.py +++ b/xknx/remote_value/remote_value_1count.py @@ -3,22 +3,32 @@ DPT 6.010. """ -from xknx.dpt import DPTArray, DPTValue1Count +from typing import Optional, Union + +from xknx.dpt import DPTArray, DPTBinary, DPTValue1Count from .remote_value import RemoteValue -class RemoteValue1Count(RemoteValue): +class RemoteValue1Count(RemoteValue[DPTArray]): """Abstraction for remote value of KNX 6.010 (DPT_Value_1_Count).""" - def payload_valid(self, payload): + # pylint: disable=no-self-use + + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 1 + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 1 + else None + ) - def to_knx(self, value): + def to_knx(self, value: int) -> DPTArray: """Convert value to payload.""" return DPTArray(DPTValue1Count.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> int: """Convert current payload to value.""" return DPTValue1Count.from_knx(payload.value) diff --git a/xknx/remote_value/remote_value_climate_mode.py b/xknx/remote_value/remote_value_climate_mode.py index 545c38c3c..6756e9728 100644 --- a/xknx/remote_value/remote_value_climate_mode.py +++ b/xknx/remote_value/remote_value_climate_mode.py @@ -5,18 +5,7 @@ """ from abc import abstractmethod from enum import Enum -from typing import ( - TYPE_CHECKING, - Any, - Awaitable, - Callable, - Generic, - List, - Optional, - TypeVar, - Union, - cast, -) +from typing import TYPE_CHECKING, Any, Generic, List, Optional, Union from xknx.dpt import ( DPTArray, @@ -25,22 +14,20 @@ DPTHVACContrMode, DPTHVACMode, ) -from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACOperationMode +from xknx.dpt.dpt import DPTPayloadType +from xknx.dpt.dpt_hvac_mode import HVACControllerMode, HVACModeType, HVACOperationMode from xknx.exceptions import ConversionError, CouldNotParseTelegram -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, 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]): +class RemoteValueClimateModeBase( + RemoteValue[DPTPayloadType], Generic[DPTPayloadType, HVACModeType] +): """Base class for binary climate mode remote values.""" @abstractmethod @@ -50,14 +37,13 @@ def supported_operation_modes( """Return a list of all supported operation modes.""" -class RemoteValueClimateMode(RemoteValueClimateModeBase[HVACModeType]): - """Abstraction for remote value of KNX climate modes.""" +class RemoteValueOperationMode(RemoteValueClimateModeBase[DPTArray, HVACOperationMode]): + """Abstraction for remote value of KNX climate operation modes.""" class ClimateModeType(Enum): """Implemented climate mode types.""" CONTROLLER_STATUS = DPTControllerStatus - HVAC_CONTR_MODE = DPTHVACContrMode HVAC_MODE = DPTHVACMode def __init__( @@ -69,7 +55,7 @@ def __init__( device_name: Optional[str] = None, feature_name: str = "Climate mode", climate_mode_type: Optional[ClimateModeType] = None, - after_update_cb: Optional["AsyncCallback"] = None, + after_update_cb: Optional[AsyncCallbackType] = None, passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX climate mode.""" @@ -91,31 +77,94 @@ def __init__( device_name=str(device_name), feature_name=feature_name, ) - self._climate_mode_transcoder = climate_mode_type.value + self._climate_mode_transcoder: Union[ + DPTControllerStatus, DPTHVACMode + ] = climate_mode_type.value - def supported_operation_modes(self) -> List["HVACModeType"]: + def supported_operation_modes(self) -> List[HVACOperationMode]: """Return a list of all supported operation modes.""" return list(self._climate_mode_transcoder.SUPPORTED_MODES.values()) - @staticmethod - def payload_valid(payload: Optional["DPTPayload"]) -> bool: + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 1 + # pylint: disable=no-self-use + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 1 + else None + ) - def to_knx(self, value: Any) -> "DPTPayload": + def to_knx(self, value: Any) -> DPTArray: """Convert value to payload.""" return DPTArray(self._climate_mode_transcoder.to_knx(value)) - def from_knx(self, payload: "DPTPayload") -> Optional[HVACModeType]: + def from_knx(self, payload: DPTArray) -> Optional[HVACOperationMode]: """Convert current payload to value.""" - # TODO: typing - remove cast - return cast( - Optional[HVACModeType], - self._climate_mode_transcoder.from_knx(payload.value), + return self._climate_mode_transcoder.from_knx(payload.value) + + +class RemoteValueControllerMode( + RemoteValueClimateModeBase[DPTArray, HVACControllerMode] +): + """Abstraction for remote value of KNX climate controller modes.""" + + # pylint: disable=no-self-use + + def __init__( + self, + 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", + after_update_cb: Optional[AsyncCallbackType] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, + ): + """Initialize remote value of KNX climate mode.""" + # pylint: disable=too-many-arguments + super().__init__( + xknx, + group_address=group_address, + group_address_state=group_address_state, + sync_state=sync_state, + device_name=device_name, + feature_name=feature_name, + after_update_cb=after_update_cb, + passive_group_addresses=passive_group_addresses, + ) + + @staticmethod + def supported_operation_modes() -> List[HVACControllerMode]: + """Return a list of all supported operation modes.""" + return list(DPTHVACContrMode.SUPPORTED_MODES.values()) + + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: + """Test if telegram payload may be parsed.""" + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 1 + else None ) + @staticmethod + def to_knx(value: Any) -> DPTArray: + """Convert value to payload.""" + return DPTArray(DPTHVACContrMode.to_knx(value)) -class RemoteValueBinaryOperationMode(RemoteValueClimateModeBase[HVACOperationMode]): + @staticmethod + def from_knx(payload: DPTArray) -> Optional[HVACControllerMode]: + """Convert current payload to value.""" + return DPTHVACContrMode.from_knx(payload.value) + + +class RemoteValueBinaryOperationMode( + RemoteValueClimateModeBase[DPTBinary, HVACOperationMode] +): """Abstraction for remote value of split up KNX climate modes.""" def __init__( @@ -126,7 +175,7 @@ def __init__( sync_state: bool = True, device_name: Optional[str] = None, feature_name: str = "Climate mode binary", - after_update_cb: Optional["AsyncCallback"] = None, + after_update_cb: Optional[AsyncCallbackType] = None, operation_mode: Optional[HVACOperationMode] = None, ): """Initialize remote value of KNX DPT 1 representing a climate operation mode.""" @@ -156,12 +205,14 @@ def __init__( after_update_cb=after_update_cb, ) - @staticmethod - def payload_valid(payload: Optional["DPTPayload"]) -> bool: + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTBinary]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTBinary) + # pylint: disable=no-self-use + return payload if isinstance(payload, DPTBinary) else None - def to_knx(self, value: Any) -> "DPTPayload": + def to_knx(self, value: Any) -> DPTBinary: """Convert value to payload.""" if isinstance(value, HVACOperationMode): # foreign operation modes will set the RemoteValue to False @@ -183,7 +234,7 @@ def supported_operation_modes() -> List[HVACOperationMode]: HVACOperationMode.STANDBY, ] - def from_knx(self, payload: "DPTPayload") -> Optional[HVACOperationMode]: + def from_knx(self, payload: DPTPayloadType) -> Optional[HVACOperationMode]: """Convert current payload to value.""" if payload == DPTBinary(1): return self.operation_mode @@ -197,7 +248,9 @@ def from_knx(self, payload: "DPTPayload") -> Optional[HVACOperationMode]: ) -class RemoteValueBinaryHeatCool(RemoteValueClimateModeBase[HVACControllerMode]): +class RemoteValueBinaryHeatCool( + RemoteValueClimateModeBase[DPTBinary, HVACControllerMode] +): """Abstraction for remote value of heat/cool controller mode.""" def __init__( @@ -208,7 +261,7 @@ def __init__( sync_state: bool = True, device_name: Optional[str] = None, feature_name: str = "Controller mode Heat/Cool", - after_update_cb: Optional["AsyncCallback"] = None, + after_update_cb: Optional[AsyncCallbackType] = None, controller_mode: Optional[HVACControllerMode] = None, ): """Initialize remote value of KNX DPT 1 representing a climate controller mode.""" @@ -238,17 +291,19 @@ def __init__( after_update_cb=after_update_cb, ) - @staticmethod - def payload_valid(payload: Optional["DPTPayload"]) -> bool: + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTBinary]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTBinary) + # pylint: disable=no-self-use + return payload if isinstance(payload, DPTBinary) else None @staticmethod def supported_operation_modes() -> List[HVACControllerMode]: """Return a list of the configured operation mode.""" return [HVACControllerMode.HEAT, HVACControllerMode.COOL] - def to_knx(self, value: Any) -> "DPTPayload": + def to_knx(self, value: Any) -> DPTBinary: """Convert value to payload.""" if isinstance(value, HVACControllerMode): # foreign operation modes will set the RemoteValue to False @@ -260,7 +315,7 @@ def to_knx(self, value: Any) -> "DPTPayload": feature_name=self.feature_name, ) - def from_knx(self, payload: "DPTPayload") -> Optional[HVACControllerMode]: + def from_knx(self, payload: DPTPayloadType) -> Optional[HVACControllerMode]: """Convert current payload to value.""" if payload == DPTBinary(1): return self.controller_mode diff --git a/xknx/remote_value/remote_value_color_rgb.py b/xknx/remote_value/remote_value_color_rgb.py index 9fa8f47c9..c0d44adb7 100644 --- a/xknx/remote_value/remote_value_color_rgb.py +++ b/xknx/remote_value/remote_value_color_rgb.py @@ -3,26 +3,32 @@ DPT 232.600. """ -from typing import List +from typing import TYPE_CHECKING, List, Optional, Sequence, Tuple, Union -from xknx.dpt import DPTArray +from xknx.dpt import DPTArray, DPTBinary from xknx.exceptions import ConversionError -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueColorRGB(RemoteValue): + +class RemoteValueColorRGB(RemoteValue[DPTArray]): """Abstraction for remote value of KNX DPT 232.600 (DPT_Color_RGB).""" + # pylint: disable=no-self-use + def __init__( self, - xknx, - group_address=None, - group_address_state=None, - device_name=None, - feature_name="Color RGB", - after_update_cb=None, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + device_name: Optional[str] = None, + feature_name: str = "Color RGB", + after_update_cb: Optional[AsyncCallbackType] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX DPT 232.600 (DPT_Color_RGB).""" # pylint: disable=too-many-arguments @@ -36,11 +42,17 @@ def __init__( passive_group_addresses=passive_group_addresses, ) - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 3 + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 3 + else None + ) - def to_knx(self, value): + def to_knx(self, value: Sequence[int]) -> DPTArray: """Convert value to payload.""" if not isinstance(value, (list, tuple)): raise ConversionError( @@ -65,6 +77,6 @@ def to_knx(self, value): return DPTArray(list(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> Tuple[int, int, int]: """Convert current payload to value.""" return payload.value[0], payload.value[1], payload.value[2] diff --git a/xknx/remote_value/remote_value_color_rgbw.py b/xknx/remote_value/remote_value_color_rgbw.py index 28bd041e7..5f6bec178 100644 --- a/xknx/remote_value/remote_value_color_rgbw.py +++ b/xknx/remote_value/remote_value_color_rgbw.py @@ -3,26 +3,30 @@ DPT 251.600. """ -from typing import List +from typing import TYPE_CHECKING, List, Optional, Sequence, Tuple, Union -from xknx.dpt import DPTArray +from xknx.dpt import DPTArray, DPTBinary from xknx.exceptions import ConversionError -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueColorRGBW(RemoteValue): + +class RemoteValueColorRGBW(RemoteValue[DPTArray]): """Abstraction for remote value of KNX DPT 251.600 (DPT_Color_RGBW).""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - device_name=None, - feature_name="Color RGBW", - after_update_cb=None, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + device_name: Optional[str] = None, + feature_name: str = "Color RGBW", + after_update_cb: Optional[AsyncCallbackType] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX DPT 251.600 (DPT_Color_RGBW).""" # pylint: disable=too-many-arguments @@ -35,13 +39,20 @@ def __init__( after_update_cb=after_update_cb, passive_group_addresses=passive_group_addresses, ) - self.previous_value = (0, 0, 0, 0) + self.previous_value: Tuple[int, ...] = (0, 0, 0, 0) - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 6 + # pylint: disable=no-self-use + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 6 + else None + ) - def to_knx(self, value): + def to_knx(self, value: Sequence[int]) -> DPTArray: """ Convert value (4-6 bytes) to payload (6 bytes). @@ -65,6 +76,7 @@ def to_knx(self, value): * 4 bytes: 0x000f right padding to 6 bytes * < 4 bytes: error """ + # pylint: disable=no-self-use if not isinstance(value, (list, tuple)): raise ConversionError( "Could not serialize RemoteValueColorRGBW (wrong type, expecting list of 4-6 bytes))", @@ -92,16 +104,17 @@ def to_knx(self, value): return DPTArray(list(rgbw) + [0x00] + list(value[4:])) return DPTArray(value) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> Tuple[int, ...]: """ Convert current payload to value. Always 4 byte (RGBW). If one element is invalid, use the previous value. All previous element values are initialized to 0. """ - result = [] + _result = [] for i in range(0, len(payload.value) - 2): valid = (payload.value[5] & (0x08 >> i)) != 0 # R,G,B,W value valid? - result.append(payload.value[i] if valid else self.previous_value[i]) + _result.append(payload.value[i] if valid else self.previous_value[i]) + result = tuple(_result) self.previous_value = result return result diff --git a/xknx/remote_value/remote_value_control.py b/xknx/remote_value/remote_value_control.py index 96ea9d51e..f27e5dcd5 100644 --- a/xknx/remote_value/remote_value_control.py +++ b/xknx/remote_value/remote_value_control.py @@ -4,39 +4,44 @@ Examples are switching commands with priority control, relative dimming or blinds control commands. DPT 2.yyy and DPT 3.yyy """ -from typing import List, Optional +from typing import TYPE_CHECKING, Any, List, Optional, Type, Union -from xknx.dpt import DPTBase, DPTBinary +from xknx.dpt import DPTArray, DPTBase, DPTBinary, DPTControlStepCode from xknx.exceptions import ConversionError -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueControl(RemoteValue): + +class RemoteValueControl(RemoteValue[DPTBinary]): """Abstraction for remote value used for controling.""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - sync_state=True, - value_type=None, - device_name=None, - feature_name="Control", - after_update_cb=None, - invert=False, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + sync_state: bool = True, + value_type: Optional[str] = None, + device_name: Optional[str] = None, + feature_name: str = "Control", + after_update_cb: Optional[AsyncCallbackType] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize control remote value.""" # pylint: disable=too-many-arguments - self.invert = invert + if value_type is None: + raise ConversionError("no value type given", device_name=device_name) + # TODO: typing - parse from DPTControlStepCode when parse_transcoder is a classmethod _dpt_class = DPTBase.parse_transcoder(value_type) - if _dpt_class is None: + if _dpt_class is None or not isinstance(_dpt_class(), DPTControlStepCode): raise ConversionError( "invalid value type", value_type=value_type, device_name=device_name ) - self.dpt_class = _dpt_class + self.dpt_class: Type[DPTControlStepCode] = _dpt_class # type: ignore super().__init__( xknx, group_address, @@ -48,17 +53,21 @@ def __init__( passive_group_addresses=passive_group_addresses, ) - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTBinary]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTBinary) + # pylint: disable=no-self-use + return payload if isinstance(payload, DPTBinary) else None - def to_knx(self, value): + def to_knx(self, value: Any) -> DPTBinary: """Convert value to payload.""" - return DPTBinary(self.dpt_class.to_knx(value, invert=self.invert)) + return DPTBinary(self.dpt_class.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTBinary) -> Any: """Convert current payload to value.""" - return self.dpt_class.from_knx(payload.value, invert=self.invert) + # TODO: DPTBinary.value is int - DPTBase.from_knx requires Tuple[int, ...] - maybe use bytes + return self.dpt_class.from_knx((payload.value,)) @property def unit_of_measurement(self) -> Optional[str]: @@ -68,4 +77,4 @@ def unit_of_measurement(self) -> Optional[str]: @property def ha_device_class(self) -> Optional[str]: """Return a string representing the home assistant device class.""" - return getattr(self.dpt_class, "ha_device_class", None) + return getattr(self.dpt_class, "ha_device_class", None) # type: ignore diff --git a/xknx/remote_value/remote_value_datetime.py b/xknx/remote_value/remote_value_datetime.py index 0b2ada73e..f776f1067 100644 --- a/xknx/remote_value/remote_value_datetime.py +++ b/xknx/remote_value/remote_value_datetime.py @@ -5,12 +5,16 @@ """ from enum import Enum import time -from typing import List +from typing import TYPE_CHECKING, List, Optional, Type, Union -from xknx.dpt import DPTArray, DPTDate, DPTDateTime, DPTTime +from xknx.dpt import DPTArray, DPTBinary, DPTDate, DPTDateTime, DPTTime from xknx.exceptions import ConversionError -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue + +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX class DateTimeType(Enum): @@ -21,25 +25,27 @@ class DateTimeType(Enum): TIME = DPTTime -class RemoteValueDateTime(RemoteValue): +class RemoteValueDateTime(RemoteValue[DPTArray]): """Abstraction for remote value of KNX 10.001, 11.001 and 19.001 time and date objects.""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - sync_state=True, - value_type="time", - device_name=None, - feature_name="DateTime", - 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, + value_type: str = "time", + device_name: Optional[str] = None, + feature_name: str = "DateTime", + after_update_cb: Optional[AsyncCallbackType] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize RemoteValueSensor class.""" # pylint: disable=too-many-arguments try: - self.dpt_class = DateTimeType[value_type.upper()].value + self.dpt_class: Type[Union[DPTDate, DPTDateTime, DPTTime]] = DateTimeType[ + value_type.upper() + ].value except KeyError: raise ConversionError( "invalid datetime value type", @@ -58,17 +64,21 @@ def __init__( passive_group_addresses=passive_group_addresses, ) - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" return ( - isinstance(payload, DPTArray) + payload + if isinstance(payload, DPTArray) and len(payload.value) == self.dpt_class.payload_length + else None ) - def to_knx(self, value: time.struct_time): + def to_knx(self, value: time.struct_time) -> DPTArray: """Convert value to payload.""" return DPTArray(self.dpt_class.to_knx(value)) - def from_knx(self, payload) -> time.struct_time: + def from_knx(self, payload: DPTArray) -> time.struct_time: """Convert current payload to value.""" return self.dpt_class.from_knx(payload.value) diff --git a/xknx/remote_value/remote_value_dpt_2_byte_unsigned.py b/xknx/remote_value/remote_value_dpt_2_byte_unsigned.py index 7e536075b..448d51fd4 100644 --- a/xknx/remote_value/remote_value_dpt_2_byte_unsigned.py +++ b/xknx/remote_value/remote_value_dpt_2_byte_unsigned.py @@ -3,25 +3,31 @@ DPT 7.001. """ -from typing import List +from typing import TYPE_CHECKING, List, Optional, Union -from xknx.dpt import DPT2ByteUnsigned, DPTArray +from xknx.dpt import DPT2ByteUnsigned, DPTArray, DPTBinary -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueDpt2ByteUnsigned(RemoteValue): + +class RemoteValueDpt2ByteUnsigned(RemoteValue[DPTArray]): """Abstraction for remote value of KNX DPT 7.001.""" + # pylint: disable=no-self-use + def __init__( self, - xknx, - group_address=None, - group_address_state=None, - device_name=None, - feature_name="Value", - after_update_cb=None, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + device_name: Optional[str] = None, + feature_name: str = "Value", + after_update_cb: Optional[AsyncCallbackType] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX DPT 7.001.""" # pylint: disable=too-many-arguments @@ -35,14 +41,20 @@ def __init__( passive_group_addresses=passive_group_addresses, ) - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 2 + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 2 + else None + ) - def to_knx(self, value): + def to_knx(self, value: int) -> DPTArray: """Convert value to payload.""" return DPTArray(DPT2ByteUnsigned.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> int: """Convert current payload to value.""" return DPT2ByteUnsigned.from_knx(payload.value) diff --git a/xknx/remote_value/remote_value_dpt_value_1_ucount.py b/xknx/remote_value/remote_value_dpt_value_1_ucount.py index 8f7ee885e..7ce7caa9a 100644 --- a/xknx/remote_value/remote_value_dpt_value_1_ucount.py +++ b/xknx/remote_value/remote_value_dpt_value_1_ucount.py @@ -3,22 +3,32 @@ DPT 5.010. """ -from xknx.dpt import DPTArray, DPTValue1Ucount +from typing import Optional, Union + +from xknx.dpt import DPTArray, DPTBinary, DPTValue1Ucount from .remote_value import RemoteValue -class RemoteValueDptValue1Ucount(RemoteValue): +class RemoteValueDptValue1Ucount(RemoteValue[DPTArray]): """Abstraction for remote value of KNX DPT 5.010.""" - def payload_valid(self, payload): + # pylint: disable=no-self-use + + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 1 + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 1 + else None + ) - def to_knx(self, value): + def to_knx(self, value: int) -> DPTArray: """Convert value to payload.""" return DPTArray(DPTValue1Ucount.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> int: """Convert current payload to value.""" return DPTValue1Ucount.from_knx(payload.value) diff --git a/xknx/remote_value/remote_value_scaling.py b/xknx/remote_value/remote_value_scaling.py index ed8b52e4f..d76ef92e5 100644 --- a/xknx/remote_value/remote_value_scaling.py +++ b/xknx/remote_value/remote_value_scaling.py @@ -3,27 +3,31 @@ DPT 5.001. """ -from typing import List +from typing import TYPE_CHECKING, List, Optional, Union -from xknx.dpt import DPTArray +from xknx.dpt import DPTArray, DPTBinary -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueScaling(RemoteValue): + +class RemoteValueScaling(RemoteValue[DPTArray]): """Abstraction for remote value of KNX DPT 5.001 (DPT_Scaling).""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - device_name=None, - feature_name="Value", - after_update_cb=None, - range_from=0, - range_to=100, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + device_name: Optional[str] = None, + feature_name: str = "Value", + after_update_cb: Optional[AsyncCallbackType] = None, + range_from: int = 0, + range_to: int = 100, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX DPT 5.001 (DPT_Scaling).""" # pylint: disable=too-many-arguments @@ -39,30 +43,37 @@ def __init__( self.range_from = range_from self.range_to = range_to - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 1 + # pylint: disable=no-self-use + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 1 + else None + ) - def to_knx(self, value): + def to_knx(self, value: float) -> DPTArray: """Convert value to payload.""" knx_value = self._calc_to_knx(self.range_from, self.range_to, value) return DPTArray(knx_value) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> int: """Convert current payload to value.""" return self._calc_from_knx(self.range_from, self.range_to, payload.value[0]) @property - def unit_of_measurement(self): + def unit_of_measurement(self) -> Optional[str]: """Return the unit of measurement.""" return "%" @staticmethod - def _calc_from_knx(range_from, range_to, raw): + def _calc_from_knx(range_from: int, range_to: int, raw: int) -> int: delta = range_to - range_from return round((raw / 255) * delta) + range_from @staticmethod - def _calc_to_knx(range_from, range_to, value): + def _calc_to_knx(range_from: int, range_to: int, value: float) -> int: delta = range_to - range_from return round((value - range_from) / delta * 255) diff --git a/xknx/remote_value/remote_value_scene_number.py b/xknx/remote_value/remote_value_scene_number.py index 06ea56d8c..9b824dabb 100644 --- a/xknx/remote_value/remote_value_scene_number.py +++ b/xknx/remote_value/remote_value_scene_number.py @@ -3,22 +3,32 @@ DPT 17.001. """ -from xknx.dpt import DPTArray, DPTSceneNumber +from typing import Optional, Union + +from xknx.dpt import DPTArray, DPTBinary, DPTSceneNumber from .remote_value import RemoteValue -class RemoteValueSceneNumber(RemoteValue): +class RemoteValueSceneNumber(RemoteValue[DPTArray]): """Abstraction for remote value of KNX DPT 17.001 (DPT_Scene_Number).""" - def payload_valid(self, payload): + # pylint: disable=no-self-use + + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 1 + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 1 + else None + ) - def to_knx(self, value): + def to_knx(self, value: int) -> DPTArray: """Convert value to payload.""" return DPTArray(DPTSceneNumber.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> int: """Convert current payload to value.""" return DPTSceneNumber.from_knx(payload.value) diff --git a/xknx/remote_value/remote_value_sensor.py b/xknx/remote_value/remote_value_sensor.py index afde29ce8..2ca57d958 100644 --- a/xknx/remote_value/remote_value_sensor.py +++ b/xknx/remote_value/remote_value_sensor.py @@ -4,31 +4,37 @@ The module maps a given value_type to a DPT class and uses this class for serialization and deserialization of the KNX value. """ -from typing import List, Optional +from typing import TYPE_CHECKING, List, Optional, Union -from xknx.dpt import DPTArray, DPTBase +from xknx.dpt import DPTArray, DPTBase, DPTBinary from xknx.exceptions import ConversionError -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueSensor(RemoteValue): + +class RemoteValueSensor(RemoteValue[DPTArray]): """Abstraction for many different sensor DPT types.""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - sync_state=True, - value_type=None, - device_name=None, - feature_name="Value", - 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, + value_type: Optional[str] = None, + device_name: Optional[str] = None, + feature_name: str = "Value", + after_update_cb: Optional[AsyncCallbackType] = None, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize RemoteValueSensor class.""" # pylint: disable=too-many-arguments + if value_type is None: + raise ConversionError("no value type given", device_name=device_name) _dpt_class = DPTBase.parse_transcoder(value_type) if _dpt_class is None: raise ConversionError( @@ -46,20 +52,24 @@ def __init__( passive_group_addresses=passive_group_addresses, ) - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" return ( - isinstance(payload, DPTArray) + payload + if isinstance(payload, DPTArray) and len(payload.value) == self.dpt_class.payload_length + else None ) - def to_knx(self, value): + def to_knx(self, value: Union[int, float, str]) -> DPTArray: """Convert value to payload.""" return DPTArray(self.dpt_class.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> Union[int, float, str]: """Convert current payload to value.""" - return self.dpt_class.from_knx(payload.value) + return self.dpt_class.from_knx(payload.value) # type: ignore @property def unit_of_measurement(self) -> Optional[str]: @@ -69,4 +79,4 @@ def unit_of_measurement(self) -> Optional[str]: @property def ha_device_class(self) -> Optional[str]: """Return a string representing the home assistant device class.""" - return getattr(self.dpt_class, "ha_device_class", None) + return getattr(self.dpt_class, "ha_device_class", None) # type: ignore diff --git a/xknx/remote_value/remote_value_setpoint_shift.py b/xknx/remote_value/remote_value_setpoint_shift.py index d0cae883a..d53f11e04 100644 --- a/xknx/remote_value/remote_value_setpoint_shift.py +++ b/xknx/remote_value/remote_value_setpoint_shift.py @@ -3,23 +3,29 @@ DPT 6.010. """ -from typing import List +from typing import TYPE_CHECKING, List, Optional, Union -from xknx.remote_value import RemoteValue1Count +from xknx.dpt import DPTArray, DPTBinary, DPTValue1Count +from .remote_value import AsyncCallbackType, RemoteValue -class RemoteValueSetpointShift(RemoteValue1Count): +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX + + +class RemoteValueSetpointShift(RemoteValue[DPTArray]): """Abstraction for remote value of KNX DPT 6.010.""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, - device_name=None, - after_update_cb=None, - setpoint_shift_step=0.1, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + device_name: Optional[str] = None, + after_update_cb: Optional[AsyncCallbackType] = None, + setpoint_shift_step: float = 0.1, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize RemoteValueSetpointShift class.""" # pylint: disable=too-many-arguments @@ -35,12 +41,23 @@ def __init__( self.setpoint_shift_step = setpoint_shift_step - def to_knx(self, value): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: + """Test if telegram payload may be parsed.""" + # pylint: disable=no-self-use + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 1 + else None + ) + + def to_knx(self, value: float) -> DPTArray: """Convert value to payload.""" converted_value = int(value / self.setpoint_shift_step) - return super().to_knx(converted_value) + return DPTArray(DPTValue1Count.to_knx(converted_value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> float: """Convert current payload to value.""" - converted_payload = super().from_knx(payload) + converted_payload = DPTValue1Count.from_knx(payload.value) return converted_payload * self.setpoint_shift_step diff --git a/xknx/remote_value/remote_value_step.py b/xknx/remote_value/remote_value_step.py index ae62c7e6a..4acdfc31c 100644 --- a/xknx/remote_value/remote_value_step.py +++ b/xknx/remote_value/remote_value_step.py @@ -4,15 +4,19 @@ DPT 1.007. """ from enum import Enum -from typing import List +from typing import TYPE_CHECKING, List, Optional, Union -from xknx.dpt import DPTBinary +from xknx.dpt import DPTArray, DPTBinary from xknx.exceptions import ConversionError, CouldNotParseTelegram -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueStep(RemoteValue): + +class RemoteValueStep(RemoteValue[DPTBinary]): """Abstraction for remote value of KNX DPT 1.007 / DPT_Step.""" class Direction(Enum): @@ -23,14 +27,14 @@ class Direction(Enum): def __init__( self, - xknx, - group_address=None, - group_address_state=None, - device_name=None, - feature_name="Step", - after_update_cb=None, - invert=False, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + device_name: Optional[str] = None, + feature_name: str = "Step", + after_update_cb: Optional[AsyncCallbackType] = None, + invert: bool = False, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX DPT 1.007.""" # pylint: disable=too-many-arguments @@ -45,15 +49,18 @@ def __init__( ) self.invert = invert - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTBinary]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTBinary) + # pylint: disable=no-self-use + return payload if isinstance(payload, DPTBinary) else None # from KNX Association System Specifications AS v1.5.00: # 1.007 DPT_Step 0 = Decrease 1 = Increase # 1.008 DPT_UpDown 0 = Up 1 = Down - def to_knx(self, value): + def to_knx(self, value: "RemoteValueStep.Direction") -> DPTBinary: """Convert value to payload.""" if value == self.Direction.INCREASE: return DPTBinary(0) if self.invert else DPTBinary(1) @@ -66,7 +73,7 @@ def to_knx(self, value): feature_name=self.feature_name, ) - def from_knx(self, payload): + def from_knx(self, payload: DPTBinary) -> "RemoteValueStep.Direction": """Convert current payload to value.""" if payload == DPTBinary(1): return self.Direction.DECREASE if self.invert else self.Direction.INCREASE diff --git a/xknx/remote_value/remote_value_string.py b/xknx/remote_value/remote_value_string.py index 88b85a994..085a67357 100644 --- a/xknx/remote_value/remote_value_string.py +++ b/xknx/remote_value/remote_value_string.py @@ -3,25 +3,33 @@ DPT 16.000 """ -from xknx.dpt import DPTArray, DPTString +from typing import Optional, Union + +from xknx.dpt import DPTArray, DPTBinary, DPTString from .remote_value import RemoteValue -class RemoteValueString(RemoteValue): +class RemoteValueString(RemoteValue[DPTArray]): """Abstraction for remote value of KNX 16.000 (DPT_String_ASCII).""" - def payload_valid(self, payload): + # pylint: disable=no-self-use + + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" return ( - isinstance(payload, DPTArray) + payload + if isinstance(payload, DPTArray) and len(payload.value) == DPTString.payload_length + else None ) - def to_knx(self, value): + def to_knx(self, value: str) -> DPTArray: """Convert value to payload.""" return DPTArray(DPTString.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> str: """Convert current payload to value.""" return DPTString.from_knx(payload.value) diff --git a/xknx/remote_value/remote_value_switch.py b/xknx/remote_value/remote_value_switch.py index 3a7ba8aad..b2b7a6e94 100644 --- a/xknx/remote_value/remote_value_switch.py +++ b/xknx/remote_value/remote_value_switch.py @@ -3,28 +3,32 @@ DPT 1.001. """ -from typing import List, Optional +from typing import TYPE_CHECKING, List, Optional, Union -from xknx.dpt import DPTBinary +from xknx.dpt import DPTArray, DPTBinary from xknx.exceptions import ConversionError, CouldNotParseTelegram -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueSwitch(RemoteValue): + +class RemoteValueSwitch(RemoteValue[DPTBinary]): """Abstraction for remote value of KNX DPT 1.001 / DPT_Switch.""" def __init__( self, - xknx, - group_address=None, - group_address_state=None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, sync_state: bool = True, - device_name: str = None, + device_name: Optional[str] = None, feature_name: str = "State", - after_update_cb=None, - invert: Optional[bool] = False, - passive_group_addresses: List[str] = None, + after_update_cb: Optional[AsyncCallbackType] = None, + invert: bool = False, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX DPT 1.001.""" # pylint: disable=too-many-arguments @@ -40,11 +44,14 @@ def __init__( ) self.invert = bool(invert) - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTBinary]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTBinary) + # pylint: disable=no-self-use + return payload if isinstance(payload, DPTBinary) else None - def to_knx(self, value: bool): + def to_knx(self, value: bool) -> DPTBinary: """Convert value to payload.""" if isinstance(value, bool): return DPTBinary(value ^ self.invert) diff --git a/xknx/remote_value/remote_value_temp.py b/xknx/remote_value/remote_value_temp.py index f20641396..5c6ff5d5f 100644 --- a/xknx/remote_value/remote_value_temp.py +++ b/xknx/remote_value/remote_value_temp.py @@ -3,22 +3,32 @@ DPT 9.001. """ -from xknx.dpt import DPTArray, DPTTemperature +from typing import Optional, Union + +from xknx.dpt import DPTArray, DPTBinary, DPTTemperature from .remote_value import RemoteValue -class RemoteValueTemp(RemoteValue): +class RemoteValueTemp(RemoteValue[DPTArray]): """Abstraction for remote value of KNX 9.001 (DPT_Value_Temp).""" - def payload_valid(self, payload): + # pylint: disable=no-self-use + + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTArray]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTArray) and len(payload.value) == 2 + return ( + payload + if isinstance(payload, DPTArray) and len(payload.value) == 2 + else None + ) - def to_knx(self, value): + def to_knx(self, value: float) -> DPTArray: """Convert value to payload.""" return DPTArray(DPTTemperature.to_knx(value)) - def from_knx(self, payload): + def from_knx(self, payload: DPTArray) -> float: """Convert current payload to value.""" return DPTTemperature.from_knx(payload.value) diff --git a/xknx/remote_value/remote_value_updown.py b/xknx/remote_value/remote_value_updown.py index 7fbd3ad86..8d5a96144 100644 --- a/xknx/remote_value/remote_value_updown.py +++ b/xknx/remote_value/remote_value_updown.py @@ -4,15 +4,19 @@ DPT 1.008. """ from enum import Enum -from typing import List +from typing import TYPE_CHECKING, List, Optional, Union -from xknx.dpt import DPTBinary +from xknx.dpt import DPTArray, DPTBinary from xknx.exceptions import ConversionError, CouldNotParseTelegram -from .remote_value import RemoteValue +from .remote_value import AsyncCallbackType, RemoteValue +if TYPE_CHECKING: + from xknx.telegram.address import GroupAddressableType + from xknx.xknx import XKNX -class RemoteValueUpDown(RemoteValue): + +class RemoteValueUpDown(RemoteValue[DPTBinary]): """Abstraction for remote value of KNX DPT 1.008 / DPT_UpDown.""" class Direction(Enum): @@ -24,14 +28,14 @@ class Direction(Enum): def __init__( self, - xknx, - group_address=None, - group_address_state=None, - device_name=None, - feature_name="Up/Down", - after_update_cb=None, - invert=False, - passive_group_addresses: List[str] = None, + xknx: "XKNX", + group_address: Optional["GroupAddressableType"] = None, + group_address_state: Optional["GroupAddressableType"] = None, + device_name: Optional[str] = None, + feature_name: str = "Up/Down", + after_update_cb: Optional[AsyncCallbackType] = None, + invert: bool = False, + passive_group_addresses: Optional[List["GroupAddressableType"]] = None, ): """Initialize remote value of KNX DPT 1.008.""" # pylint: disable=too-many-arguments @@ -46,11 +50,14 @@ def __init__( ) self.invert = invert - def payload_valid(self, payload): + def payload_valid( + self, payload: Optional[Union[DPTArray, DPTBinary]] + ) -> Optional[DPTBinary]: """Test if telegram payload may be parsed.""" - return isinstance(payload, DPTBinary) + # pylint: disable=no-self-use + return payload if isinstance(payload, DPTBinary) else None - def to_knx(self, value): + def to_knx(self, value: "RemoteValueUpDown.Direction") -> DPTBinary: """Convert value to payload.""" if value == self.Direction.UP: return DPTBinary(1) if self.invert else DPTBinary(0) @@ -63,7 +70,7 @@ def to_knx(self, value): feature_name=self.feature_name, ) - def from_knx(self, payload): + def from_knx(self, payload: DPTBinary) -> "RemoteValueUpDown.Direction": """Convert current payload to value.""" if payload == DPTBinary(0): return self.Direction.DOWN if self.invert else self.Direction.UP