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 enums, freeathome, pairing #36

Merged
merged 1 commit into from
Oct 14, 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
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import enum


class FunctionID(enum.Enum):
class Function(enum.Enum):
"""An Enum class for all Free@Home functions."""

FID_SWITCH_SENSOR = 0
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,10 @@
import enum


class PairingId(enum.Enum):
class Pairing(enum.Enum):
"""An Enum class for all Free@Home pairings."""

AL_INVALID = 0
AL_SWITCH_ON_OFF = 1
AL_TIMED_START_STOP = 2
AL_FORCED = 3
Expand All @@ -21,6 +22,7 @@ class PairingId(enum.Enum):
AL_ABSOLUTE_SET_VALUE_CONTROL = 17
AL_NIGHT = 18
AL_RESET_ERROR = 19
AL_NIGHT_ACTUATOR_FOR_SYSAP = 20
AL_RGB = 21
AL_COLOR_TEMPERATURE = 22
AL_HSV = 23
Expand Down Expand Up @@ -330,4 +332,3 @@ class PairingId(enum.Enum):
AL_MEASURED_TEMPERATURE_3 = 65283
AL_MEASURED_TEMPERATURE_4 = 65284
AL_IGNORE = 65534
AL_INVALID = 65535
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import enum # pragma: no cover


class ParameterId(enum.Enum): # pragma: no cover
class Parameter(enum.Enum): # pragma: no cover
"""An Enum class for all Free@Home parameters."""

PID_LED_DAY_BRIGHTNESS = 1
Expand Down
19 changes: 12 additions & 7 deletions src/abbfreeathome/devices/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,8 @@
from typing import Any

from ..api import FreeAtHomeApi
from ..exceptions import InvalidDeviceChannelPairingId
from ..bin.pairing import Pairing
from ..exceptions import InvalidDeviceChannelPairing


class Base:
Expand Down Expand Up @@ -67,21 +68,25 @@ def room_name(self) -> str | None:
"""Get the room name of the device."""
return self._room_name

def get_input_by_pairing_id(self, pairing_id: int) -> tuple[str, Any]:
def get_input_by_pairing(self, pairing: Pairing) -> tuple[str, Any]:
"""Get the channel input by pairing id."""
for _input_id, _input in self._inputs.items():
if _input.get("pairingID") == pairing_id:
if _input.get("pairingID") == pairing.value:
return _input_id, _input.get("value")

raise InvalidDeviceChannelPairingId(self.device_id, self.channel_id, pairing_id)
raise InvalidDeviceChannelPairing(
self.device_id, self.channel_id, pairing.value
)

def get_output_by_pairing_id(self, pairing_id: int) -> tuple[str, Any]:
def get_output_by_pairing(self, pairing: Pairing) -> tuple[str, Any]:
"""Get the channel output by pairing id."""
for _output_id, _output in self._outputs.items():
if _output.get("pairingID") == pairing_id:
if _output.get("pairingID") == pairing.value:
return _output_id, _output.get("value")

raise InvalidDeviceChannelPairingId(self.device_id, self.channel_id, pairing_id)
raise InvalidDeviceChannelPairing(
self.device_id, self.channel_id, pairing.value
)

def update_device():
"""Update a devices state."""
Expand Down
14 changes: 7 additions & 7 deletions src/abbfreeathome/devices/switch_actuator.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from typing import Any

from ..api import FreeAtHomeApi
from ..bin.pairing_id import PairingId
from ..bin.pairing import Pairing
from .base import Base

_LOGGER = logging.getLogger(__name__)
Expand Down Expand Up @@ -62,8 +62,8 @@ async def turn_off(self):

