Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Make a number of refactors to how the state of devices is set. #53

Merged
merged 2 commits into from
Oct 18, 2024
Merged

Make a number of refactors to how the state of devices is set. #53

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f1d8557
Make a number of refactor changes to make adding/updating devices qui…
kingsleyadam Oct 18, 2024
ddce49e
Small comments/naming convention updates
kingsleyadam Oct 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 27 additions & 1 deletion src/abbfreeathome/devices/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -40,6 +43,9 @@ def __init__(
self._room_name = room_name
self._callbacks = set()

# Set the initial state of the device based on output
self._refresh_state_from_outputs()

@property
def device_id(self) -> str:
"""Get the device id."""
Expand Down Expand Up @@ -111,13 +117,33 @@ def update_device(self, datapoint_key: str, datapoint_value: str):
callback()

def register_callback(self, callback: Callable[[], None]) -> None:
"""Register callback, called when switch changes state."""
"""Register callback, called when device changes state."""
self._callbacks.add(callback)

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:
_output_id, _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=_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."""

Expand Down
39 changes: 7 additions & 32 deletions src/abbfreeathome/devices/dimming_actuator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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."""
Expand Down Expand Up @@ -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.
Expand Down
44 changes: 10 additions & 34 deletions src/abbfreeathome/devices/movement_detector.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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:
"""
Expand All @@ -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
35 changes: 5 additions & 30 deletions src/abbfreeathome/devices/switch_actuator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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."""
Expand All @@ -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.
Expand Down
35 changes: 5 additions & 30 deletions src/abbfreeathome/devices/switch_sensor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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.
Expand Down
35 changes: 5 additions & 30 deletions src/abbfreeathome/devices/window_door_sensor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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,
Expand All @@ -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.
Expand Down
Loading
Loading