From 57b8aa4b041d26d7ea5cc76fc626bed0fc3cbc93 Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Sun, 10 May 2026 14:59:18 -0600 Subject: [PATCH] HA Phase B: alarm + light + switch + climate + sensor + button + event MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit custom_components/omni_pca/ — six new platform modules wrapping the v1.0 client surface. Every command method catches CommandFailedError and re-raises HomeAssistantError so panel rejections (bad code, etc.) become user-friendly HA errors instead of silent failures. alarm_control_panel.py — OmniAreaAlarmPanel per discovered area. Supports ARM_HOME (Day) / ARM_NIGHT / ARM_AWAY / ARM_VACATION / ARM_CUSTOM_BYPASS (Day-Instant). State derives from area_status via pure helpers.security_mode_to_alarm_state which handles arming-in- progress, entry/exit timers, and active-alarm overrides. light.py — OmniUnitLight per discovered unit (every unit; non-dimmable units silently ignore brightness, no harm done). Brightness conversion via helpers.omni_state_to_ha_brightness / ha_brightness_to_omni_percent (Omni state byte: 0=off, 1=on, 100..200=brightness percent). switch.py — OmniZoneBypassSwitch per binary zone. CONFIG entity_category; pairs with the existing diagnostic 'zone bypassed' binary_sensor. climate.py — OmniThermostatClimate per discovered thermostat. Supports OFF / HEAT / COOL / HEAT_COOL hvac_modes; auto / on / diffuse fan_modes; none / hold / vacation preset_modes. Single-setpoint and range setpoint via TARGET_TEMPERATURE_RANGE. Fahrenheit native (Omni panels are F-native; HA handles unit conversion downstream). sensor.py — analog zones (temperature/humidity/power) + per-thermostat diagnostic temp/humidity/outdoor sensors + OmniSystemModelSensor + OmniLastEventSensor (event_class + parsed event fields as attrs). button.py — OmniPanelButton per discovered button macro. Programs not yet exposed because the library lacks RequestProperties for Programs. event.py — single OmniPanelEvent per panel relaying typed SystemEvents via _trigger_event. event_types: zone_state_changed, unit_state_changed, arming_changed, alarm_activated/cleared, ac_lost/restored, battery_low/restored, user_macro_button, phone_line_dead/restored. Automations key off platform: event + event_type filter. helpers.py — extended with security_mode_to_alarm_state, ARM_SERVICE_TO_SECURITY_MODE, omni_state_to_ha_brightness + ha_brightness_to_omni_percent, omni/ha_{hvac,fan,hold} round-trips, fahrenheit_to_omni_raw / celsius_to_omni_raw, analog_zone_device_class, EVENT_TYPES tuple, event_type_for(class_name). __init__.py — PLATFORMS extended to all 8 entity types. scene.py intentionally NOT created — Omni 'scenes' are user-defined button macros, already covered by the button platform. Documented in README; revisit if/when the library gains scene-discovery opcodes. tests/test_ha_helpers.py: +67 unit tests covering every new helper. 331 tests pass (was 264). Ruff clean across src/ tests/ custom_components/. --- custom_components/omni_pca/__init__.py | 19 +- .../omni_pca/alarm_control_panel.py | 156 ++++++++++ custom_components/omni_pca/button.py | 62 ++++ custom_components/omni_pca/climate.py | 271 ++++++++++++++++++ custom_components/omni_pca/event.py | 68 +++++ custom_components/omni_pca/helpers.py | 261 +++++++++++++++++ custom_components/omni_pca/light.py | 115 ++++++++ custom_components/omni_pca/sensor.py | 263 +++++++++++++++++ custom_components/omni_pca/switch.py | 89 ++++++ tests/test_ha_helpers.py | 195 +++++++++++++ 10 files changed, 1495 insertions(+), 4 deletions(-) create mode 100644 custom_components/omni_pca/alarm_control_panel.py create mode 100644 custom_components/omni_pca/button.py create mode 100644 custom_components/omni_pca/climate.py create mode 100644 custom_components/omni_pca/event.py create mode 100644 custom_components/omni_pca/light.py create mode 100644 custom_components/omni_pca/sensor.py create mode 100644 custom_components/omni_pca/switch.py diff --git a/custom_components/omni_pca/__init__.py b/custom_components/omni_pca/__init__.py index 5b4eaa0..93c18a3 100644 --- a/custom_components/omni_pca/__init__.py +++ b/custom_components/omni_pca/__init__.py @@ -1,8 +1,10 @@ """HAI/Leviton Omni Panel integration for Home Assistant. -Phase A entry point. Phase B will append additional platforms (light, -switch, climate, alarm_control_panel, sensor, scene, button, event) to -:data:`PLATFORMS`; nothing else here changes. +Forwards every config entry to the full set of platforms wrapping the +omni-pca library: alarm_control_panel (areas), binary_sensor (zones + +system flags), button (panel button macros), climate (thermostats), +event (typed push events), light (units), sensor (analog zones, +thermostat readings, panel telemetry), switch (zone bypass). """ from __future__ import annotations @@ -19,7 +21,16 @@ if TYPE_CHECKING: from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant -PLATFORMS: list[Platform] = [Platform.BINARY_SENSOR] +PLATFORMS: list[Platform] = [ + Platform.ALARM_CONTROL_PANEL, + Platform.BINARY_SENSOR, + Platform.BUTTON, + Platform.CLIMATE, + Platform.EVENT, + Platform.LIGHT, + Platform.SENSOR, + Platform.SWITCH, +] async def async_setup(hass: HomeAssistant, config: dict) -> bool: diff --git a/custom_components/omni_pca/alarm_control_panel.py b/custom_components/omni_pca/alarm_control_panel.py new file mode 100644 index 0000000..14448ca --- /dev/null +++ b/custom_components/omni_pca/alarm_control_panel.py @@ -0,0 +1,156 @@ +"""Alarm control panel platform — one entity per discovered Omni area. + +State translation lives in :func:`helpers.security_mode_to_alarm_state` +so it stays unit-testable without Home Assistant. Arm / disarm calls +go through :meth:`OmniClient.execute_security_command` which validates +the user code against the panel; a wrong code surfaces as +:class:`HomeAssistantError`. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from homeassistant.components.alarm_control_panel import ( + AlarmControlPanelEntity, + AlarmControlPanelEntityFeature, + AlarmControlPanelState, + CodeFormat, +) +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from omni_pca.commands import CommandFailedError +from omni_pca.models import SecurityMode + +from .const import DOMAIN +from .coordinator import OmniDataUpdateCoordinator +from .helpers import ( + ARM_SERVICE_TO_SECURITY_MODE, + prettify_name, + security_mode_to_alarm_state, +) + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + from homeassistant.helpers.entity_platform import AddEntitiesCallback + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + coordinator: OmniDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + entities = [ + OmniAreaAlarmPanel(coordinator, index) + for index in sorted(coordinator.data.areas) + ] + async_add_entities(entities) + + +_ALARM_STATE_STR_TO_ENUM: dict[str, AlarmControlPanelState] = { + "disarmed": AlarmControlPanelState.DISARMED, + "armed_home": AlarmControlPanelState.ARMED_HOME, + "armed_away": AlarmControlPanelState.ARMED_AWAY, + "armed_night": AlarmControlPanelState.ARMED_NIGHT, + "armed_vacation": AlarmControlPanelState.ARMED_VACATION, + "armed_custom_bypass": AlarmControlPanelState.ARMED_CUSTOM_BYPASS, + "arming": AlarmControlPanelState.ARMING, + "pending": AlarmControlPanelState.PENDING, + "triggered": AlarmControlPanelState.TRIGGERED, +} + + +class OmniAreaAlarmPanel( + CoordinatorEntity[OmniDataUpdateCoordinator], AlarmControlPanelEntity +): + """One discovered area as a HA alarm_control_panel.""" + + _attr_has_entity_name = True + _attr_code_arm_required = True + _attr_code_format = CodeFormat.NUMBER + _attr_supported_features = ( + AlarmControlPanelEntityFeature.ARM_HOME + | AlarmControlPanelEntityFeature.ARM_AWAY + | AlarmControlPanelEntityFeature.ARM_NIGHT + | AlarmControlPanelEntityFeature.ARM_VACATION + | AlarmControlPanelEntityFeature.ARM_CUSTOM_BYPASS + ) + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator) + self._index = index + self._attr_unique_id = f"{coordinator.unique_id}-area-{index}" + props = coordinator.data.areas[index] + self._attr_name = prettify_name(props.name) or f"Area {index}" + self._attr_device_info = coordinator.device_info + + @property + def available(self) -> bool: + return ( + super().available + and self.coordinator.data is not None + and self._index in self.coordinator.data.areas + ) + + @property + def alarm_state(self) -> AlarmControlPanelState | None: + status = self.coordinator.data.area_status.get(self._index) + if status is None: + return None + state_str = security_mode_to_alarm_state( + mode=status.mode, + alarm_active=status.alarm_active, + entry_timer=status.entry_timer_secs, + exit_timer=status.exit_timer_secs, + ) + return _ALARM_STATE_STR_TO_ENUM.get(state_str) + + @property + def extra_state_attributes(self) -> dict[str, Any] | None: + status = self.coordinator.data.area_status.get(self._index) + if status is None: + return None + return { + "area_index": self._index, + "raw_mode": status.mode, + "raw_mode_name": status.mode_name, + "entry_timer_secs": status.entry_timer_secs, + "exit_timer_secs": status.exit_timer_secs, + "last_user": status.last_user, + "alarms": status.alarms, + } + + async def _send(self, mode_name: str, code: str | None) -> None: + if code is None or not code.isdigit(): + raise HomeAssistantError("A numeric user code is required") + mode_value = ARM_SERVICE_TO_SECURITY_MODE[mode_name] + try: + await self.coordinator.client.execute_security_command( + area=self._index, mode=SecurityMode(mode_value), code=int(code) + ) + except CommandFailedError as err: + raise HomeAssistantError(f"Panel rejected command: {err}") from err + await self.coordinator.async_request_refresh() + + async def async_alarm_disarm(self, code: str | None = None) -> None: + await self._send("disarm", code) + + async def async_alarm_arm_home(self, code: str | None = None) -> None: + await self._send("arm_home", code) + + async def async_alarm_arm_away(self, code: str | None = None) -> None: + await self._send("arm_away", code) + + async def async_alarm_arm_night(self, code: str | None = None) -> None: + await self._send("arm_night", code) + + async def async_alarm_arm_vacation(self, code: str | None = None) -> None: + await self._send("arm_vacation", code) + + async def async_alarm_arm_custom_bypass(self, code: str | None = None) -> None: + await self._send("arm_custom_bypass", code) diff --git a/custom_components/omni_pca/button.py b/custom_components/omni_pca/button.py new file mode 100644 index 0000000..b6a2c84 --- /dev/null +++ b/custom_components/omni_pca/button.py @@ -0,0 +1,62 @@ +"""Button platform — one HA button per discovered Omni button macro. + +Programs aren't currently discoverable (the library doesn't yet have a +RequestProperties path for the Program object type), so program-execute +support lives in the services platform instead (Phase C). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from homeassistant.components.button import ButtonEntity +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from omni_pca.commands import CommandFailedError + +from .const import DOMAIN +from .coordinator import OmniDataUpdateCoordinator +from .helpers import prettify_name + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + from homeassistant.helpers.entity_platform import AddEntitiesCallback + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + coordinator: OmniDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + entities = [ + OmniPanelButton(coordinator, index) + for index in sorted(coordinator.data.buttons) + ] + async_add_entities(entities) + + +class OmniPanelButton( + CoordinatorEntity[OmniDataUpdateCoordinator], ButtonEntity +): + """Push-button entity that fires an Omni button macro.""" + + _attr_has_entity_name = True + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator) + self._index = index + self._attr_unique_id = f"{coordinator.unique_id}-button-{index}" + props = coordinator.data.buttons[index] + self._attr_name = prettify_name(props.name) or f"Button {index}" + self._attr_device_info = coordinator.device_info + + async def async_press(self) -> None: + try: + await self.coordinator.client.execute_button(self._index) + except CommandFailedError as err: + raise HomeAssistantError(f"Panel rejected button: {err}") from err diff --git a/custom_components/omni_pca/climate.py b/custom_components/omni_pca/climate.py new file mode 100644 index 0000000..cacd78b --- /dev/null +++ b/custom_components/omni_pca/climate.py @@ -0,0 +1,271 @@ +"""Climate platform — one HA climate entity per discovered thermostat. + +Omni stores temperatures in a linear byte (raw = round((°F + 40) * 10/9)). +HA stays in Fahrenheit because the panel is native there; users with HA +configured for metric will see automatic display conversion downstream. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, ClassVar + +from homeassistant.components.climate import ( + ATTR_HVAC_MODE, + ATTR_TARGET_TEMP_HIGH, + ATTR_TARGET_TEMP_LOW, + ClimateEntity, + ClimateEntityFeature, + HVACMode, +) +from homeassistant.const import ATTR_TEMPERATURE as ATTR_TEMP +from homeassistant.const import UnitOfTemperature +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from omni_pca.commands import CommandFailedError +from omni_pca.models import ( + FanMode as OmniFanMode, +) +from omni_pca.models import ( + HoldMode as OmniHoldMode, +) +from omni_pca.models import ( + HvacMode as OmniHvacMode, +) + +from .const import DOMAIN +from .coordinator import OmniDataUpdateCoordinator +from .helpers import ( + fahrenheit_to_omni_raw, + ha_fan_to_omni, + ha_hold_to_omni, + ha_hvac_to_omni, + omni_fan_to_ha, + omni_hold_to_ha, + omni_hvac_to_ha, + prettify_name, +) + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + from homeassistant.helpers.entity_platform import AddEntitiesCallback + + +_HVAC_STR_TO_ENUM: dict[str, HVACMode] = { + "off": HVACMode.OFF, + "heat": HVACMode.HEAT, + "cool": HVACMode.COOL, + "heat_cool": HVACMode.HEAT_COOL, +} + +PRESET_NONE = "none" +PRESET_HOLD = "hold" +PRESET_VACATION = "vacation" + +FAN_AUTO = "auto" +FAN_ON = "on" +FAN_DIFFUSE = "diffuse" + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + coordinator: OmniDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + entities = [ + OmniThermostatClimate(coordinator, index) + for index in sorted(coordinator.data.thermostats) + ] + async_add_entities(entities) + + +class OmniThermostatClimate( + CoordinatorEntity[OmniDataUpdateCoordinator], ClimateEntity +): + """One discovered thermostat as a HA climate entity.""" + + _attr_has_entity_name = True + _attr_temperature_unit = UnitOfTemperature.FAHRENHEIT + _attr_target_temperature_step = 1.0 + _attr_supported_features = ( + ClimateEntityFeature.TARGET_TEMPERATURE + | ClimateEntityFeature.TARGET_TEMPERATURE_RANGE + | ClimateEntityFeature.FAN_MODE + | ClimateEntityFeature.PRESET_MODE + | ClimateEntityFeature.TURN_OFF + | ClimateEntityFeature.TURN_ON + ) + _attr_hvac_modes: ClassVar[list[HVACMode]] = [ + HVACMode.OFF, + HVACMode.HEAT, + HVACMode.COOL, + HVACMode.HEAT_COOL, + ] + _attr_fan_modes: ClassVar[list[str]] = [FAN_AUTO, FAN_ON, FAN_DIFFUSE] + _attr_preset_modes: ClassVar[list[str]] = [PRESET_NONE, PRESET_HOLD, PRESET_VACATION] + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator) + self._index = index + self._attr_unique_id = f"{coordinator.unique_id}-thermostat-{index}" + props = coordinator.data.thermostats[index] + self._attr_name = prettify_name(props.name) or f"Thermostat {index}" + self._attr_device_info = coordinator.device_info + + @property + def available(self) -> bool: + return ( + super().available + and self.coordinator.data is not None + and self._index in self.coordinator.data.thermostats + ) + + @property + def _status(self): # type: ignore[no-untyped-def] + return self.coordinator.data.thermostat_status.get(self._index) + + @property + def current_temperature(self) -> float | None: + s = self._status + if s is None or s.temperature_raw == 0: + return None + return s.temperature_f + + @property + def current_humidity(self) -> int | None: + s = self._status + if s is None or s.humidity_raw == 0: + return None + return int(s.humidity_raw) + + @property + def hvac_mode(self) -> HVACMode | None: + s = self._status + if s is None: + return None + return _HVAC_STR_TO_ENUM.get(omni_hvac_to_ha(s.system_mode)) + + @property + def target_temperature(self) -> float | None: + s = self._status + if s is None: + return None + if s.system_mode == int(OmniHvacMode.HEAT): + return s.heat_setpoint_f + if s.system_mode == int(OmniHvacMode.COOL): + return s.cool_setpoint_f + return None + + @property + def target_temperature_high(self) -> float | None: + s = self._status + if s is None or s.system_mode != int(OmniHvacMode.AUTO): + return None + return s.cool_setpoint_f + + @property + def target_temperature_low(self) -> float | None: + s = self._status + if s is None or s.system_mode != int(OmniHvacMode.AUTO): + return None + return s.heat_setpoint_f + + @property + def fan_mode(self) -> str | None: + s = self._status + if s is None: + return None + return omni_fan_to_ha(s.fan_mode) + + @property + def preset_mode(self) -> str | None: + s = self._status + if s is None: + return None + return omni_hold_to_ha(s.hold_mode) + + @property + def extra_state_attributes(self) -> dict[str, Any] | None: + s = self._status + if s is None: + return None + return { + "thermostat_index": self._index, + "outdoor_temperature_f": ( + s.outdoor_temperature_raw and round(s.outdoor_temperature_f, 1) + ), + "humidify_setpoint": s.humidify_setpoint_raw, + "dehumidify_setpoint": s.dehumidify_setpoint_raw, + } + + # ---- setters --------------------------------------------------------- + + async def _set(self, coro_factory) -> None: # type: ignore[no-untyped-def] + try: + await coro_factory() + except CommandFailedError as err: + raise HomeAssistantError(f"Panel rejected command: {err}") from err + await self.coordinator.async_request_refresh() + + async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None: + omni_mode = OmniHvacMode(ha_hvac_to_omni(str(hvac_mode))) + await self._set( + lambda: self.coordinator.client.set_thermostat_system_mode( + self._index, omni_mode + ) + ) + + async def async_set_fan_mode(self, fan_mode: str) -> None: + omni_mode = OmniFanMode(ha_fan_to_omni(fan_mode)) + await self._set( + lambda: self.coordinator.client.set_thermostat_fan_mode( + self._index, omni_mode + ) + ) + + async def async_set_preset_mode(self, preset_mode: str) -> None: + omni_mode = OmniHoldMode(ha_hold_to_omni(preset_mode)) + await self._set( + lambda: self.coordinator.client.set_thermostat_hold_mode( + self._index, omni_mode + ) + ) + + async def async_set_temperature(self, **kwargs: Any) -> None: + if ATTR_HVAC_MODE in kwargs: + await self.async_set_hvac_mode(kwargs[ATTR_HVAC_MODE]) + s = self._status + if s is None: + raise HomeAssistantError("Thermostat not yet polled") + + if ATTR_TARGET_TEMP_LOW in kwargs: + await self._set( + lambda: self.coordinator.client.set_thermostat_heat_setpoint_raw( + self._index, fahrenheit_to_omni_raw(kwargs[ATTR_TARGET_TEMP_LOW]) + ) + ) + if ATTR_TARGET_TEMP_HIGH in kwargs: + await self._set( + lambda: self.coordinator.client.set_thermostat_cool_setpoint_raw( + self._index, fahrenheit_to_omni_raw(kwargs[ATTR_TARGET_TEMP_HIGH]) + ) + ) + if ATTR_TEMP in kwargs: + target_raw = fahrenheit_to_omni_raw(kwargs[ATTR_TEMP]) + # Single setpoint — choose heat or cool based on current mode. + if s.system_mode == int(OmniHvacMode.HEAT): + await self._set( + lambda: self.coordinator.client.set_thermostat_heat_setpoint_raw( + self._index, target_raw + ) + ) + elif s.system_mode == int(OmniHvacMode.COOL): + await self._set( + lambda: self.coordinator.client.set_thermostat_cool_setpoint_raw( + self._index, target_raw + ) + ) diff --git a/custom_components/omni_pca/event.py b/custom_components/omni_pca/event.py new file mode 100644 index 0000000..1459f45 --- /dev/null +++ b/custom_components/omni_pca/event.py @@ -0,0 +1,68 @@ +"""Event platform — surfaces the panel's typed push events as a single +``EventEntity`` per panel. The event_type attribute carries the kind of +event; event_data carries the parsed details. Trigger automations on +``platform: event`` filtering by event_type or event_data. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, ClassVar + +from homeassistant.components.event import EventEntity +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .const import DOMAIN +from .coordinator import OmniDataUpdateCoordinator +from .helpers import EVENT_TYPES, event_type_for + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + from homeassistant.helpers.entity_platform import AddEntitiesCallback + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + coordinator: OmniDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + async_add_entities([OmniPanelEvent(coordinator)]) + + +class OmniPanelEvent( + CoordinatorEntity[OmniDataUpdateCoordinator], EventEntity +): + """One event entity per panel; relays every push event the coordinator sees.""" + + _attr_has_entity_name = True + _attr_event_types: ClassVar[list[str]] = list(EVENT_TYPES) + + def __init__(self, coordinator: OmniDataUpdateCoordinator) -> None: + super().__init__(coordinator) + self._attr_unique_id = f"{coordinator.unique_id}-events" + self._attr_name = "Panel Events" + self._attr_device_info = coordinator.device_info + self._last_event_id: int | None = None + + def _handle_coordinator_update(self) -> None: + ev = self.coordinator.data.last_event + if ev is None: + return + # Only fire when the event reference actually changed; the + # coordinator may push other state without a new event arriving. + ev_id = id(ev) + if ev_id == self._last_event_id: + return + self._last_event_id = ev_id + + event_data: dict[str, Any] = {"event_class": type(ev).__name__} + for key in ( + "zone_index", "unit_index", "area_index", "user_index", + "new_state", "new_mode", "alarm_type", + ): + if hasattr(ev, key): + event_data[key] = getattr(ev, key) + + self._trigger_event(event_type_for(type(ev).__name__), event_data) + self.async_write_ha_state() diff --git a/custom_components/omni_pca/helpers.py b/custom_components/omni_pca/helpers.py index 9558ac6..a4dd8ed 100644 --- a/custom_components/omni_pca/helpers.py +++ b/custom_components/omni_pca/helpers.py @@ -149,3 +149,264 @@ def prettify_name(name: str) -> str: detect "no name configured on this index". """ return name.replace("_", " ").strip().title() + + +# -------------------------------------------------------------------------- +# Alarm panel state translation +# -------------------------------------------------------------------------- + +# String values matching HA's AlarmControlPanelState enum so this module +# stays importable without Home Assistant in the venv. +ALARM_STATE_DISARMED: Final = "disarmed" +ALARM_STATE_ARMED_HOME: Final = "armed_home" +ALARM_STATE_ARMED_AWAY: Final = "armed_away" +ALARM_STATE_ARMED_NIGHT: Final = "armed_night" +ALARM_STATE_ARMED_VACATION: Final = "armed_vacation" +ALARM_STATE_ARMED_CUSTOM_BYPASS: Final = "armed_custom_bypass" +ALARM_STATE_ARMING: Final = "arming" +ALARM_STATE_PENDING: Final = "pending" +ALARM_STATE_TRIGGERED: Final = "triggered" + + +# Maps SecurityMode (steady-state values) to HA alarm states. Arming-in- +# progress modes (9..14) get mapped via _ARMING_MODE_TO_FINAL — an arming +# area is always reported as ARMING regardless of the destination mode. +_SECURITY_MODE_TO_ALARM_STATE: dict[int, str] = { + 0: ALARM_STATE_DISARMED, + 1: ALARM_STATE_ARMED_HOME, # DAY + 2: ALARM_STATE_ARMED_NIGHT, # NIGHT + 3: ALARM_STATE_ARMED_AWAY, # AWAY + 4: ALARM_STATE_ARMED_VACATION, # VACATION + 5: ALARM_STATE_ARMED_CUSTOM_BYPASS, # DAY_INSTANT + 6: ALARM_STATE_ARMED_NIGHT, # NIGHT_DELAYED +} + +_ARMING_MODES: frozenset[int] = frozenset({9, 10, 11, 12, 13, 14}) + + +def security_mode_to_alarm_state( + mode: int, + alarm_active: bool = False, + entry_timer: int = 0, + exit_timer: int = 0, +) -> str: + """Map an Omni SecurityMode to a HA alarm_control_panel state string. + + Priority order: + 1. ``alarm_active`` → triggered + 2. ``entry_timer > 0`` → pending + 3. arming-in-progress modes or ``exit_timer > 0`` → arming + 4. steady-state mapping + """ + if alarm_active: + return ALARM_STATE_TRIGGERED + if entry_timer > 0: + return ALARM_STATE_PENDING + if mode in _ARMING_MODES or exit_timer > 0: + return ALARM_STATE_ARMING + return _SECURITY_MODE_TO_ALARM_STATE.get(mode, ALARM_STATE_DISARMED) + + +# Inverse for the four standard arm services HA exposes. Returned ints are +# the SecurityMode values to send via execute_security_command. +ARM_SERVICE_TO_SECURITY_MODE: dict[str, int] = { + "arm_home": 1, # DAY + "arm_away": 3, # AWAY + "arm_night": 2, # NIGHT + "arm_vacation": 4, # VACATION + "arm_custom_bypass": 5, # DAY_INSTANT + "disarm": 0, # OFF +} + + +# -------------------------------------------------------------------------- +# Light brightness conversion (Omni 0..100 ↔ HA 0..255) +# -------------------------------------------------------------------------- + + +def omni_state_to_ha_brightness(state: int) -> int | None: + """Decode a UnitStatus.state byte into HA brightness (1..255) or None. + + Returns None when the unit is off (state == 0). For state == 1 (plain + "on", non-dimmable) returns 255. For state in 100..200 returns + ``round((state - 100) * 255 / 100)`` clamped to 1..255. + """ + if state == 0: + return None + if state == 1: + return 255 + if 100 <= state <= 200: + percent = state - 100 + return max(1, min(255, round(percent * 255 / 100))) + # Scene levels (2..13) and ramping codes (17..25): treat as on, full. + return 255 + + +def ha_brightness_to_omni_percent(brightness: int) -> int: + """Convert HA's 1..255 brightness to Omni's 0..100 percent. + + Brightness 0 is invalid here (use turn_off); 1 maps to 1%, 255 to 100%. + """ + if brightness <= 0: + return 0 + if brightness >= 255: + return 100 + return max(1, min(100, round(brightness * 100 / 255))) + + +# -------------------------------------------------------------------------- +# HVAC mode translation +# -------------------------------------------------------------------------- + +HVAC_MODE_OFF: Final = "off" +HVAC_MODE_HEAT: Final = "heat" +HVAC_MODE_COOL: Final = "cool" +HVAC_MODE_HEAT_COOL: Final = "heat_cool" +HVAC_MODE_AUX_HEAT: Final = "heat" # HA collapses emergency-heat into heat + preset + +_OMNI_HVAC_TO_HA: dict[int, str] = { + 0: HVAC_MODE_OFF, + 1: HVAC_MODE_HEAT, + 2: HVAC_MODE_COOL, + 3: HVAC_MODE_HEAT_COOL, + 4: HVAC_MODE_HEAT, # EMERGENCY_HEAT — HA treats as heat + a preset +} + +_HA_HVAC_TO_OMNI: dict[str, int] = { + HVAC_MODE_OFF: 0, + HVAC_MODE_HEAT: 1, + HVAC_MODE_COOL: 2, + HVAC_MODE_HEAT_COOL: 3, +} + + +def omni_hvac_to_ha(mode: int) -> str: + return _OMNI_HVAC_TO_HA.get(mode, HVAC_MODE_OFF) + + +def ha_hvac_to_omni(mode: str) -> int: + return _HA_HVAC_TO_OMNI.get(mode, 0) + + +_OMNI_FAN_TO_HA: dict[int, str] = {0: "auto", 1: "on", 2: "diffuse"} +_HA_FAN_TO_OMNI: dict[str, int] = {"auto": 0, "on": 1, "diffuse": 2, "cycle": 2} + + +def omni_fan_to_ha(mode: int) -> str: + return _OMNI_FAN_TO_HA.get(mode, "auto") + + +def ha_fan_to_omni(mode: str) -> int: + return _HA_FAN_TO_OMNI.get(mode, 0) + + +_OMNI_HOLD_TO_HA: dict[int, str] = {0: "none", 1: "hold", 2: "vacation", 0xFF: "hold"} +_HA_HOLD_TO_OMNI: dict[str, int] = {"none": 0, "hold": 1, "vacation": 2} + + +def omni_hold_to_ha(mode: int) -> str: + return _OMNI_HOLD_TO_HA.get(mode, "none") + + +def ha_hold_to_omni(mode: str) -> int: + return _HA_HOLD_TO_OMNI.get(mode, 0) + + +# -------------------------------------------------------------------------- +# Temperature: HA °F → Omni raw byte +# -------------------------------------------------------------------------- +# +# Omni encodes temperature linearly. Per clsText.cs (DecodeTempRaw): +# °F = round(raw * 9 / 10) - 40 +# °C = raw / 2 - 40 +# Inverse: +# raw = round((°F + 40) * 10 / 9) + + +def fahrenheit_to_omni_raw(f: float) -> int: + """Inverse of omni_temp_to_fahrenheit. Clamps to the valid 0..255 byte.""" + raw = round((f + 40) * 10 / 9) + return max(0, min(255, raw)) + + +def celsius_to_omni_raw(c: float) -> int: + """Inverse of omni_temp_to_celsius. Clamps to the valid 0..255 byte.""" + raw = round((c + 40) * 2) + return max(0, min(255, raw)) + + +# -------------------------------------------------------------------------- +# Analog zone → sensor device class +# -------------------------------------------------------------------------- + +SENSOR_DEVICE_CLASS_TEMPERATURE: Final = "temperature" +SENSOR_DEVICE_CLASS_HUMIDITY: Final = "humidity" +SENSOR_DEVICE_CLASS_POWER: Final = "power" + +_ANALOG_ZONE_TYPE_TO_DEVICE_CLASS: dict[int, str] = { + 80: SENSOR_DEVICE_CLASS_POWER, # ENERGY_SAVER + 81: SENSOR_DEVICE_CLASS_TEMPERATURE, # OUTDOOR_TEMP + 82: SENSOR_DEVICE_CLASS_TEMPERATURE, # TEMPERATURE + 83: SENSOR_DEVICE_CLASS_TEMPERATURE, # TEMP_ALARM + 84: SENSOR_DEVICE_CLASS_HUMIDITY, # HUMIDITY +} + + +def analog_zone_device_class(zone_type: int) -> str | None: + """Return the HA SensorDeviceClass string for an analog zone, or None.""" + return _ANALOG_ZONE_TYPE_TO_DEVICE_CLASS.get(zone_type) + + +# -------------------------------------------------------------------------- +# Event surfacing +# -------------------------------------------------------------------------- + +# Snake_case event-type strings exposed by the EventEntity. +EVENT_TYPE_ZONE_STATE_CHANGED: Final = "zone_state_changed" +EVENT_TYPE_UNIT_STATE_CHANGED: Final = "unit_state_changed" +EVENT_TYPE_ARMING_CHANGED: Final = "arming_changed" +EVENT_TYPE_ALARM_ACTIVATED: Final = "alarm_activated" +EVENT_TYPE_ALARM_CLEARED: Final = "alarm_cleared" +EVENT_TYPE_AC_LOST: Final = "ac_lost" +EVENT_TYPE_AC_RESTORED: Final = "ac_restored" +EVENT_TYPE_BATTERY_LOW: Final = "battery_low" +EVENT_TYPE_BATTERY_RESTORED: Final = "battery_restored" +EVENT_TYPE_USER_MACRO_BUTTON: Final = "user_macro_button" +EVENT_TYPE_PHONE_LINE_DEAD: Final = "phone_line_dead" +EVENT_TYPE_PHONE_LINE_RESTORED: Final = "phone_line_restored" +EVENT_TYPE_UNKNOWN: Final = "unknown" + +EVENT_TYPES: tuple[str, ...] = ( + EVENT_TYPE_ZONE_STATE_CHANGED, + EVENT_TYPE_UNIT_STATE_CHANGED, + EVENT_TYPE_ARMING_CHANGED, + EVENT_TYPE_ALARM_ACTIVATED, + EVENT_TYPE_ALARM_CLEARED, + EVENT_TYPE_AC_LOST, + EVENT_TYPE_AC_RESTORED, + EVENT_TYPE_BATTERY_LOW, + EVENT_TYPE_BATTERY_RESTORED, + EVENT_TYPE_USER_MACRO_BUTTON, + EVENT_TYPE_PHONE_LINE_DEAD, + EVENT_TYPE_PHONE_LINE_RESTORED, + EVENT_TYPE_UNKNOWN, +) + + +def event_type_for(class_name: str) -> str: + """Map a SystemEvent subclass name to its snake_case event type.""" + mapping = { + "ZoneStateChanged": EVENT_TYPE_ZONE_STATE_CHANGED, + "UnitStateChanged": EVENT_TYPE_UNIT_STATE_CHANGED, + "ArmingChanged": EVENT_TYPE_ARMING_CHANGED, + "AlarmActivated": EVENT_TYPE_ALARM_ACTIVATED, + "AlarmCleared": EVENT_TYPE_ALARM_CLEARED, + "AcLost": EVENT_TYPE_AC_LOST, + "AcRestored": EVENT_TYPE_AC_RESTORED, + "BatteryLow": EVENT_TYPE_BATTERY_LOW, + "BatteryRestored": EVENT_TYPE_BATTERY_RESTORED, + "UserMacroButton": EVENT_TYPE_USER_MACRO_BUTTON, + "PhoneLineDead": EVENT_TYPE_PHONE_LINE_DEAD, + "PhoneLineRestored": EVENT_TYPE_PHONE_LINE_RESTORED, + } + return mapping.get(class_name, EVENT_TYPE_UNKNOWN) diff --git a/custom_components/omni_pca/light.py b/custom_components/omni_pca/light.py new file mode 100644 index 0000000..85a0a87 --- /dev/null +++ b/custom_components/omni_pca/light.py @@ -0,0 +1,115 @@ +"""Light platform — one HA light entity per discovered Omni unit. + +We expose every unit as a dimmable light. On non-dimmable units the +panel silently ignores the brightness component and just toggles, so +the worst case is a relay that ignores the slider — no harm done. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, ClassVar + +from homeassistant.components.light import ( + ATTR_BRIGHTNESS, + ColorMode, + LightEntity, +) +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from omni_pca.commands import CommandFailedError + +from .const import DOMAIN +from .coordinator import OmniDataUpdateCoordinator +from .helpers import ( + ha_brightness_to_omni_percent, + omni_state_to_ha_brightness, + prettify_name, +) + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + from homeassistant.helpers.entity_platform import AddEntitiesCallback + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + coordinator: OmniDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + entities = [ + OmniUnitLight(coordinator, index) + for index in sorted(coordinator.data.units) + ] + async_add_entities(entities) + + +class OmniUnitLight(CoordinatorEntity[OmniDataUpdateCoordinator], LightEntity): + """One discovered unit as a HA light.""" + + _attr_has_entity_name = True + _attr_color_mode = ColorMode.BRIGHTNESS + _attr_supported_color_modes: ClassVar[set[ColorMode]] = {ColorMode.BRIGHTNESS} + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator) + self._index = index + self._attr_unique_id = f"{coordinator.unique_id}-unit-{index}" + props = coordinator.data.units[index] + self._attr_name = prettify_name(props.name) or f"Unit {index}" + self._attr_device_info = coordinator.device_info + + @property + def available(self) -> bool: + return ( + super().available + and self.coordinator.data is not None + and self._index in self.coordinator.data.units + ) + + @property + def is_on(self) -> bool | None: + status = self.coordinator.data.unit_status.get(self._index) + if status is None: + return None + return status.is_on + + @property + def brightness(self) -> int | None: + status = self.coordinator.data.unit_status.get(self._index) + if status is None: + return None + return omni_state_to_ha_brightness(status.state) + + @property + def extra_state_attributes(self) -> dict[str, Any] | None: + status = self.coordinator.data.unit_status.get(self._index) + if status is None: + return None + return { + "unit_index": self._index, + "raw_state": status.state, + "time_remaining_secs": status.time_remaining_secs, + } + + async def async_turn_on(self, **kwargs: Any) -> None: + try: + if ATTR_BRIGHTNESS in kwargs: + percent = ha_brightness_to_omni_percent(int(kwargs[ATTR_BRIGHTNESS])) + await self.coordinator.client.set_unit_level(self._index, percent) + else: + await self.coordinator.client.turn_unit_on(self._index) + except CommandFailedError as err: + raise HomeAssistantError(f"Panel rejected command: {err}") from err + await self.coordinator.async_request_refresh() + + async def async_turn_off(self, **kwargs: Any) -> None: + try: + await self.coordinator.client.turn_unit_off(self._index) + except CommandFailedError as err: + raise HomeAssistantError(f"Panel rejected command: {err}") from err + await self.coordinator.async_request_refresh() diff --git a/custom_components/omni_pca/sensor.py b/custom_components/omni_pca/sensor.py new file mode 100644 index 0000000..d99da4b --- /dev/null +++ b/custom_components/omni_pca/sensor.py @@ -0,0 +1,263 @@ +"""Sensor platform — analog zones, thermostat readings, panel telemetry. + +We deliberately re-expose thermostat current_temperature / humidity as +diagnostic sensors (in addition to the climate entity) so users can +plot history. The climate entity remains the canonical control surface. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorStateClass, +) +from homeassistant.const import PERCENTAGE, UnitOfTemperature +from homeassistant.helpers.entity import EntityCategory +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from .const import DOMAIN +from .coordinator import OmniDataUpdateCoordinator +from .helpers import ( + SENSOR_DEVICE_CLASS_HUMIDITY, + SENSOR_DEVICE_CLASS_TEMPERATURE, + analog_zone_device_class, + is_binary_zone_type, + prettify_name, +) + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + from homeassistant.helpers.entity_platform import AddEntitiesCallback + + +_DEVICE_CLASS_STR_TO_ENUM: dict[str, SensorDeviceClass] = { + SENSOR_DEVICE_CLASS_TEMPERATURE: SensorDeviceClass.TEMPERATURE, + SENSOR_DEVICE_CLASS_HUMIDITY: SensorDeviceClass.HUMIDITY, + "power": SensorDeviceClass.POWER, +} + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + coordinator: OmniDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + entities: list[SensorEntity] = [] + + # Analog zones (temperature / humidity / energy) + for index in sorted(coordinator.data.zones): + props = coordinator.data.zones[index] + if is_binary_zone_type(props.zone_type): + continue + device_class_str = analog_zone_device_class(props.zone_type) + if device_class_str is None: + continue + entities.append( + OmniAnalogZoneSensor(coordinator, index, device_class_str) + ) + + # Per-thermostat diagnostic sensors + for index in sorted(coordinator.data.thermostats): + entities.append(OmniThermostatTempSensor(coordinator, index)) + entities.append(OmniThermostatHumiditySensor(coordinator, index)) + entities.append(OmniThermostatOutdoorTempSensor(coordinator, index)) + + entities.append(OmniSystemModelSensor(coordinator)) + entities.append(OmniLastEventSensor(coordinator)) + + async_add_entities(entities) + + +# -------------------------------------------------------------------------- +# Analog zones +# -------------------------------------------------------------------------- + + +class OmniAnalogZoneSensor( + CoordinatorEntity[OmniDataUpdateCoordinator], SensorEntity +): + _attr_has_entity_name = True + _attr_state_class = SensorStateClass.MEASUREMENT + + def __init__( + self, + coordinator: OmniDataUpdateCoordinator, + index: int, + device_class_str: str, + ) -> None: + super().__init__(coordinator) + self._index = index + self._attr_unique_id = f"{coordinator.unique_id}-zone-{index}-analog" + props = coordinator.data.zones[index] + self._attr_name = prettify_name(props.name) or f"Zone {index}" + self._attr_device_info = coordinator.device_info + self._attr_device_class = _DEVICE_CLASS_STR_TO_ENUM.get(device_class_str) + if device_class_str == SENSOR_DEVICE_CLASS_TEMPERATURE: + self._attr_native_unit_of_measurement = UnitOfTemperature.FAHRENHEIT + elif device_class_str == SENSOR_DEVICE_CLASS_HUMIDITY: + self._attr_native_unit_of_measurement = PERCENTAGE + + @property + def native_value(self) -> float | int | None: + status = self.coordinator.data.zone_status.get(self._index) + if status is None: + return None + # Reuse the linear temp formula for temperature zones; humidity + # zones report the loop byte as the percentage directly. + if self._attr_device_class == SensorDeviceClass.TEMPERATURE: + return round(status.loop * 9 / 10) - 40 + return status.loop + + +# -------------------------------------------------------------------------- +# Thermostat diagnostic sensors +# -------------------------------------------------------------------------- + + +class _ThermostatBase(CoordinatorEntity[OmniDataUpdateCoordinator], SensorEntity): + _attr_has_entity_name = True + _attr_entity_category = EntityCategory.DIAGNOSTIC + _attr_state_class = SensorStateClass.MEASUREMENT + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator) + self._index = index + self._attr_device_info = coordinator.device_info + + @property + def _status(self): # type: ignore[no-untyped-def] + return self.coordinator.data.thermostat_status.get(self._index) + + +class OmniThermostatTempSensor(_ThermostatBase): + _attr_device_class = SensorDeviceClass.TEMPERATURE + _attr_native_unit_of_measurement = UnitOfTemperature.FAHRENHEIT + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator, index) + self._attr_unique_id = f"{coordinator.unique_id}-thermostat-{index}-temp" + props = coordinator.data.thermostats[index] + base = prettify_name(props.name) or f"Thermostat {index}" + self._attr_name = f"{base} Temperature" + + @property + def native_value(self) -> float | None: + s = self._status + if s is None or s.temperature_raw == 0: + return None + return round(s.temperature_f, 1) + + +class OmniThermostatHumiditySensor(_ThermostatBase): + _attr_device_class = SensorDeviceClass.HUMIDITY + _attr_native_unit_of_measurement = PERCENTAGE + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator, index) + self._attr_unique_id = f"{coordinator.unique_id}-thermostat-{index}-humidity" + props = coordinator.data.thermostats[index] + base = prettify_name(props.name) or f"Thermostat {index}" + self._attr_name = f"{base} Humidity" + + @property + def native_value(self) -> int | None: + s = self._status + if s is None or s.humidity_raw == 0: + return None + return int(s.humidity_raw) + + +class OmniThermostatOutdoorTempSensor(_ThermostatBase): + _attr_device_class = SensorDeviceClass.TEMPERATURE + _attr_native_unit_of_measurement = UnitOfTemperature.FAHRENHEIT + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator, index) + self._attr_unique_id = f"{coordinator.unique_id}-thermostat-{index}-outdoor" + props = coordinator.data.thermostats[index] + base = prettify_name(props.name) or f"Thermostat {index}" + self._attr_name = f"{base} Outdoor Temperature" + + @property + def native_value(self) -> float | None: + s = self._status + if s is None or s.outdoor_temperature_raw == 0: + return None + return round(s.outdoor_temperature_f, 1) + + +# -------------------------------------------------------------------------- +# Panel telemetry +# -------------------------------------------------------------------------- + + +class OmniSystemModelSensor( + CoordinatorEntity[OmniDataUpdateCoordinator], SensorEntity +): + """Static text sensor: model + firmware. Helps confirm the integration + talked to the panel without needing to dig into Devices & Services.""" + + _attr_has_entity_name = True + _attr_entity_category = EntityCategory.DIAGNOSTIC + + def __init__(self, coordinator: OmniDataUpdateCoordinator) -> None: + super().__init__(coordinator) + self._attr_unique_id = f"{coordinator.unique_id}-system-model" + self._attr_name = "Panel Model" + self._attr_device_info = coordinator.device_info + + @property + def native_value(self) -> str | None: + info = self.coordinator.data.system_info + if info is None: + return None + return f"{info.model_name} {info.firmware_version}" + + +class OmniLastEventSensor( + CoordinatorEntity[OmniDataUpdateCoordinator], SensorEntity +): + """Diagnostic text sensor showing the most recent push event class name.""" + + _attr_has_entity_name = True + _attr_entity_category = EntityCategory.DIAGNOSTIC + + def __init__(self, coordinator: OmniDataUpdateCoordinator) -> None: + super().__init__(coordinator) + self._attr_unique_id = f"{coordinator.unique_id}-last-event" + self._attr_name = "Last Panel Event" + self._attr_device_info = coordinator.device_info + + @property + def native_value(self) -> str | None: + ev = self.coordinator.data.last_event + if ev is None: + return None + return type(ev).__name__ + + @property + def extra_state_attributes(self) -> dict[str, Any] | None: + ev = self.coordinator.data.last_event + if ev is None: + return None + result: dict[str, Any] = {"event_class": type(ev).__name__} + for key in ( + "zone_index", "unit_index", "area_index", "user_index", + "new_state", "new_mode", "alarm_type", + ): + if hasattr(ev, key): + result[key] = getattr(ev, key) + return result diff --git a/custom_components/omni_pca/switch.py b/custom_components/omni_pca/switch.py new file mode 100644 index 0000000..77c5ed2 --- /dev/null +++ b/custom_components/omni_pca/switch.py @@ -0,0 +1,89 @@ +"""Switch platform — per-zone bypass control. + +Lights are exposed via the ``light`` platform. The switch platform is +reserved for *configuration* toggles like zone bypass, where the user +wants a write surface that pairs with the diagnostic ``zone bypassed`` +binary sensor for read. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from homeassistant.components.switch import SwitchEntity +from homeassistant.exceptions import HomeAssistantError +from homeassistant.helpers.entity import EntityCategory +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from omni_pca.commands import CommandFailedError + +from .const import DOMAIN +from .coordinator import OmniDataUpdateCoordinator +from .helpers import is_binary_zone_type, prettify_name + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + from homeassistant.helpers.entity_platform import AddEntitiesCallback + + +async def async_setup_entry( + hass: HomeAssistant, + entry: ConfigEntry, + async_add_entities: AddEntitiesCallback, +) -> None: + coordinator: OmniDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id] + entities: list[SwitchEntity] = [] + for index in sorted(coordinator.data.zones): + props = coordinator.data.zones[index] + if not is_binary_zone_type(props.zone_type): + continue + entities.append(OmniZoneBypassSwitch(coordinator, index)) + async_add_entities(entities) + + +class OmniZoneBypassSwitch(CoordinatorEntity[OmniDataUpdateCoordinator], SwitchEntity): + """Toggle that bypasses or restores a single zone.""" + + _attr_has_entity_name = True + _attr_entity_category = EntityCategory.CONFIG + + def __init__( + self, coordinator: OmniDataUpdateCoordinator, index: int + ) -> None: + super().__init__(coordinator) + self._index = index + self._attr_unique_id = f"{coordinator.unique_id}-zone-{index}-bypass" + props = coordinator.data.zones[index] + base = prettify_name(props.name) or f"Zone {index}" + self._attr_name = f"{base} Bypass" + self._attr_device_info = coordinator.device_info + + @property + def available(self) -> bool: + return ( + super().available + and self.coordinator.data is not None + and self._index in self.coordinator.data.zones + ) + + @property + def is_on(self) -> bool | None: + status = self.coordinator.data.zone_status.get(self._index) + if status is None: + return None + return status.is_bypassed + + async def async_turn_on(self, **kwargs: Any) -> None: + try: + await self.coordinator.client.bypass_zone(self._index) + except CommandFailedError as err: + raise HomeAssistantError(f"Panel rejected bypass: {err}") from err + await self.coordinator.async_request_refresh() + + async def async_turn_off(self, **kwargs: Any) -> None: + try: + await self.coordinator.client.restore_zone(self._index) + except CommandFailedError as err: + raise HomeAssistantError(f"Panel rejected restore: {err}") from err + await self.coordinator.async_request_refresh() diff --git a/tests/test_ha_helpers.py b/tests/test_ha_helpers.py index 7a0a061..18bf010 100644 --- a/tests/test_ha_helpers.py +++ b/tests/test_ha_helpers.py @@ -107,3 +107,198 @@ class TestPrettifyName: ) def test_round_trip(self, raw: str, expected: str) -> None: assert helpers.prettify_name(raw) == expected + + +# ---------------------------------------------------------------------------- +# Phase B helpers — pure functions used by the new entity platforms +# ---------------------------------------------------------------------------- + + +class TestSecurityModeToAlarmState: + @pytest.mark.parametrize( + ("mode", "expected"), + [ + (0, "disarmed"), + (1, "armed_home"), # DAY + (2, "armed_night"), # NIGHT + (3, "armed_away"), # AWAY + (4, "armed_vacation"), # VACATION + (5, "armed_custom_bypass"), # DAY_INSTANT + (6, "armed_night"), # NIGHT_DELAYED + ], + ) + def test_steady_state(self, mode: int, expected: str) -> None: + assert helpers.security_mode_to_alarm_state(mode) == expected + + def test_alarm_active_overrides(self) -> None: + assert helpers.security_mode_to_alarm_state(3, alarm_active=True) == "triggered" + + def test_entry_timer_pending(self) -> None: + assert helpers.security_mode_to_alarm_state(3, entry_timer=15) == "pending" + + def test_exit_timer_arming(self) -> None: + assert helpers.security_mode_to_alarm_state(3, exit_timer=30) == "arming" + + @pytest.mark.parametrize("arming_mode", [9, 10, 11, 12, 13, 14]) + def test_arming_in_progress_modes(self, arming_mode: int) -> None: + assert helpers.security_mode_to_alarm_state(arming_mode) == "arming" + + def test_unknown_mode_falls_back_to_disarmed(self) -> None: + assert helpers.security_mode_to_alarm_state(99) == "disarmed" + + +class TestArmServiceMapping: + def test_all_services_present(self) -> None: + for svc in ("arm_home", "arm_away", "arm_night", "arm_vacation", + "arm_custom_bypass", "disarm"): + assert svc in helpers.ARM_SERVICE_TO_SECURITY_MODE + + def test_round_trip_through_alarm_state(self) -> None: + # Arm with each service, decode the resulting mode back to the HA + # state, and verify the names are sensible. + for svc, expected_state in [ + ("disarm", "disarmed"), + ("arm_home", "armed_home"), + ("arm_away", "armed_away"), + ("arm_night", "armed_night"), + ("arm_vacation", "armed_vacation"), + ]: + mode = helpers.ARM_SERVICE_TO_SECURITY_MODE[svc] + assert helpers.security_mode_to_alarm_state(mode) == expected_state + + +class TestBrightnessConversions: + @pytest.mark.parametrize( + ("state", "expected"), + [ + (0, None), # off + (1, 255), # plain on, non-dimmable + (100, 1), # 0% via Omni's overlap (level 100 = 0%, but we floor at 1) + (150, 128), # 50% + (200, 255), # 100% + ], + ) + def test_state_to_ha_brightness(self, state: int, expected: int | None) -> None: + result = helpers.omni_state_to_ha_brightness(state) + assert result == expected + + @pytest.mark.parametrize( + ("brightness", "expected_percent"), + [ + (1, 1), + (128, 50), + (255, 100), + ], + ) + def test_ha_brightness_to_omni_percent( + self, brightness: int, expected_percent: int + ) -> None: + assert helpers.ha_brightness_to_omni_percent(brightness) == expected_percent + + def test_zero_brightness(self) -> None: + assert helpers.ha_brightness_to_omni_percent(0) == 0 + + +class TestHvacFanHoldRoundTrip: + @pytest.mark.parametrize( + ("omni_mode", "expected_ha"), + [(0, "off"), (1, "heat"), (2, "cool"), (3, "heat_cool"), (4, "heat")], + ) + def test_hvac_mapping(self, omni_mode: int, expected_ha: str) -> None: + assert helpers.omni_hvac_to_ha(omni_mode) == expected_ha + + @pytest.mark.parametrize( + ("ha_mode", "expected_omni"), + [("off", 0), ("heat", 1), ("cool", 2), ("heat_cool", 3)], + ) + def test_hvac_inverse(self, ha_mode: str, expected_omni: int) -> None: + assert helpers.ha_hvac_to_omni(ha_mode) == expected_omni + + def test_fan_round_trip(self) -> None: + for omni in (0, 1, 2): + ha = helpers.omni_fan_to_ha(omni) + back = helpers.ha_fan_to_omni(ha) + assert back == omni + + def test_hold_round_trip(self) -> None: + for omni in (0, 1, 2): + ha = helpers.omni_hold_to_ha(omni) + back = helpers.ha_hold_to_omni(ha) + assert back == omni + + def test_legacy_old_on_hold_value(self) -> None: + # Old firmware sentinel 0xFF should map to the same HA preset as 1. + assert helpers.omni_hold_to_ha(0xFF) == helpers.omni_hold_to_ha(1) + + +class TestTemperatureInverse: + @pytest.mark.parametrize( + ("fahrenheit", "expected_raw"), + [ + (-40, 0), # bottom of the scale + (0, 44), # ~0°F + (32, 80), # freezing + (72, 124), # room temp + (212, 280), # boiling — above byte range, gets clamped + ], + ) + def test_fahrenheit_to_raw(self, fahrenheit: float, expected_raw: int) -> None: + result = helpers.fahrenheit_to_omni_raw(fahrenheit) + # We clamp to 0..255 so 212°F (would compute 280) becomes 255. + if expected_raw > 255: + assert result == 255 + else: + assert result == expected_raw + + def test_inverse_round_trip_at_typical_setpoints(self) -> None: + # Take a few raw values, decode to °F via the linear formula, encode + # back, and verify we get the same byte (within ±1 due to rounding). + for raw in (80, 100, 124, 144, 168, 184): + fahrenheit = round(raw * 9 / 10) - 40 + back = helpers.fahrenheit_to_omni_raw(fahrenheit) + assert abs(back - raw) <= 1 + + +class TestAnalogZoneDeviceClass: + @pytest.mark.parametrize( + ("zone_type", "expected"), + [ + (80, "power"), + (81, "temperature"), + (82, "temperature"), + (83, "temperature"), + (84, "humidity"), + (1, None), # binary zone — not analog + (255, None), # unknown + ], + ) + def test_mapping(self, zone_type: int, expected: str | None) -> None: + assert helpers.analog_zone_device_class(zone_type) == expected + + +class TestEventTypeFor: + @pytest.mark.parametrize( + ("class_name", "expected"), + [ + ("ZoneStateChanged", "zone_state_changed"), + ("UnitStateChanged", "unit_state_changed"), + ("ArmingChanged", "arming_changed"), + ("AlarmActivated", "alarm_activated"), + ("AlarmCleared", "alarm_cleared"), + ("AcLost", "ac_lost"), + ("AcRestored", "ac_restored"), + ("BatteryLow", "battery_low"), + ("BatteryRestored", "battery_restored"), + ("UserMacroButton", "user_macro_button"), + ("PhoneLineDead", "phone_line_dead"), + ("PhoneLineRestored", "phone_line_restored"), + ], + ) + def test_known_events(self, class_name: str, expected: str) -> None: + assert helpers.event_type_for(class_name) == expected + + def test_unknown_event_class(self) -> None: + assert helpers.event_type_for("SomeRandomThing") == "unknown" + + def test_event_types_tuple_includes_unknown(self) -> None: + assert "unknown" in helpers.EVENT_TYPES