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.

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

Refactor/clean-up parts of the code #71

Merged
merged 1 commit into from
Oct 24, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 3 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ There an additional class called `FreeAtHome`. This class attempts to put it all

### Get Devices

This example will load the `FreeAtHome` class with all potential devices from the api. Once loaded another function `get_device_by_class` is used to pull all devices that call under a specific "class".
This example will load the `FreeAtHome` class with all potential devices from the api. Once loaded another function `get_devices_by_class` is used to pull all devices that call under a specific "class".

```python
from abbfreeathome.api import FreeAtHomeApi
Expand All @@ -223,7 +223,7 @@ if __name__ == "__main__":
asyncio.run(_free_at_home.load_devices())

# Fetch just the list of switches
_switches = _free_at_home.get_device_by_class(device_class=SwitchActuator)
_switches = _free_at_home.get_devices_by_class(device_class=SwitchActuator)

# Loop through each switch showing the name and whether is On/Off
for _switch in _switches:
Expand Down Expand Up @@ -261,7 +261,7 @@ if __name__ == "__main__":
asyncio.run(_free_at_home.load_devices())

# Set the callback function on each switch.
for _switch in _free_at_home.get_device_by_class(device_class=SwitchActuator):
for _switch in _free_at_home.get_devices_by_class(device_class=SwitchActuator):
_switch.register_callback(my_very_own_callback)

# Start listenting
Expand Down
26 changes: 26 additions & 0 deletions src/abbfreeathome/const.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
"""Constants for the FreeAtHome package."""

from .bin.function import Function
from .devices.base import Base
from .devices.carbon_monoxide_sensor import CarbonMonoxideSensor
from .devices.des_door_ringing_sensor import DesDoorRingingSensor
from .devices.dimming_actuator import DimmingActuator
from .devices.movement_detector import MovementDetector
from .devices.smoke_detector import SmokeDetector
from .devices.switch_actuator import SwitchActuator
from .devices.switch_sensor import SwitchSensor
from .devices.trigger import Trigger
from .devices.window_door_sensor import WindowDoorSensor

FUNCTION_DEVICE_MAPPING: dict[Function, Base] = {
Function.FID_CARBON_MONOXIDE_SENSOR: CarbonMonoxideSensor,
Function.FID_DES_DOOR_RINGING_SENSOR: DesDoorRingingSensor,
Function.FID_DIMMING_ACTUATOR: DimmingActuator,
Function.FID_MOVEMENT_DETECTOR: MovementDetector,
Function.FID_SMOKE_DETECTOR: SmokeDetector,
Function.FID_SWITCH_ACTUATOR: SwitchActuator,
Function.FID_SWITCH_SENSOR: SwitchSensor,
Function.FID_TRIGGER: Trigger,
Function.FID_WINDOW_DOOR_POSITION_SENSOR: WindowDoorSensor,
Function.FID_WINDOW_DOOR_SENSOR: WindowDoorSensor,
}
101 changes: 26 additions & 75 deletions src/abbfreeathome/freeathome.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,24 +3,13 @@
from .api import FreeAtHomeApi
from .bin.function import Function
from .bin.interface import Interface
from .const import FUNCTION_DEVICE_MAPPING
from .devices.base import Base
from .devices.carbon_monoxide_sensor import CarbonMonoxideSensor
from .devices.des_door_ringing_sensor import DesDoorRingingSensor
from .devices.dimming_actuator import DimmingActuator
from .devices.movement_detector import MovementDetector
from .devices.smoke_detector import SmokeDetector
from .devices.switch_actuator import SwitchActuator
from .devices.switch_sensor import SwitchSensor
from .devices.trigger import Trigger
from .devices.window_door_sensor import WindowDoorSensor


class FreeAtHome:
"""Provides a class for interacting with the ABB-free@home API."""

_config: dict | None = None
_devices: dict = {}

def __init__(
self,
api: FreeAtHomeApi,
Expand All @@ -29,11 +18,14 @@ def __init__(
include_orphan_channels: bool = False,
) -> None:
"""Initialize the FreeAtHome class."""
self._config: dict | None = None
self._devices: dict[str, Base] = {}

self.api: FreeAtHomeApi = api

self._interfaces: list[Interface] = interfaces
self._device_classes: list[Base] = device_classes
self._include_orphan_channels = include_orphan_channels
self._include_orphan_channels: bool = include_orphan_channels

def clear_devices(self):
"""Clear all devices in the device list."""
Expand All @@ -46,6 +38,18 @@ async def get_config(self, refresh: bool = False) -> dict:

return self._config

def get_devices(self) -> dict[str, Base]:
"""Get the list of devices."""
return self._devices

def get_devices_by_class(self, device_class: Base) -> list[Base]:
"""Get the list of devices by class."""
return [
_device
for _device in self._devices.values()
if isinstance(_device, device_class)
]

async def get_devices_by_function(self, function: Function) -> list[dict]:
"""Get the list of devices by function."""
_devices = []
Expand Down Expand Up @@ -128,21 +132,11 @@ async def get_room_name(
.get("name")
)

def get_device_by_class(self, device_class: Base) -> list[Base]:
"""Get the list of devices by class."""
return [
_device
for _device in self._devices.values()
if isinstance(_device, device_class)
]

async def load_devices(self):
"""Load all of the devices into the devices object."""
self.clear_devices()
for _mapping in self._get_function_to_device_mapping():
await self._load_devices_by_function(
_mapping.get("function"), _mapping.get("device_class")
)
for _function, _device_class in self._get_function_to_device_mapping().items():
await self._load_devices_by_function(_function, _device_class)

def unload_device_by_device_serial(self, device_serial: str):
"""Unload all devices by device serial id."""
Expand Down Expand Up @@ -189,56 +183,13 @@ async def update_device(self, data: dict):
except KeyError:
continue

def _get_function_to_device_mapping(self) -> list[dict[str, Function | Base]]:
_function_to_device_mapping = [
{
"function": Function.FID_SWITCH_ACTUATOR,
"device_class": SwitchActuator,
},
{
"function": Function.FID_SWITCH_SENSOR,
"device_class": SwitchSensor,
},
{
"function": Function.FID_TRIGGER,
"device_class": Trigger,
},
{
"function": Function.FID_MOVEMENT_DETECTOR,
"device_class": MovementDetector,
},
{
"function": Function.FID_DIMMING_ACTUATOR,
"device_class": DimmingActuator,
},
{
"function": Function.FID_WINDOW_DOOR_SENSOR,
"device_class": WindowDoorSensor,
},
{
"function": Function.FID_WINDOW_DOOR_POSITION_SENSOR,
"device_class": WindowDoorSensor,
},
{
"function": Function.FID_DES_DOOR_RINGING_SENSOR,
"device_class": DesDoorRingingSensor,
},
{
"function": Function.FID_SMOKE_DETECTOR,
"device_class": SmokeDetector,
},
{
"function": Function.FID_CARBON_MONOXIDE_SENSOR,
"device_class": CarbonMonoxideSensor,
},
]

def _get_function_to_device_mapping(self) -> dict[Function, Base]:
return (
_function_to_device_mapping
FUNCTION_DEVICE_MAPPING
if not self._device_classes
else [
_mapping
for _mapping in _function_to_device_mapping
if _mapping.get("device_class") in self._device_classes
]
else {
key: value
for key, value in FUNCTION_DEVICE_MAPPING.items()
if value in self._device_classes
}
)
43 changes: 23 additions & 20 deletions tests/test_freeathome.py
Original file line number Diff line number Diff line change
Expand Up @@ -323,7 +323,7 @@ def test_get_device_by_class(freeathome):
"""Test the get_device_class function."""
device = MagicMock(spec=SwitchActuator)
freeathome._devices = {"device1": device}
devices = freeathome.get_device_by_class(SwitchActuator)
devices = freeathome.get_devices_by_class(SwitchActuator)
assert devices == [device]


Expand All @@ -332,43 +332,46 @@ async def test_load_devices(freeathome):
"""Test the load_devices function."""
await freeathome.load_devices()

# Get the dict of devices
devices = freeathome.get_devices()

# Verify that the devices are loaded correctly
assert len(freeathome._devices) == 4
assert len(devices) == 4

# Check a single device
device_key = "ABB7F500E17A/ch0003"
assert device_key in freeathome._devices
assert isinstance(freeathome._devices[device_key], SwitchActuator)
assert freeathome._devices[device_key].device_name == "Study Area Rocker"
assert freeathome._devices[device_key].channel_name == "Study Area Light"
assert freeathome._devices[device_key].floor_name == "Ground Floor"
assert freeathome._devices[device_key].room_name == "Living Room"
assert device_key in devices
assert isinstance(devices[device_key], SwitchActuator)
assert devices[device_key].device_name == "Study Area Rocker"
assert devices[device_key].channel_name == "Study Area Light"
assert devices[device_key].floor_name == "Ground Floor"
assert devices[device_key].room_name == "Living Room"

# Unload a single device and test it's been removed
freeathome.unload_device_by_device_serial(device_serial="ABB7F62F6C0B")
assert len(freeathome._devices) == 2
devices = freeathome.get_devices()
assert len(devices) == 2


@pytest.mark.asyncio
async def test_load_devices_with_orphans(freeathome_orphans):
"""Test the load_devices function."""
await freeathome_orphans.load_devices()

# Get the dict of devices
devices = freeathome_orphans.get_devices()

# Verify that the devices are loaded correctly
assert len(freeathome_orphans._devices) == 6
assert len(devices) == 6

# Check a single orphan device
device_key = "ABB28CBC3651/ch0006"
assert device_key in freeathome_orphans._devices
assert isinstance(freeathome_orphans._devices[device_key], SwitchSensor)
assert (
freeathome_orphans._devices[device_key].device_name == "Sensor/switch actuator"
)
assert (
freeathome_orphans._devices[device_key].channel_name == "Sensor/switch actuator"
)
assert freeathome_orphans._devices[device_key].floor_name is None
assert freeathome_orphans._devices[device_key].room_name is None
assert device_key in devices
assert isinstance(devices[device_key], SwitchSensor)
assert devices[device_key].device_name == "Sensor/switch actuator"
assert devices[device_key].channel_name == "Sensor/switch actuator"
assert devices[device_key].floor_name is None
assert devices[device_key].room_name is None


@pytest.mark.asyncio
Expand Down
Loading