jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 04:37:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve Elmax alarm control panel (#117689)

Browse Source
This commit is contained in:
Alberto Geniola 2024-07-06 18:21:15 +02:00 committed by GitHub
parent 131d9ec51b
commit 490dd53edf
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 142 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
homeassistant/components/elmax/__init__.py
View File

@ -10,7 +10,8 @@ from elmax_api.http import Elmax, ElmaxLocal, GenericElmax
from elmax_api.model.panel import PanelEntry
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.const import EVENT_HOMEASSISTANT_STOP
from homeassistant.core import Event, HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from .common import DirectPanel, build_direct_ssl_context, get_direct_api_url
@ -104,6 +105,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
update_interval=timedelta(seconds=POLLING_SECONDS),
)
async def _async_on_hass_stop(_: Event) -> None:
"""Close connection when hass stops."""
await coordinator.async_shutdown()
entry.async_on_unload(
hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, _async_on_hass_stop)
)
# Issue a first refresh, so that we trigger a re-auth flow if necessary
await coordinator.async_config_entry_first_refresh()

83
homeassistant/components/elmax/alarm_control_panel.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations
from elmax_api.exceptions import ElmaxApiError
from elmax_api.model.alarm_status import AlarmArmStatus, AlarmStatus
from elmax_api.model.command import AreaCommand
from elmax_api.model.panel import PanelStatus
@ -12,10 +13,17 @@ from homeassistant.components.alarm_control_panel import (
CodeFormat,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import STATE_ALARM_ARMED_AWAY, STATE_ALARM_DISARMED
from homeassistant.const import (
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMING,
STATE_ALARM_DISARMED,
STATE_ALARM_DISARMING,
STATE_ALARM_TRIGGERED,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import InvalidStateError
from homeassistant.exceptions import HomeAssistantError, InvalidStateError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .common import ElmaxEntity
from .const import DOMAIN
@ -66,6 +74,7 @@ class ElmaxArea(ElmaxEntity, AlarmControlPanelEntity):
_attr_code_arm_required = False
_attr_has_entity_name = True
_attr_supported_features = AlarmControlPanelEntityFeature.ARM_AWAY
_pending_state: str | None = None
async def async_alarm_arm_away(self, code: str | None = None) -> None:
"""Send arm away command."""
@ -74,31 +83,70 @@ class ElmaxArea(ElmaxEntity, AlarmControlPanelEntity):
f"Cannot arm {self.name}: please check for open windows/doors first"
)
await self.coordinator.http_client.execute_command(
endpoint_id=self._device.endpoint_id,
command=AreaCommand.ARM_TOTALLY,
extra_payload={"code": code},
)
await self.coordinator.async_refresh()
self._pending_state = STATE_ALARM_ARMING
self.async_write_ha_state()
try:
await self.coordinator.http_client.execute_command(
endpoint_id=self._device.endpoint_id,
command=AreaCommand.ARM_TOTALLY,
extra_payload={"code": code},
)
except ElmaxApiError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="alarm_operation_failed_generic",
translation_placeholders={"operation": "arm"},
) from err
finally:
await self.coordinator.async_refresh()
async def async_alarm_disarm(self, code: str | None = None) -> None:
"""Send disarm command."""
# Elmax alarm panels do always require a code to be passed for disarm operations
if code is None or code == "":
raise ValueError("Please input the disarm code.")
await self.coordinator.http_client.execute_command(
endpoint_id=self._device.endpoint_id,
command=AreaCommand.DISARM,
extra_payload={"code": code},
)
await self.coordinator.async_refresh()
self._pending_state = STATE_ALARM_DISARMING
self.async_write_ha_state()
try:
await self.coordinator.http_client.execute_command(
endpoint_id=self._device.endpoint_id,
command=AreaCommand.DISARM,
extra_payload={"code": code},
)
except ElmaxApiError as err:
if err.status_code == 403:
raise HomeAssistantError(
translation_domain=DOMAIN, translation_key="invalid_disarm_code"
) from err
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="alarm_operation_failed_generic",
translation_placeholders={"operation": "disarm"},
) from err
finally:
await self.coordinator.async_refresh()
@property
def state(self) -> StateType:
"""Return the state of the entity."""
if self._pending_state is not None:
return self._pending_state
if (
state := self.coordinator.get_area_state(self._device.endpoint_id)
) is not None:
if state.status == AlarmStatus.TRIGGERED:
return ALARM_STATE_TO_HA.get(AlarmStatus.TRIGGERED)
return ALARM_STATE_TO_HA.get(state.armed_status)
return None
@callback
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._attr_state = ALARM_STATE_TO_HA.get(
self.coordinator.get_area_state(self._device.endpoint_id).armed_status
)
# Just reset the local pending_state so that it no longer overrides the one from coordinator.
self._pending_state = None
super()._handle_coordinator_update()
@ -108,4 +156,5 @@ ALARM_STATE_TO_HA = {
AlarmArmStatus.ARMED_P2: STATE_ALARM_ARMED_AWAY,
AlarmArmStatus.ARMED_P1: STATE_ALARM_ARMED_AWAY,
AlarmArmStatus.NOT_ARMED: STATE_ALARM_DISARMED,
AlarmStatus.TRIGGERED: STATE_ALARM_TRIGGERED,
}

