HA Phase B: alarm + light + switch + climate + sensor + button + event
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/.
This commit is contained in:
parent
e8ed7d1b89
commit
57b8aa4b04
@ -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:
|
||||
|
||||
156
custom_components/omni_pca/alarm_control_panel.py
Normal file
156
custom_components/omni_pca/alarm_control_panel.py
Normal file
@ -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)
|
||||
62
custom_components/omni_pca/button.py
Normal file
62
custom_components/omni_pca/button.py
Normal file
@ -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
|
||||
271
custom_components/omni_pca/climate.py
Normal file
271
custom_components/omni_pca/climate.py
Normal file
@ -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
|
||||
)
|
||||
)
|
||||
68
custom_components/omni_pca/event.py
Normal file
68
custom_components/omni_pca/event.py
Normal file
@ -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()
|
||||
@ -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)
|
||||
|
||||
115
custom_components/omni_pca/light.py
Normal file
115
custom_components/omni_pca/light.py
Normal file
@ -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()
|
||||
263
custom_components/omni_pca/sensor.py
Normal file
263
custom_components/omni_pca/sensor.py
Normal file
@ -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
|
||||
89
custom_components/omni_pca/switch.py
Normal file
89
custom_components/omni_pca/switch.py
Normal file
@ -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()
|
||||
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user