"""Controller holding and managing Afero IoT resources of type `fan`."""
from aioafero.device import AferoDevice, get_function_from_device
from aioafero.util import ordered_list_item_to_percentage
from aioafero.v1.models import features
from aioafero.v1.models.fan import Fan, FanPut
from aioafero.v1.models.resource import DeviceInformation, ResourceTypes
from .base import AferoBinarySensor, AferoSensor, BaseResourcesController
KNOWN_PRESETS = {"comfort-breeze"}
[docs]
class FanController(BaseResourcesController[Fan]):
"""Fan devices on ``bridge.fans``."""
ITEM_TYPE_ID = ResourceTypes.DEVICE
ITEM_TYPES = [ResourceTypes.FAN]
ITEM_CLS = Fan
ITEM_MAPPING = {
"on": "power",
"speed": "fan-speed",
"direction": "fan-reverse",
}
[docs]
async def turn_on(self, device_id: str) -> None:
"""Turn on the fan.
Args:
device_id: Device ID from this controller.
"""
await self.set_state(device_id, on=True)
[docs]
async def turn_off(self, device_id: str) -> None:
"""Turn off the fan.
Args:
device_id: Device ID from this controller.
"""
await self.set_state(device_id, on=False)
[docs]
async def set_speed(self, device_id: str, speed: int) -> None:
"""Set fan speed as a percentage.
Args:
device_id: Device ID from this controller.
speed: Speed percentage (``0`` turns the fan off).
"""
await self.set_state(device_id, on=True, speed=speed)
[docs]
async def set_direction(self, device_id: str, forward: bool) -> None:
"""Set fan rotation direction.
Args:
device_id: Device ID from this controller.
forward: ``True`` for forward, ``False`` for reverse. Ignored if the fan
is off (Hubspace API quirk).
"""
cur_item = self.get_device(device_id)
if not cur_item.is_on:
# Thanks Hubspace for this one! Additionally, turning it on and setting
# direction at the same time does not work as expected
self._logger.info("Fan is not running so direction will not be set")
await self.set_state(device_id, forward=forward)
[docs]
async def set_preset(self, device_id: str, preset: bool) -> None:
"""Enable or disable the comfort-breeze preset.
Args:
device_id: Device ID from this controller.
preset: ``True`` to enable the preset, ``False`` to disable.
"""
await self.set_state(device_id, on=True, preset=preset)
[docs]
async def initialize_elem(self, afero_device: AferoDevice) -> Fan:
"""Initialize the element.
:param afero_device: Afero Device that contains the updated states
:return: Newly initialized resource
"""
available: bool = False
on: features.OnFeature | None = None
speed: features.SpeedFeature | None = None
direction: features.DirectionFeature | None = None
preset: features.PresetFeature | None = None
sensors: dict[str, AferoSensor] = {}
binary_sensors: dict[str, AferoBinarySensor] = {}
for state in afero_device.states:
if state.functionClass == "power":
on = features.OnFeature(on=state.value == "on")
elif state.functionClass == "fan-speed":
speeds = get_function_from_device(
afero_device.functions, state.functionClass, state.functionInstance
)
tmp_speed = set()
for value in speeds["values"]:
if not value["name"].endswith("-000"):
tmp_speed.add(value["name"])
speeds = sorted(tmp_speed)
percentage = ordered_list_item_to_percentage(speeds, state.value)
speed = features.SpeedFeature(speed=percentage, speeds=speeds)
elif state.functionClass == "fan-reverse":
direction = features.DirectionFeature(forward=state.value == "forward")
elif (
state.functionClass == "toggle"
and state.functionInstance in KNOWN_PRESETS
):
# I have only seen fans with a single preset
preset = features.PresetFeature(
enabled=state.value == "enabled",
func_class=state.functionClass,
func_instance=state.functionInstance,
)
elif state.functionClass == "available":
available = state.value
self._items[afero_device.id] = Fan(
_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,
),
on=on,
speed=speed,
direction=direction,
preset=preset,
)
return self._items[afero_device.id]
[docs]
async def update_elem(self, afero_device: AferoDevice) -> set:
"""Update the Fan with the latest API data.
:param afero_device: Afero Device that contains the updated states
:return: States that have been modified
"""
updated_keys = set()
cur_item = self.get_device(afero_device.id)
for state in afero_device.states:
if state.functionClass == "power":
new_val = state.value == "on"
if cur_item.on.on != new_val:
cur_item.on.on = new_val
updated_keys.add("on")
elif state.functionClass == "fan-speed":
new_val = ordered_list_item_to_percentage(
cur_item.speed.speeds, state.value
)
if cur_item.speed.speed != new_val:
cur_item.speed.speed = new_val
updated_keys.add("speed")
elif state.functionClass == "fan-reverse":
new_val = state.value == "forward"
if cur_item.direction.forward != new_val:
cur_item.direction.forward = new_val
updated_keys.add("direction")
elif (
state.functionClass == "toggle"
and state.functionInstance in KNOWN_PRESETS
):
new_val = state.value == "enabled"
if cur_item.preset.enabled != new_val:
cur_item.preset.enabled = new_val
updated_keys.add("preset")
elif state.functionClass == "available":
if cur_item.available != state.value:
cur_item.available = state.value
updated_keys.add("available")
return updated_keys
[docs]
async def set_state(
self,
device_id: str,
on: bool | None = None,
speed: int | None = None,
forward: bool | None = None,
preset: bool | None = None,
) -> None:
"""Update fan state in the cloud.
Args:
device_id: Device ID from this controller.
on: Power state.
speed: Fan speed percentage (``0`` turns off).
forward: ``True`` for forward rotation, ``False`` for reverse.
preset: Comfort-breeze preset enabled state.
"""
update_obj = FanPut()
cur_item = self.get_device(device_id)
if on is not None:
update_obj.on = features.OnFeature(on=on)
if speed is not None and cur_item.speed is not None:
if speed == 0:
update_obj.on = features.OnFeature(on=False)
else:
update_obj.speed = features.SpeedFeature(
speed=speed, speeds=cur_item.speed.speeds
)
if preset is not None and cur_item.preset is not None:
update_obj.preset = features.PresetFeature(
enabled=preset,
func_class=cur_item.preset.func_class,
func_instance=cur_item.preset.func_instance,
)
if forward is not None and cur_item.direction is not None:
update_obj.direction = features.DirectionFeature(forward=forward)
await self.update(device_id, obj_in=update_obj)