"""Controller holding and managing Hubspace resources of type `light`."""
from contextlib import suppress
import copy
import logging
from aioafero import device, errors
from aioafero.device import AferoDevice, AferoState
from aioafero.util import process_range
from aioafero.v1.models import features
from aioafero.v1.models.light import Light, LightPut
from aioafero.v1.models.resource import DeviceInformation, ResourceTypes
from .base import AferoBinarySensor, AferoSensor, BaseResourcesController, NumbersName
from .event import CallbackResponse
SPLIT_IDENTIFIER: str = "light"
logger = logging.getLogger(__name__)
[docs]
def process_names(values: list[dict]) -> set[str]:
"""Extract unique names from the elements."""
vals = set()
for val in values:
vals.add(val["name"])
return vals
[docs]
def generate_split_name(afero_device: AferoDevice, instance: str) -> str:
"""Generate the name for an instanced element."""
return f"{afero_device.id}-{SPLIT_IDENTIFIER}-{instance}"
[docs]
def get_split_instances(afero_dev: AferoDevice) -> list[tuple[str, ResourceTypes]]:
"""Determine available instances from the states."""
instances = set()
lights = []
toggles = []
for state in afero_dev.states:
# We do not want to add something that controls everything, but individual only
# We should skip None as its typically a single instance
if state.functionInstance in ["global", "primary", None]:
continue
if state.functionClass == "brightness":
lights.append(state.functionInstance)
elif state.functionClass == "toggle":
toggles.append(state.functionInstance)
# If there is only one instance, treat it as a light only with no splits
if len(lights) > 1:
for light_instance in lights:
instances.add((light_instance, ResourceTypes.LIGHT))
for toggle_instance in toggles:
if toggle_instance in [x[0] for x in instances]:
continue
instances.add((toggle_instance, ResourceTypes.SWITCH))
return sorted(instances)
[docs]
def state_belongs_to_light_instance(
afero_dev: AferoDevice, state: AferoState, instance: str
) -> bool:
"""Return whether a state belongs to a light zone instance."""
if state.functionClass == "available":
return True
if afero_dev.model == "LCN3002LM-01 WH":
if state.functionInstance == "primary":
return False
return (instance == "white" and state.functionInstance == instance) or (
instance != "white" and state.functionInstance != "white"
)
if state.functionInstance in (None, "global", "primary"):
return False
return state.functionInstance == instance
[docs]
def state_matches_instance(afero_device: AferoDevice, state: AferoState) -> bool:
"""Return whether a state belongs to a split light instance."""
if not afero_device.split_identifier:
return True
instance = afero_device.id.rsplit(f"-{afero_device.split_identifier}-", 1)[1]
return state_belongs_to_light_instance(afero_device, state, instance)
[docs]
def resolve_function_instance(afero_device: AferoDevice) -> str | None:
"""Return the functionInstance key for this light resource (split zone or single)."""
if afero_device.split_identifier:
return afero_device.id.rsplit(f"-{afero_device.split_identifier}-", 1)[1]
for state in afero_device.states:
if state.functionClass == "color-mode":
return state.functionInstance
return None
[docs]
def get_color_modes_for_device(afero_device: AferoDevice) -> list[str]:
"""Return supported color-mode names for this zone's color-mode function."""
instance = resolve_function_instance(afero_device)
for function in afero_device.functions:
if function.get("functionClass") != "color-mode":
continue
if function.get("functionInstance") == instance:
return [value["name"] for value in function.get("values", [])]
for function in afero_device.functions:
if function.get("functionClass") == "color-mode":
return [value["name"] for value in function.get("values", [])]
return []
[docs]
def get_valid_states(afero_dev: AferoDevice, instance: str) -> list:
"""Find states associated with the specific instance."""
return [
state
for state in afero_dev.states
if state_belongs_to_light_instance(afero_dev, state, instance)
]
[docs]
def light_callback(afero_device: AferoDevice) -> CallbackResponse:
"""Convert an AferoDevice into multiple devices."""
multi_devs: list[AferoDevice] = []
instances: list[tuple[str, ResourceTypes]] = []
remove_parent: bool = False
instances = get_split_instances(afero_device)
logger.debug("Light instances found: %s", instances)
light_instances = [x[0] for x in instances if x[1] == ResourceTypes.LIGHT]
children: list[str] = []
if afero_device.device_class == ResourceTypes.LIGHT.value:
for instance, resource_type in instances:
instance_name = instance or "primary"
cloned = copy.deepcopy(afero_device)
cloned.device_class = resource_type.value
cloned.id = generate_split_name(afero_device, instance)
cloned.split_identifier = SPLIT_IDENTIFIER
cloned.friendly_name = f"{afero_device.friendly_name} - {instance_name}"
cloned.states = get_valid_states(afero_device, instance)
cloned.children = []
multi_devs.append(cloned)
children.append(cloned.id)
if len(light_instances) > 1 and None not in light_instances:
remove_parent = True
cloned = copy.deepcopy(afero_device)
valid_states = [
state
for state in afero_device.states
if state.functionClass
in [
"available",
"wifi-ssid",
"wifi-rssi",
"wifi-steady-state",
"wifi-setup-state",
"wifi-mac-address",
"ble-mac-address",
]
]
cloned.states = valid_states
cloned.children = children
cloned.device_class = "parent-device"
multi_devs.append(cloned)
return CallbackResponse(
split_devices=multi_devs,
remove_original=remove_parent,
)
[docs]
class LightController(BaseResourcesController[Light]):
"""Light devices on ``bridge.lights`` (including split zones)."""
ITEM_TYPE_ID = ResourceTypes.DEVICE
ITEM_TYPES = [ResourceTypes.LIGHT]
ITEM_CLS = Light
ITEM_MAPPING = {
"color": "color-rgb",
"color_mode": "color-mode",
"color_temperature": "color-temperature",
"dimming": "brightness",
"effect": "color-sequence",
}
ITEM_NUMBERS: dict[tuple[str, str | None], NumbersName] = {
("speed", "color-sequence"): NumbersName(unit="speed"),
}
# Split Lights from the primary payload
DEVICE_SPLIT_CALLBACKS: dict[str, callable] = {
ResourceTypes.LIGHT.value: light_callback
}
[docs]
async def turn_on(self, device_id: str) -> None:
"""Turn on the light.
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 light.
Args:
device_id: Device ID from this controller.
"""
await self.set_state(device_id, on=False)
[docs]
async def set_color_temperature(self, device_id: str, temperature: int) -> None:
"""Set color temperature or white mode when CCT is unavailable.
Args:
device_id: Device ID from this controller.
temperature: Color temperature in kelvin.
"""
try:
cur_item = self.get_device(device_id)
except errors.DeviceNotFound:
self._logger.info("Unable to find device %s", device_id)
return
if cur_item.color_temperature is not None:
await self.set_state(
device_id, on=True, temperature=temperature, color_mode="white"
)
elif cur_item.supports_color_white:
self._logger.info(
"Device %s has no color-temperature function; ignoring %d K",
device_id,
temperature,
)
await self.set_white(device_id, on=True)
else:
self._logger.info(
"Device %s does not support color temperature or white mode", device_id
)
[docs]
async def set_white(
self,
device_id: str,
*,
on: bool | None = True,
brightness: int | None = None,
) -> None:
"""Set white mode without a color-temperature function.
Args:
device_id: Device ID from this controller.
on: Power state (defaults to ``True``).
brightness: Brightness percentage when ``on`` is ``True``.
"""
await self.set_state(
device_id, on=on, color_mode="white", brightness=brightness
)
[docs]
async def set_brightness(self, device_id: str, brightness: int) -> None:
"""Set brightness, turning the light on if needed.
Args:
device_id: Device ID from this controller.
brightness: Brightness percentage.
"""
await self.set_state(device_id, on=True, brightness=brightness)
[docs]
async def set_rgb(self, device_id: str, red: int, green: int, blue: int) -> None:
"""Set RGB color, turning the light on if needed.
Args:
device_id: Device ID from this controller.
red: Red channel ``0``–``255``.
green: Green channel ``0``–``255``.
blue: Blue channel ``0``–``255``.
"""
await self.set_state(
device_id, on=True, color=(red, green, blue), color_mode="color"
)
[docs]
async def set_effect(self, device_id: str, effect: str) -> None:
"""Set a color sequence effect, turning the light on if needed.
Args:
device_id: Device ID from this controller.
effect: Effect name from the model's ``effect.effects`` list.
"""
await self.set_state(device_id, on=True, effect=effect, color_mode="sequence")
[docs]
async def initialize_elem(self, afero_device: AferoDevice) -> Light:
"""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
color_temp: features.ColorTemperatureFeature | None = None
color: features.ColorFeature | None = None
color_mode: features.ColorModeFeature | None = None
dimming: features.DimmingFeature | None = None
effect: features.EffectFeature | None = None
sensors: dict[str, AferoSensor] = {}
binary_sensors: dict[str, AferoBinarySensor] = {}
numbers: dict[tuple[str, str], features.NumbersFeature] = {}
for state in afero_device.states:
func_def = device.get_function_from_device(
afero_device.functions, state.functionClass, state.functionInstance
)
if state.functionClass == "power" or (
afero_device.split_identifier and state.functionClass == "toggle"
):
on = features.OnFeature(
on=state.value == "on",
func_class=state.functionClass,
func_instance=state.functionInstance,
)
elif state.functionClass == "color-temperature":
if len(func_def["values"]) > 1:
avail_temps = process_color_temps(func_def["values"])
else:
avail_temps = process_range(func_def["values"][0])
prefix = "K" if func_def.get("type", None) != "numeric" else ""
current_temp = state.value
if isinstance(current_temp, str) and current_temp.endswith("K"):
current_temp = current_temp[:-1]
color_temp = features.ColorTemperatureFeature(
temperature=int(current_temp), supported=avail_temps, prefix=prefix
)
elif state.functionClass == "brightness":
temp_bright = process_range(func_def["values"][0])
dimming = features.DimmingFeature(
brightness=int(state.value), supported=temp_bright
)
elif state.functionClass == "color-sequence":
current_effect = state.value
effects = process_effects(afero_device.functions)
effect = features.EffectFeature(effect=current_effect, effects=effects)
elif state.functionClass == "color-rgb":
color = features.ColorFeature(
red=state.value["color-rgb"].get("r", 0),
green=state.value["color-rgb"].get("g", 0),
blue=state.value["color-rgb"].get("b", 0),
)
elif state.functionClass == "color-mode":
color_mode = features.ColorModeFeature(state.value)
elif state.functionClass == "available":
available = state.value
if number := await self.initialize_number(func_def, state):
numbers[number[0]] = number[1]
supported_color_modes = get_color_modes_for_device(afero_device)
self._items[afero_device.id] = Light(
_id=afero_device.id,
available=available,
split_identifier=afero_device.split_identifier,
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,
dimming=dimming,
color_mode=color_mode,
color_temperature=color_temp,
color=color,
color_modes=supported_color_modes,
effect=effect,
numbers=numbers,
)
return self._items[afero_device.id]
[docs]
async def update_elem(self, afero_device: AferoDevice) -> set:
"""Update the Light 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()
color_seq_states: dict[str, AferoState] = {}
for state in afero_device.states:
# Split clones are pre-filtered in light_callback; this guards other paths.
if not state_matches_instance(afero_device, state):
continue
if state.functionClass == "power" or (
afero_device.split_identifier and state.functionClass == "toggle"
):
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 == "color-temperature":
if cur_item.color_temperature is None:
continue
current_temp = state.value
if isinstance(current_temp, str) and current_temp.endswith("K"):
current_temp = current_temp[:-1]
new_val = int(current_temp)
if cur_item.color_temperature.temperature != new_val:
cur_item.color_temperature.temperature = new_val
updated_keys.add("color_temperature")
elif state.functionClass == "brightness":
new_val = int(state.value)
if cur_item.dimming.brightness != new_val:
cur_item.dimming.brightness = int(state.value)
updated_keys.add("dimming")
elif state.functionClass == "color-sequence":
color_seq_states[state.functionInstance] = state
elif state.functionClass == "color-rgb":
color_red = state.value["color-rgb"].get("r", 0)
color_green = state.value["color-rgb"].get("g", 0)
color_blue = state.value["color-rgb"].get("b", 0)
if (
cur_item.color.red != color_red
or cur_item.color.green != color_green
or cur_item.color.blue != color_blue
):
cur_item.color.red = color_red
cur_item.color.green = color_green
cur_item.color.blue = color_blue
updated_keys.add("color")
elif state.functionClass == "color-mode":
if cur_item.color_mode.mode != state.value:
cur_item.color_mode.mode = state.value
updated_keys.add("color_mode")
elif state.functionClass == "available":
if cur_item.available != state.value:
cur_item.available = state.value
updated_keys.add("available")
elif (update_key := await self.update_number(state, cur_item)) or (
update_key := await self.update_select(state, cur_item)
):
updated_keys.add(update_key)
# Several states hold the effect, but its always derived from the preset functionInstance
return updated_keys.union(
await self.update_elem_color(cur_item, color_seq_states)
)
[docs]
async def update_elem_color(self, cur_item: Light, color_seq_states: dict) -> set:
"""Perform the update for effects."""
updated_keys = set()
if color_seq_states and "preset" in color_seq_states:
preset_val = color_seq_states["preset"].value
if cur_item.effect.is_preset(preset_val):
if cur_item.effect.effect != preset_val:
cur_item.effect.effect = preset_val
updated_keys.add("effect")
else:
new_val = color_seq_states[color_seq_states["preset"].value].value
if cur_item.effect.effect != new_val:
cur_item.effect.effect = color_seq_states[
color_seq_states["preset"].value
].value
updated_keys.add("effect")
return updated_keys
[docs]
async def set_state(
self,
device_id: str,
on: bool | None = None,
temperature: int | None = None,
brightness: int | None = None,
color_mode: str | None = None,
color: tuple[int, int, int] | None = None,
effect: str | None = None,
force_white_mode: int | None = None,
numbers: dict[tuple[str, str], float] | None = None,
) -> None:
"""Update light state in the cloud.
Args:
device_id: Device ID from this controller.
on: Power state.
temperature: Color temperature in kelvin.
brightness: Brightness percentage.
color_mode: API color mode (``white``, ``color``, ``sequence``, etc.).
color: RGB tuple ``(red, green, blue)``.
effect: Named color sequence effect.
force_white_mode: Brightness to apply after forcing white mode.
numbers: Number features keyed by ``(functionClass, functionInstance)``.
"""
update_obj = LightPut()
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:
update_obj.on = features.OnFeature(
on=on,
func_class=cur_item.on.func_class,
func_instance=cur_item.on.func_instance,
)
send_duplicate_states = False
if force_white_mode is not None:
send_duplicate_states = True
update_obj.color_mode = features.ColorModeFeature(mode="white")
update_obj.dimming = features.DimmingFeature(
brightness=force_white_mode, supported=cur_item.dimming.supported
)
else:
if numbers:
for key, val in numbers.items():
if key not in cur_item.numbers:
continue
update_obj.numbers[key] = features.NumbersFeature(
value=val,
min=cur_item.numbers[key].min,
max=cur_item.numbers[key].max,
step=cur_item.numbers[key].step,
name=cur_item.numbers[key].name,
unit=cur_item.numbers[key].unit,
)
if temperature is not None and cur_item.color_temperature is not None:
adjusted_temp = min(
cur_item.color_temperature.supported,
key=lambda x: abs(x - temperature),
)
update_obj.color_temperature = features.ColorTemperatureFeature(
temperature=adjusted_temp,
supported=cur_item.color_temperature.supported,
prefix=cur_item.color_temperature.prefix,
)
if color_mode is None:
color_mode = "white"
elif temperature is not None and cur_item.supports_color_white:
if color_mode is None:
color_mode = "white"
if brightness is not None and cur_item.dimming is not None:
update_obj.dimming = features.DimmingFeature(
brightness=brightness, supported=cur_item.dimming.supported
)
if color is not None and cur_item.color is not None:
update_obj.color = features.ColorFeature(
red=color[0], green=color[1], blue=color[2]
)
if color_mode is not None and cur_item.color_mode is not None:
update_obj.color_mode = features.ColorModeFeature(mode=color_mode)
if (
color_mode == "white"
and cur_item.color_temperature is None
and cur_item.supports_color_white
and cur_item.color_mode.mode != "white"
):
send_duplicate_states = True
if effect is not None and cur_item.effect is not None:
update_obj.effect = features.EffectFeature(
effect=effect, effects=cur_item.effect.effects
)
await self.update(
device_id, obj_in=update_obj, send_duplicate_states=send_duplicate_states
)
[docs]
def process_color_temps(color_temps: dict) -> list[int]:
"""Determine the supported color temps.
:param color_temps: Result from functions["values"]
"""
supported_temps = []
for temp in color_temps:
color_temp = temp["name"]
if isinstance(color_temp, str) and color_temp.endswith("K"):
color_temp = color_temp[0:-1]
supported_temps.append(int(color_temp))
return sorted(supported_temps)
[docs]
def process_effects(functions: list[dict]) -> dict[str, set]:
"""Determine the supported effects."""
supported_effects = {}
for function in functions:
if function["functionClass"] == "color-sequence":
supported_effects[function["functionInstance"]] = set(
process_names(function["values"])
)
# custom shouldn't be a value in preset
with suppress(KeyError):
supported_effects["preset"].remove("custom")
return supported_effects