58
homeassistant/components/elmax/coordinator.py
View File

@ -17,7 +17,9 @@ from elmax_api.http import Elmax, GenericElmax
from elmax_api.model.actuator import Actuator
from elmax_api.model.area import Area
from elmax_api.model.cover import Cover
from elmax_api.model.endpoint import DeviceEndpoint
from elmax_api.model.panel import PanelEntry, PanelStatus
from elmax_api.push.push import PushNotificationHandler
from httpx import ConnectError, ConnectTimeout
from homeassistant.core import HomeAssistant
@ -30,6 +32,8 @@ from .const import DEFAULT_TIMEOUT
class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]):
"""Coordinator helper to handle Elmax API polling."""
_state_by_endpoint: dict[str, Actuator | Area | Cover | DeviceEndpoint]
def __init__(
self,
hass: HomeAssistant,
@ -42,7 +46,8 @@ class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]):
"""Instantiate the object."""
self._client = elmax_api_client
self._panel_entry = panel
self._state_by_endpoint = None
self._state_by_endpoint = {}
self._push_notification_handler = None
super().__init__(
hass=hass, logger=logger, name=name, update_interval=update_interval
)
@ -93,12 +98,6 @@ class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]):
# We handle this case in the following exception blocks.
status = await self._client.get_current_panel_status()
# Store a dictionary for fast endpoint state access
self._state_by_endpoint = {
k.endpoint_id: k for k in status.all_endpoints
}
return status
except ElmaxBadPinError as err:
raise ConfigEntryAuthFailed("Control panel pin was refused") from err
except ElmaxBadLoginError as err:
@ -122,3 +121,48 @@ class ElmaxCoordinator(DataUpdateCoordinator[PanelStatus]):
"Make sure the panel is online and that "
"your firewall allows communication with it."
) from err
# Store a dictionary for fast endpoint state access
self._state_by_endpoint = {k.endpoint_id: k for k in status.all_endpoints}
# If panel supports it and a it hasn't been registered yet, register the push notification handler
if status.push_feature and self._push_notification_handler is None:
self._register_push_notification_handler()
self._fire_data_update(status)
return status
def _fire_data_update(self, status: PanelStatus):
# Store a dictionary for fast endpoint state access
self._state_by_endpoint = {k.endpoint_id: k for k in status.all_endpoints}
self.async_set_updated_data(status)
def _register_push_notification_handler(self):
ws_ep = (
f"{'wss' if self.http_client.base_url.scheme == 'https' else 'ws'}"
f"://{self.http_client.base_url.host}"
f":{self.http_client.base_url.port}"
f"{self.http_client.base_url.path}/push"
)
self._push_notification_handler = PushNotificationHandler(
endpoint=str(ws_ep),
http_client=self.http_client,
ssl_context=self.http_client.ssl_context,
)
self._push_notification_handler.register_push_notification_handler(
self._push_handler
)
self._push_notification_handler.start(loop=self.hass.loop)
async def _push_handler(self, status: PanelStatus) -> None:
self._fire_data_update(status)
async def async_shutdown(self) -> None:
"""Cancel any scheduled call, and ignore new runs."""
if self._push_notification_handler is not None:
self._push_notification_handler.unregister_push_notification_handler(
self._push_handler
)
self._push_notification_handler.stop()
self._push_notification_handler = None
return await super().async_shutdown()

8
homeassistant/components/elmax/strings.json
View File

@ -67,5 +67,13 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"reauth_successful": "[%key:common::config_flow::abort::reauth_successful%]"
}
},
"exceptions": {
"alarm_operation_failed_generic": {
"message": "Failed to {operation} the alarm. An API error occurred."
},
"invalid_disarm_code": {
"message": "Invalid disarm code provided."
}
}
}

6
tests/components/elmax/snapshots/test_alarm_control_panel.ambr
View File

@ -46,7 +46,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
'state': 'disarmed',
})
# ---
# name: test_alarm_control_panels[alarm_control_panel.direct_panel_https_1_1_1_1_443_api_v2_area_2-entry]
@ -96,7 +96,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
'state': 'disarmed',
})
# ---
# name: test_alarm_control_panels[alarm_control_panel.direct_panel_https_1_1_1_1_443_api_v2_area_3-entry]
@ -146,6 +146,6 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
'state': 'disarmed',
})
# ---

4
tests/components/elmax/test_alarm_control_panel.py
View File

@ -1,9 +1,11 @@
"""Tests for the Elmax alarm control panels."""
from datetime import timedelta
from unittest.mock import patch
from syrupy import SnapshotAssertion
from homeassistant.components.elmax import POLLING_SECONDS
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@ -12,6 +14,8 @@ from . import init_integration
from tests.common import snapshot_platform
WAIT = timedelta(seconds=POLLING_SECONDS)
async def test_alarm_control_panels(
hass: HomeAssistant,