"""Controller holding and managing Afero IoT resources of type `valve`."""
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.valve import Valve, ValvePut
from .base import AferoBinarySensor, AferoSensor, BaseResourcesController
[docs]
class ValveController(BaseResourcesController[Valve]):
"""Valve / water-timer devices on ``bridge.valves``.
Valves can have multiple toggleable elements controlled by ``instance``.
"""
ITEM_TYPE_ID = ResourceTypes.DEVICE
ITEM_TYPES = [ResourceTypes.WATER_TIMER]
ITEM_CLS = Valve
[docs]
async def turn_on(self, device_id: str, *, instance: str | None = None) -> None:
"""Open the valve.
Args:
device_id: Device ID from this controller.
instance: ``functionInstance`` for multi-zone valves.
"""
await self.set_state(device_id, valve_open=True, instance=instance)
[docs]
async def turn_off(self, device_id: str, *, instance: str | None = None) -> None:
"""Close the valve.
Args:
device_id: Device ID from this controller.
instance: ``functionInstance`` for multi-zone valves.
"""
await self.set_state(device_id, valve_open=False, instance=instance)
[docs]
async def initialize_elem(self, afero_device: AferoDevice) -> Valve:
"""Initialize the element.
:param afero_device: Afero Device that contains the updated states
:return: Newly initialized resource
"""
self._logger.info("Initializing %s", afero_device.id)
available: bool = False
valve_open: dict[str | None, features.OpenFeature] = {}
sensors: dict[str, AferoSensor] = {}
binary_sensors: dict[str, AferoBinarySensor] = {}
for state in afero_device.states:
if state.functionClass in ["power", "toggle"]:
valve_open[state.functionInstance] = features.OpenFeature(
open=state.value == "on",
function_class=state.functionClass,
function_instance=state.functionInstance,
)
elif state.functionClass == "available":
available = state.value
self._items[afero_device.id] = Valve(
_id=afero_device.id,
available=available,
sensors=sensors,
binary_sensors=binary_sensors,
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,
),
open=valve_open,
)
return self._items[afero_device.id]
[docs]
async def update_elem(self, afero_device: AferoDevice) -> set:
"""Update the Valve 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()
for state in afero_device.states:
if state.functionClass in ["power", "toggle"]:
new_state = state.value == "on"
if cur_item.open[state.functionInstance].open != new_state:
updated_keys.add("open")
cur_item.open[state.functionInstance].open = new_state
elif state.functionClass == "available":
if cur_item.available != state.value:
updated_keys.add("available")
cur_item.available = state.value
return updated_keys
[docs]
async def set_state(
self,
device_id: str,
*,
valve_open: bool | None = None,
instance: str | None = None,
) -> None:
"""Update valve state in the cloud.
Args:
device_id: Device ID from this controller.
valve_open: ``True`` to open, ``False`` to close.
instance: ``functionInstance`` for multi-zone valves.
"""
update_obj = ValvePut()
try:
cur_item = self.get_device(device_id)
except errors.DeviceNotFound:
self._logger.info("Unable to find device %s", device_id)
return
if valve_open is not None:
try:
update_obj.open = replace(cur_item.open[instance], open=valve_open)
except KeyError:
self._logger.info("Unable to find instance %s", instance)
await self.update(device_id, obj_in=update_obj)