async def refresh_state(self):
"""Refresh the state of the switch from the api."""
_switch_output_id, _switch_output_value = self.get_output_by_pairing_id(
pairing_id=PairingId.AL_INFO_ON_OFF.value
_switch_output_id, _switch_output_value = self.get_output_by_pairing(
pairing=Pairing.AL_INFO_ON_OFF
)

_datapoint = (
Expand All @@ -81,14 +81,14 @@ def _refresh_state_from_inputs(self):

def _refresh_state_from_outputs(self):
"""Refresh the state of the switch from the _outputs."""
_switch_output_id, _switch_output_value = self.get_output_by_pairing_id(
pairing_id=PairingId.AL_INFO_ON_OFF.value
_switch_output_id, _switch_output_value = self.get_output_by_pairing(
pairing=Pairing.AL_INFO_ON_OFF
)
self._state = _switch_output_value == "1"

async def _set_switching_datapoint(self, value: str):
_switch_input_id, _switch_input_value = self.get_input_by_pairing_id(
pairing_id=PairingId.AL_SWITCH_ON_OFF.value
_switch_input_id, _switch_input_value = self.get_input_by_pairing(
pairing=Pairing.AL_SWITCH_ON_OFF
)
return await self._api.set_datapoint(
device_id=self.device_id,
Expand Down
8 changes: 4 additions & 4 deletions src/abbfreeathome/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -52,14 +52,14 @@ def __init__(self, status_code: int) -> None:
super().__init__(self.message)


class InvalidDeviceChannelPairingId(FreeAtHomeException):
class InvalidDeviceChannelPairing(FreeAtHomeException):
"""Raise an exception for an invalid pairing id."""

def __init__(self, device_id: str, channel_id: str, pairing_id: int) -> None:
"""Initialze the InvalidDeviceChannelPairingId class."""
def __init__(self, device_id: str, channel_id: str, pairing_value: int) -> None:
"""Initialze the InvalidDeviceChannelPairing class."""
self.message = (
f"Could not find paring id for "
f"device: {device_id}; channel: {channel_id}; pairing id: {pairing_id}"
f"device: {device_id}; channel: {channel_id}; pairing id: {pairing_value}"
)
super().__init__(self.message)

Expand Down
30 changes: 14 additions & 16 deletions src/abbfreeathome/freeathome.py
Original file line number Diff line number Diff line change
@@ -1,32 +1,32 @@
"""ABB-Free@Home wrapper for interacting with the ABB-free@home API."""

from .api import FreeAtHomeApi
from .bin.function_id import FunctionID
from .bin.function import Function
from .bin.interface import Interface
from .devices.switch_actuator import Base, SwitchActuator


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

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

def __init__(
self, api: FreeAtHomeApi, interfaces: list[Interface] | None = None
) -> None:
"""Initialize the FreeAtHome class."""
self._api = api
self._interfaces = interfaces
self.api: FreeAtHomeApi = api
self._interfaces: list[Interface] = interfaces

async def get_config(self, refresh: bool = False) -> dict:
"""Get the Free@Home Configuration."""
if self._config is None or refresh:
self._config = await self._api.get_configuration()
self._config = await self.api.get_configuration()

return self._config

async def get_devices_by_function(self, function_id: FunctionID) -> list[dict]:
async def get_devices_by_function(self, function: Function) -> list[dict]:
"""Get the list of devices by function."""
_devices = []
for _device_key, _device in (await self.get_config()).get("devices").items():
Expand All @@ -39,7 +39,7 @@ async def get_devices_by_function(self, function_id: FunctionID) -> list[dict]:
for _channel_key, _channel in _device.get("channels", {}).items():
if (
_channel.get("functionID")
and int(_channel.get("functionID"), 16) == function_id.value
and int(_channel.get("functionID"), 16) == function.value
):
_channel_name = _channel.get("displayName")
if _channel_name == "Ⓐ" or _channel_name is None:
Expand Down Expand Up @@ -113,13 +113,11 @@ async def load_devices(self):

# SwitchActuator
await self._load_devices_by_function(
FunctionID.FID_SWITCH_ACTUATOR, SwitchActuator
Function.FID_SWITCH_ACTUATOR, SwitchActuator
)

async def _load_devices_by_function(
self, function_id: FunctionID, device_class: Base
):
_devices = await self.get_devices_by_function(function_id=function_id)
async def _load_devices_by_function(self, function: Function, device_class: Base):
_devices = await self.get_devices_by_function(function)
for _device in _devices:
self._devices[f"{_device.get('device_id')}/{_device.get('channel_id')}"] = (
device_class(
Expand All @@ -130,19 +128,19 @@ async def _load_devices_by_function(
inputs=_device.get("inputs"),
outputs=_device.get("outputs"),
parameters=_device.get("parameters"),
api=self._api,
api=self.api,
floor_name=_device.get("floor_name"),
room_name=_device.get("room_name"),
)
)

async def ws_close(self):
"""Close the websocker connection."""
await self._api.ws_close()
await self.api.ws_close()

async def ws_listen(self):
"""Listen on the websocket for updates to devices."""
await self._api.ws_listen(callback=self.update_device)
await self.api.ws_listen(callback=self.update_device)

async def update_device(self, data: dict):
"""Update device based on websocket data."""
Expand Down
23 changes: 12 additions & 11 deletions tests/test_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,8 +5,9 @@
import pytest

from abbfreeathome.api import FreeAtHomeApi
from abbfreeathome.bin.pairing import Pairing
from abbfreeathome.devices.base import Base
from abbfreeathome.exceptions import InvalidDeviceChannelPairingId
from abbfreeathome.exceptions import InvalidDeviceChannelPairing


@pytest.fixture
Expand Down Expand Up @@ -55,24 +56,24 @@ def test_initialization(base_instance):
assert base_instance.room_name == "Study"


def test_get_input_by_pairing_id(base_instance):
"""Test the get_input_pairing_id function."""
input_id, value = base_instance.get_input_by_pairing_id(1)
def test_get_input_by_pairing(base_instance):
"""Test the get_input_pairing function."""
input_id, value = base_instance.get_input_by_pairing(Pairing.AL_SWITCH_ON_OFF)
assert input_id == "idp0000"
assert value == "0"

with pytest.raises(InvalidDeviceChannelPairingId):
base_instance.get_input_by_pairing_id(99)
with pytest.raises(InvalidDeviceChannelPairing):
base_instance.get_input_by_pairing(Pairing.AL_HSV)


def test_get_output_by_pairing_id(base_instance):
"""Test the get_output_pairing_id function."""
output_id, value = base_instance.get_output_by_pairing_id(256)
def test_get_output_by_pairing(base_instance):
"""Test the get_output_pairing function."""
output_id, value = base_instance.get_output_by_pairing(Pairing.AL_INFO_ON_OFF)
assert output_id == "odp0000"
assert value == "0"

with pytest.raises(InvalidDeviceChannelPairingId):
base_instance.get_output_by_pairing_id(99)
with pytest.raises(InvalidDeviceChannelPairing):
base_instance.get_output_by_pairing(Pairing.AL_HSV)


def test_register_callback(base_instance):
Expand Down
8 changes: 4 additions & 4 deletions tests/test_exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
ForbiddenAuthException,
InvalidApiResponseException,
InvalidCredentialsException,
InvalidDeviceChannelPairingId,
InvalidDeviceChannelPairing,
InvalidHostException,
SetDatapointFailureException,
UserNotFoundException,
Expand Down Expand Up @@ -54,10 +54,10 @@ def test_invalid_api_response_exception():
assert str(excinfo.value) == "Invalid api response, status code: 404"


def test_invalid_device_channel_pairing_id():
def test_invalid_device_channel_pairing():
"""Test invalid device channel exception."""
with pytest.raises(InvalidDeviceChannelPairingId) as excinfo:
raise InvalidDeviceChannelPairingId("device1", "channel1", 123)
with pytest.raises(InvalidDeviceChannelPairing) as excinfo:
raise InvalidDeviceChannelPairing("device1", "channel1", 123)
assert str(excinfo.value) == (
"Could not find paring id for "
"device: device1; channel: channel1; pairing id: 123"
Expand Down
4 changes: 2 additions & 2 deletions tests/test_freeathome.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
import pytest

from abbfreeathome.api import FreeAtHomeApi
from abbfreeathome.bin.function_id import FunctionID
from abbfreeathome.bin.function import Function
from abbfreeathome.bin.interface import Interface
from abbfreeathome.devices.switch_actuator import SwitchActuator
from abbfreeathome.freeathome import FreeAtHome
Expand Down Expand Up @@ -194,7 +194,7 @@ async def test_get_config(freeathome, api_mock):
@pytest.mark.asyncio
async def test_get_devices_by_function(freeathome):
"""Test the get_devices_by_fuction function."""
devices = await freeathome.get_devices_by_function(FunctionID.FID_SWITCH_ACTUATOR)
devices = await freeathome.get_devices_by_function(Function.FID_SWITCH_ACTUATOR)
assert len(devices) == 2
assert devices[0]["device_name"] == "Study Area Rocker"
assert devices[0]["channel_name"] == "Study Area Light"
Expand Down
Loading