"""Controller holding and managing Afero IoT resources of type `switch`."""
from dataclasses import replace
from aioafero import errors
from aioafero.device import AferoDevice
from aioafero.v1.models import features
from aioafero.v1.models.resource import DeviceInformation, ResourceTypes
from aioafero.v1.models.switch import Switch, SwitchPut
from .base import AferoBinarySensor, AferoSensor, BaseResourcesController
[docs]
class SwitchController(BaseResourcesController[Switch]):
"""Switch devices on ``bridge.switches``.
A switch can have one or more toggleable elements controlled by ``instance``.
"""
ITEM_TYPE_ID = ResourceTypes.DEVICE
ITEM_TYPES = [
ResourceTypes.SWITCH,
ResourceTypes.POWER_OUTLET,
ResourceTypes.LANDSCAPE_TRANSFORMER,
]
ITEM_CLS = Switch
# Sensors map functionClass -> Unit
ITEM_SENSORS: dict[str, str] = {"watts": "W", "output-voltage-switch": "V"}
# Binary sensors map key -> alerting value
ITEM_BINARY_SENSORS: dict[str, str] = {}
[docs]
async def turn_on(self, device_id: str, *, instance: str | None = None) -> None:
"""Turn on the switch.
Args:
device_id: Device ID from this controller.
instance: ``functionInstance`` when the device has multiple toggles.
"""
await self.set_state(device_id, on=True, instance=instance)
[docs]
async def turn_off(self, device_id: str, *, instance: str | None = None) -> None:
"""Turn off the switch.
Args:
device_id: Device ID from this controller.
instance: ``functionInstance`` when the device has multiple toggles.
"""
await self.set_state(device_id, on=False, instance=instance)
[docs]
async def initialize_elem(self, afero_device: AferoDevice) -> Switch:
"""Initialize the element.
:param afero_device: Afero Device that contains the updated states
:return: Newly initialized resource
"""
available: bool = False
on: dict[str | None, features.OnFeature] = {}
sensors: dict[str, AferoSensor] = {}
binary_sensors: dict[str, AferoBinarySensor] = {}
toggle_states = ["power", "toggle"]
for state in afero_device.states:
if state.functionClass in toggle_states:
on[state.functionInstance] = features.OnFeature(
on=state.value == "on",
function_class=state.functionClass,
function_instance=state.functionInstance,
)
elif state.functionClass == "available":
available = state.value
elif sensor := await self.initialize_sensor(state, afero_device.device_id):
# Currently sensors only have sensors, not binary sensors
sensors[sensor.id] = sensor
self._items[afero_device.id] = Switch(
_id=afero_device.id,
available=available,
sensors=sensors,
binary_sensors=binary_sensors,
split=afero_device.split,
device_information=DeviceInformation(
device_class=afero_device.device_class,
default_image=afero_device.default_image,
default_name=afero_device.default_name,
manufacturer=afero_device.manufacturerName,
model=afero_device.model,
name=afero_device.friendly_name,
parent_id=afero_device.device_id,
children=afero_device.children,
functions=afero_device.functions,
),
on=on,
)
return self._items[afero_device.id]
[docs]
async def update_elem(self, afero_device: AferoDevice) -> set:
"""Update the Switch with the latest API data.
:param afero_device: Afero Device that contains the updated states
:return: States that have been modified
"""
cur_item = self.get_device(afero_device.id)
updated_keys = set()
toggle_states = ["power", "toggle"]
for state in afero_device.states:
if state.functionClass in toggle_states:
new_val = state.value == "on"
if cur_item.on[state.functionInstance].on != new_val:
updated_keys.add("on")
cur_item.on[state.functionInstance].on = state.value == "on"
elif state.functionClass == "available":
if cur_item.available != state.value:
updated_keys.add("available")
cur_item.available = state.value
elif update_key := await self.update_sensor(state, cur_item):
updated_keys.add(update_key)
return updated_keys
[docs]
async def set_state(
self,
device_id: str,
*,
on: bool | None = None,
instance: str | None = None,
) -> None:
"""Update switch state in the cloud.
Args:
device_id: Device ID from this controller.
on: Power state for the selected instance.
instance: ``functionInstance`` when the device has multiple toggles.
"""
update_obj = SwitchPut()
try:
cur_item = self.get_device(device_id)
except errors.DeviceNotFound:
self._logger.info("Unable to find device %s", device_id)
return
if on is not None:
try:
update_obj.on = replace(cur_item.on[instance], on=on)
except KeyError:
self._logger.info("Unable to find instance %s", instance)
await self.update(device_id, obj_in=update_obj)