From f1d8557d70047d90ad924add812657da90ab9377 Mon Sep 17 00:00:00 2001 From: Adam Kingsley Date: Fri, 18 Oct 2024 20:41:07 +0200 Subject: [PATCH] Make a number of refactor changes to make adding/updating devices quicker and easier --- src/abbfreeathome/devices/base.py | 28 ++++++++ src/abbfreeathome/devices/dimming_actuator.py | 39 ++-------- .../devices/movement_detector.py | 44 +++--------- src/abbfreeathome/devices/switch_actuator.py | 35 ++------- src/abbfreeathome/devices/switch_sensor.py | 35 ++------- .../devices/window_door_sensor.py | 35 ++------- tests/test_movement_detector.py | 72 ++++++++++++------- 7 files changed, 107 insertions(+), 181 deletions(-) diff --git a/src/abbfreeathome/devices/base.py b/src/abbfreeathome/devices/base.py index b81bad1..fe1941b 100644 --- a/src/abbfreeathome/devices/base.py +++ b/src/abbfreeathome/devices/base.py @@ -14,6 +14,9 @@ class Base: """Free@Home Base Class.""" + _state_refresh_output_pairings: list[Pairing] = [] + _state_refresh_input_pairings: list[Pairing] = [] + def __init__( self, device_id: str, @@ -40,6 +43,9 @@ def __init__( self._room_name = room_name self._callbacks = set() + # Set the initial state of the switch based on output + self._refresh_state_from_outputs() + @property def device_id(self) -> str: """Get the device id.""" @@ -118,6 +124,28 @@ def remove_callback(self, callback: Callable[[], None]) -> None: """Remove previously registered callback.""" self._callbacks.discard(callback) + async def refresh_state(self): + """Refresh the state of the device from the api.""" + for _pairing in self._state_refresh_output_pairings: + _switch_output_id, _switch_output_value = self.get_output_by_pairing( + pairing=_pairing + ) + + _datapoint = ( + await self._api.get_datapoint( + device_id=self.device_id, + channel_id=self.channel_id, + datapoint=_switch_output_id, + ) + )[0] + + self._refresh_state_from_output( + output={ + "pairingID": _pairing.value, + "value": _datapoint, + } + ) + def _refresh_state_from_input(self, input: dict[str, Any]) -> bool: """Refresh the state of the device a single input.""" diff --git a/src/abbfreeathome/devices/dimming_actuator.py b/src/abbfreeathome/devices/dimming_actuator.py index 9be49c2..e4b83e3 100644 --- a/src/abbfreeathome/devices/dimming_actuator.py +++ b/src/abbfreeathome/devices/dimming_actuator.py @@ -10,8 +10,10 @@ class DimmingActuator(Base): """Free@Home DimmingActuator Class.""" - _state = None - _brightness = None + _state_refresh_output_pairings: list[Pairing] = [ + Pairing.AL_INFO_ON_OFF, + Pairing.AL_INFO_ACTUAL_DIMMING_VALUE, + ] def __init__( self, @@ -27,6 +29,9 @@ def __init__( room_name: str | None = None, ) -> None: """Initialize the Free@Home DimmingActuator class.""" + self._state: bool | None = None + self._brightness: int | None = None + super().__init__( device_id, device_name, @@ -40,9 +45,6 @@ def __init__( room_name, ) - # Set the initial state of the switch based on output - self._refresh_state_from_outputs() - @property def state(self) -> bool: """Get the state of the dimming actuator.""" @@ -76,33 +78,6 @@ async def set_brightness(self, value: int): await self._set_brightness_datapoint(str(value)) self._brightness = value - async def refresh_state(self): - """Refresh the state of the device from the api.""" - _state_refresh_pairings = [ - Pairing.AL_INFO_ON_OFF, - Pairing.AL_INFO_ACTUAL_DIMMING_VALUE, - ] - - for _pairing in _state_refresh_pairings: - _switch_output_id, _switch_output_value = self.get_output_by_pairing( - pairing=_pairing - ) - - _datapoint = ( - await self._api.get_datapoint( - device_id=self.device_id, - channel_id=self.channel_id, - datapoint=_switch_output_id, - ) - )[0] - - self._refresh_state_from_output( - output={ - "pairingID": _pairing.value, - "value": _datapoint, - } - ) - def _refresh_state_from_output(self, output: dict[str, Any]) -> bool: """ Refresh the state of the device from a given output. diff --git a/src/abbfreeathome/devices/movement_detector.py b/src/abbfreeathome/devices/movement_detector.py index 490acf9..90cb77e 100644 --- a/src/abbfreeathome/devices/movement_detector.py +++ b/src/abbfreeathome/devices/movement_detector.py @@ -10,7 +10,10 @@ class MovementDetector(Base): """Free@Home SwitchActuator Class.""" - _state = None + _state_refresh_output_pairings: list[Pairing] = [ + Pairing.AL_BRIGHTNESS_LEVEL, + Pairing.AL_TIMED_MOVEMENT, + ] def __init__( self, @@ -26,6 +29,9 @@ def __init__( room_name: str | None = None, ) -> None: """Initialize the Free@Home SwitchActuator class.""" + self._state: bool | None = None + self._brightness: float | None = None + super().__init__( device_id, device_name, @@ -39,45 +45,15 @@ def __init__( room_name, ) - # Set the initial state of the switch based on output - self._refresh_state_from_outputs() - @property def state(self) -> bool | None: """Get the movement state.""" return self._state @property - def brightness(self) -> float: + def brightness(self) -> float | None: """Get the brightness level of the sensor.""" - return float(self._brightness) - - async def refresh_state(self): - """Refresh the state of the device from the api.""" - _state_refresh_pairings = [ - Pairing.AL_BRIGHTNESS_LEVEL, - Pairing.AL_TIMED_MOVEMENT, - ] - - for _pairing in _state_refresh_pairings: - _switch_output_id, _switch_output_value = self.get_output_by_pairing( - pairing=_pairing - ) - - _datapoint = ( - await self._api.get_datapoint( - device_id=self.device_id, - channel_id=self.channel_id, - datapoint=_switch_output_id, - ) - )[0] - - self._refresh_state_from_output( - output={ - "pairingID": _pairing.value, - "value": _datapoint, - } - ) + return self._brightness def _refresh_state_from_output(self, output: dict[str, Any]) -> bool: """ @@ -89,6 +65,6 @@ def _refresh_state_from_output(self, output: dict[str, Any]) -> bool: self._state = output.get("value") == "1" return True if output.get("pairingID") == Pairing.AL_BRIGHTNESS_LEVEL.value: - self._brightness = output.get("value") + self._brightness = float(output.get("value")) return True return False diff --git a/src/abbfreeathome/devices/switch_actuator.py b/src/abbfreeathome/devices/switch_actuator.py index d275dac..d1ef1db 100644 --- a/src/abbfreeathome/devices/switch_actuator.py +++ b/src/abbfreeathome/devices/switch_actuator.py @@ -10,7 +10,9 @@ class SwitchActuator(Base): """Free@Home SwitchActuator Class.""" - _state = None + _state_refresh_output_pairings: list[Pairing] = [ + Pairing.AL_INFO_ON_OFF, + ] def __init__( self, @@ -26,6 +28,8 @@ def __init__( room_name: str | None = None, ) -> None: """Initialize the Free@Home SwitchActuator class.""" + self._state: bool | None = None + super().__init__( device_id, device_name, @@ -39,9 +43,6 @@ def __init__( room_name, ) - # Set the initial state of the switch based on output - self._refresh_state_from_outputs() - @property def state(self) -> bool | None: """Get the state of the switch.""" @@ -57,32 +58,6 @@ async def turn_off(self): await self._set_switching_datapoint("0") self._state = False - async def refresh_state(self): - """Refresh the state of the device from the api.""" - _state_refresh_pairings = [ - Pairing.AL_INFO_ON_OFF, - ] - - for _pairing in _state_refresh_pairings: - _switch_output_id, _switch_output_value = self.get_output_by_pairing( - pairing=_pairing - ) - - _datapoint = ( - await self._api.get_datapoint( - device_id=self.device_id, - channel_id=self.channel_id, - datapoint=_switch_output_id, - ) - )[0] - - self._refresh_state_from_output( - output={ - "pairingID": _pairing.value, - "value": _datapoint, - } - ) - def _refresh_state_from_output(self, output: dict[str, Any]) -> bool: """ Refresh the state of the device from a given output. diff --git a/src/abbfreeathome/devices/switch_sensor.py b/src/abbfreeathome/devices/switch_sensor.py index d06ec96..c32e2e4 100644 --- a/src/abbfreeathome/devices/switch_sensor.py +++ b/src/abbfreeathome/devices/switch_sensor.py @@ -10,7 +10,9 @@ class SwitchSensor(Base): """Free@Home SwitchSensor Class.""" - _state = None + _state_refresh_output_pairings: list[Pairing] = [ + Pairing.AL_SWITCH_ON_OFF, + ] def __init__( self, @@ -26,6 +28,8 @@ def __init__( room_name: str | None = None, ) -> None: """Initialize the Free@Home SwitchSensor class.""" + self._state: bool | None = None + super().__init__( device_id, device_name, @@ -39,40 +43,11 @@ def __init__( room_name, ) - # Set the initial state of the switch based on output - self._refresh_state_from_outputs() - @property def state(self) -> bool | None: """Get the switch state.""" return self._state - async def refresh_state(self): - """Refresh the state of the device from the api.""" - _state_refresh_pairings = [ - Pairing.AL_SWITCH_ON_OFF, - ] - - for _pairing in _state_refresh_pairings: - _switch_output_id, _switch_output_value = self.get_output_by_pairing( - pairing=_pairing - ) - - _datapoint = ( - await self._api.get_datapoint( - device_id=self.device_id, - channel_id=self.channel_id, - datapoint=_switch_output_id, - ) - )[0] - - self._refresh_state_from_output( - output={ - "pairingID": _pairing.value, - "value": _datapoint, - } - ) - def _refresh_state_from_output(self, output: dict[str, Any]) -> bool: """ Refresh the state of the device from a given output. diff --git a/src/abbfreeathome/devices/window_door_sensor.py b/src/abbfreeathome/devices/window_door_sensor.py index 4f2e957..083c1be 100644 --- a/src/abbfreeathome/devices/window_door_sensor.py +++ b/src/abbfreeathome/devices/window_door_sensor.py @@ -10,7 +10,9 @@ class WindowDoorSensor(Base): """Free@Home WindowDoorSensor Class.""" - _state = None + _state_refresh_output_pairings: list[Pairing] = [ + Pairing.AL_WINDOW_DOOR, + ] def __init__( self, @@ -26,6 +28,8 @@ def __init__( room_name: str | None = None, ) -> None: """Initialize the Free@Home SwitchSensor class.""" + self._state: bool | None = None + super().__init__( device_id, device_name, @@ -39,40 +43,11 @@ def __init__( room_name, ) - # Set the initial state of the switch based on output - self._refresh_state_from_outputs() - @property def state(self) -> bool | None: """Get the sensor state.""" return self._state - async def refresh_state(self): - """Refresh the state of the device from the api.""" - _state_refresh_pairings = [ - Pairing.AL_WINDOW_DOOR, - ] - - for _pairing in _state_refresh_pairings: - _sensor_output_id, _sensor_output_value = self.get_output_by_pairing( - pairing=_pairing - ) - - _datapoint = ( - await self._api.get_datapoint( - device_id=self.device_id, - channel_id=self.channel_id, - datapoint=_sensor_output_id, - ) - )[0] - - self._refresh_state_from_output( - output={ - "pairingID": _pairing.value, - "value": _datapoint, - } - ) - def _refresh_state_from_output(self, output: dict[str, Any]) -> bool: """ Refresh the state of the device from a given output. diff --git a/tests/test_movement_detector.py b/tests/test_movement_detector.py index 60730df..6b8e438 100644 --- a/tests/test_movement_detector.py +++ b/tests/test_movement_detector.py @@ -8,15 +8,8 @@ from src.abbfreeathome.devices.movement_detector import MovementDetector -@pytest.fixture -def mock_api(): - """Create a mock api function.""" - return AsyncMock(spec=FreeAtHomeApi) - - -@pytest.fixture -def movement_detector(mock_api): - """Set up the switch instance for testing the SwitchActuator device.""" +def get_movement_detector(type: str, mock_api): + """Get the MovementDetector class to be tested against.""" inputs = {"idp0000": {"pairingID": 256, "value": "0"}} outputs = { "odp0000": {"pairingID": 6, "value": "0"}, @@ -25,6 +18,10 @@ def movement_detector(mock_api): } parameters = {"par0034": "1", "par00d5": "100"} + # If it's outdoor it won't have brightness + if type == "outdoor": + outputs.pop("odp0002") + return MovementDetector( device_id="ABB7F500E17A", device_name="Device Name", @@ -37,43 +34,68 @@ def movement_detector(mock_api): ) +@pytest.fixture +def mock_api(): + """Create a mock api function.""" + return AsyncMock(spec=FreeAtHomeApi) + + +@pytest.fixture +def movement_detector_indoor(mock_api): + """Set up the switch instance for testing the SwitchActuator device.""" + return get_movement_detector("indoor", mock_api) + + +@pytest.fixture +def movement_detector_outdoor(mock_api): + """Set up the switch instance for testing the SwitchActuator device.""" + return get_movement_detector("outdoor", mock_api) + + +@pytest.mark.asyncio +async def test_initial_state_indoor(movement_detector_indoor): + """Test the intial state of the switch.""" + assert movement_detector_indoor.state is False + assert movement_detector_indoor.brightness == 1.6 + + @pytest.mark.asyncio -async def test_initial_state(movement_detector): +async def test_initial_state_outdoor(movement_detector_outdoor): """Test the intial state of the switch.""" - assert movement_detector.state is False - assert movement_detector.brightness == 1.6 + assert movement_detector_outdoor.state is False + assert movement_detector_outdoor.brightness is None @pytest.mark.asyncio -async def test_refresh_state(movement_detector): +async def test_refresh_state(movement_detector_indoor): """Test refreshing the state of the switch.""" - movement_detector._api.get_datapoint.return_value = ["1"] - await movement_detector.refresh_state() - assert movement_detector.state is True - assert movement_detector.brightness == 1.0 - movement_detector._api.get_datapoint.assert_called_with( + movement_detector_indoor._api.get_datapoint.return_value = ["1"] + await movement_detector_indoor.refresh_state() + assert movement_detector_indoor.state is True + assert movement_detector_indoor.brightness == 1.0 + movement_detector_indoor._api.get_datapoint.assert_called_with( device_id="ABB7F500E17A", channel_id="ch0003", datapoint="odp0000", ) -def test_refresh_state_from_output(movement_detector): +def test_refresh_state_from_output(movement_detector_indoor): """Test the _refresh_state_from_output function.""" # Check output that affects the state. - movement_detector._refresh_state_from_output( + movement_detector_indoor._refresh_state_from_output( output={"pairingID": 6, "value": "1"}, ) - assert movement_detector.state is True + assert movement_detector_indoor.state is True # Check output that affects the state. - movement_detector._refresh_state_from_output( + movement_detector_indoor._refresh_state_from_output( output={"pairingID": 1027, "value": "52.3"}, ) - assert movement_detector.brightness == 52.3 + assert movement_detector_indoor.brightness == 52.3 # Check output that does NOT affect the state. - movement_detector._refresh_state_from_output( + movement_detector_indoor._refresh_state_from_output( output={"pairingID": 6, "value": "0"}, ) - assert movement_detector.brightness == 52.3 + assert movement_detector_indoor.brightness == 52.3