Add Imou alarm control panel platform (#182481)

This commit is contained in:
Imou-OpenPlatform
2026-10-01 19:42:11 +01:00
committed by GitHub
parent f90ecfdd33
commit ed31ef13de
6 changed files with 430 additions and 0 deletions
@@ -0,0 +1,150 @@
"""Support for Imou alarm control panel entities."""
import logging
from typing import override
from pyimouapi.const import PARAM_MODE, PARAM_STATE, PARAM_SUPPORTED
from pyimouapi.ha_device import ImouHaDevice
from homeassistant.components.alarm_control_panel import (
AlarmControlPanelEntity,
AlarmControlPanelEntityDescription,
AlarmControlPanelEntityFeature,
AlarmControlPanelState,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import imou_device_identifier
from .coordinator import ImouConfigEntry, ImouDataUpdateCoordinator
from .entity import ImouEntity
from .helpers import async_wrap_imou_command
_LOGGER = logging.getLogger(__name__)
PARALLEL_UPDATES = 0
_ARM_MODES = frozenset({"home", "away"})
_MODE_TO_STATE = {
"home": AlarmControlPanelState.ARMED_HOME,
"away": AlarmControlPanelState.ARMED_AWAY,
"disarm": AlarmControlPanelState.DISARMED,
}
ALARM_PANEL_DESCRIPTION = AlarmControlPanelEntityDescription(
key=PARAM_MODE,
translation_key="alarm",
name=None,
)
def _device_has_alarm_panel(device: ImouHaDevice) -> bool:
"""Return whether the device exposes a usable arming panel."""
panel = device.alarm_control_panel
if panel is None:
return False
supported = panel.get(PARAM_SUPPORTED, [])
if not isinstance(supported, list):
return False
modes = {mode for mode in supported if isinstance(mode, str)}
if "disarm" not in modes:
return False
return bool(modes & _ARM_MODES)
def _iter_alarm_control_panels(
coordinator: ImouDataUpdateCoordinator,
) -> list[tuple[AlarmControlPanelEntityDescription, ImouHaDevice]]:
"""Return (description, device) pairs for devices with an arming panel."""
return [
(ALARM_PANEL_DESCRIPTION, device)
for device in coordinator.devices
if _device_has_alarm_panel(device)
]
def _supported_features(panel: dict) -> AlarmControlPanelEntityFeature:
"""Return arm features supported by the device panel."""
supported = panel.get(PARAM_SUPPORTED, [])
features = AlarmControlPanelEntityFeature(0)
if "home" in supported:
features |= AlarmControlPanelEntityFeature.ARM_HOME
if "away" in supported:
features |= AlarmControlPanelEntityFeature.ARM_AWAY
return features
async def async_setup_entry(
hass: HomeAssistant,
entry: ImouConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Imou alarm control panel entities."""
coordinator = entry.runtime_data
def _add_alarm_panels(new_devices: list[ImouHaDevice]) -> None:
device_keys = {imou_device_identifier(device) for device in new_devices}
async_add_entities(
ImouAlarmControlPanel(coordinator, description, device)
for description, device in _iter_alarm_control_panels(coordinator)
if imou_device_identifier(device) in device_keys
)
entry.async_on_unload(coordinator.register_new_device_callback(_add_alarm_panels))
_add_alarm_panels(coordinator.devices)
class ImouAlarmControlPanel(ImouEntity, AlarmControlPanelEntity):
"""Imou alarm control panel entity."""
entity_description: AlarmControlPanelEntityDescription
_attr_code_arm_required = False
def __init__(
self,
coordinator: ImouDataUpdateCoordinator,
description: AlarmControlPanelEntityDescription,
device: ImouHaDevice,
) -> None:
"""Initialize the Imou alarm control panel entity."""
super().__init__(coordinator, description, device)
panel = device.alarm_control_panel
assert panel is not None
self._attr_supported_features = _supported_features(panel)
@property
@override
def alarm_state(self) -> AlarmControlPanelState | None:
"""Return the current arming state."""
panel = self.device.alarm_control_panel
assert panel is not None
mode = panel.get(PARAM_STATE)
if not isinstance(mode, str):
return None
mapped = _MODE_TO_STATE.get(mode)
if mapped is None:
_LOGGER.debug("Unknown alarm mode %r for %s", mode, self._device_key)
return None
return mapped
@override
async def async_alarm_arm_home(self, code: str | None = None) -> None:
"""Arm the device in home mode."""
await self._async_set_mode("home")
@override
async def async_alarm_arm_away(self, code: str | None = None) -> None:
"""Arm the device in away mode."""
await self._async_set_mode("away")
@override
async def async_alarm_disarm(self, code: str | None = None) -> None:
"""Disarm the device."""
await self._async_set_mode("disarm")
@async_wrap_imou_command("alarm_arm_disarm_failed")
async def _async_set_mode(self, mode: str) -> None:
"""Call the vendor library to change arming mode."""
await self.coordinator.device_manager.async_set_alarm_mode(self.device, mode)
await self.coordinator.async_request_refresh()
+1
View File
@@ -45,6 +45,7 @@ PTZ_MOVE_DURATION_MS = 500
UPDATE_TIMEOUT = 300
PLATFORMS = [
Platform.ALARM_CONTROL_PANEL,
Platform.BINARY_SENSOR,
Platform.BUTTON,
Platform.CAMERA,
+5
View File
@@ -1,5 +1,10 @@
{
"entity": {
"alarm_control_panel": {
"alarm": {
"default": "mdi:shield-home"
}
},
"button": {
"ptz_down": {
"default": "mdi:arrow-down-bold"
@@ -141,6 +141,9 @@
}
},
"exceptions": {
"alarm_arm_disarm_failed": {
"message": "Imou rejected the arming change"
},
"get_image_failed": {
"message": "Could not get a snapshot from Imou"
},
@@ -0,0 +1,55 @@
# serializer version: 1
# name: test_alarm_control_panel_snapshot[alarm_mock_devices-platforms0][alarm_control_panel.gateway-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'alarm_control_panel',
'entity_category': None,
'entity_id': 'alarm_control_panel.gateway',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': None,
'platform': 'imou',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': <AlarmControlPanelEntityFeature: 3>,
'translation_key': 'alarm',
'unique_id': 'd1$mode',
'unit_of_measurement': None,
})
# ---
# name: test_alarm_control_panel_snapshot[alarm_mock_devices-platforms0][alarm_control_panel.gateway-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<AlarmControlPanelEntityStateAttribute.CHANGED_BY: 'changed_by'>: None,
<AlarmControlPanelEntityStateAttribute.CODE_ARM_REQUIRED: 'code_arm_required'>: False,
<AlarmControlPanelEntityStateAttribute.CODE_FORMAT: 'code_format'>: None,
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'Gateway',
<EntityStateAttribute.SUPPORTED_FEATURES: 'supported_features'>: <AlarmControlPanelEntityFeature: 3>,
}),
'context': <ANY>,
'entity_id': 'alarm_control_panel.gateway',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'disarmed',
})
# ---
@@ -0,0 +1,216 @@
"""Tests for Imou alarm control panel platform."""
from unittest.mock import MagicMock
from pyimouapi.const import PARAM_REF, PARAM_STATE, PARAM_SUPPORTED, PARAM_VALUE_TYPE
from pyimouapi.exceptions import ImouException, InvalidAppIdOrSecretException
from pyimouapi.ha_device import ImouHaDevice
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.alarm_control_panel import (
DOMAIN as ALARM_DOMAIN,
AlarmControlPanelState,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigEntryState
from homeassistant.const import (
ATTR_ENTITY_ID,
SERVICE_ALARM_ARM_AWAY,
SERVICE_ALARM_ARM_HOME,
SERVICE_ALARM_DISARM,
Platform,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from .const import create_online_device
from tests.common import MockConfigEntry, snapshot_platform
ALARM_PANEL = {
PARAM_REF: "arm_mode",
PARAM_STATE: "disarm",
PARAM_SUPPORTED: ["home", "away", "disarm"],
PARAM_VALUE_TYPE: "int",
}
def alarm_mock_devices() -> list[ImouHaDevice]:
"""Return a device list with an arming panel."""
device = create_online_device("d1", "Gateway", button_keys=())
device.alarm_control_panel = dict(ALARM_PANEL)
return [device]
def alarm_panel_without_modes_devices() -> list[ImouHaDevice]:
"""Return a device whose panel lists no supported modes."""
device = create_online_device("d1", "Gateway", button_keys=())
device.alarm_control_panel = {**ALARM_PANEL, PARAM_SUPPORTED: []}
return [device]
def alarm_panel_disarm_only_devices() -> list[ImouHaDevice]:
"""Return a device that supports disarm but no arm modes."""
device = create_online_device("d1", "Gateway", button_keys=())
device.alarm_control_panel = {**ALARM_PANEL, PARAM_SUPPORTED: ["disarm"]}
return [device]
def alarm_panel_home_without_disarm_devices() -> list[ImouHaDevice]:
"""Return a device that supports home but not disarm."""
device = create_online_device("d1", "Gateway", button_keys=())
device.alarm_control_panel = {**ALARM_PANEL, PARAM_SUPPORTED: ["home"]}
return [device]
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize("imou_mock_devices", [alarm_mock_devices], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_alarm_control_panel_snapshot(
hass: HomeAssistant,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
mock_config_entry: MockConfigEntry,
) -> None:
"""Snapshot alarm control panel entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize("imou_mock_devices", [alarm_mock_devices], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_arm_home_via_service(
hass: HomeAssistant,
mock_imou_ha_device_manager: MagicMock,
) -> None:
"""Arming home calls the vendor library."""
entity_id = hass.states.async_all(ALARM_DOMAIN)[0].entity_id
await hass.services.async_call(
ALARM_DOMAIN,
SERVICE_ALARM_ARM_HOME,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_imou_ha_device_manager.async_set_alarm_mode.assert_called_once()
assert mock_imou_ha_device_manager.async_set_alarm_mode.call_args[0][1] == "home"
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize("imou_mock_devices", [alarm_mock_devices], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_arm_away_via_service(
hass: HomeAssistant,
mock_imou_ha_device_manager: MagicMock,
) -> None:
"""Arming away calls the vendor library."""
entity_id = hass.states.async_all(ALARM_DOMAIN)[0].entity_id
await hass.services.async_call(
ALARM_DOMAIN,
SERVICE_ALARM_ARM_AWAY,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_imou_ha_device_manager.async_set_alarm_mode.assert_called_once()
assert mock_imou_ha_device_manager.async_set_alarm_mode.call_args[0][1] == "away"
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize("imou_mock_devices", [alarm_mock_devices], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_disarm_via_service(
hass: HomeAssistant,
mock_imou_ha_device_manager: MagicMock,
) -> None:
"""Disarm calls the vendor library."""
entity_id = hass.states.async_all(ALARM_DOMAIN)[0].entity_id
await hass.services.async_call(
ALARM_DOMAIN,
SERVICE_ALARM_DISARM,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
mock_imou_ha_device_manager.async_set_alarm_mode.assert_called_once()
assert mock_imou_ha_device_manager.async_set_alarm_mode.call_args[0][1] == "disarm"
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize("imou_mock_devices", [alarm_mock_devices], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_arm_home_propagates_api_error(
hass: HomeAssistant,
mock_imou_ha_device_manager: MagicMock,
) -> None:
"""An API error while arming raises a translated error."""
mock_imou_ha_device_manager.async_set_alarm_mode.side_effect = ImouException("fail")
entity_id = hass.states.async_all(ALARM_DOMAIN)[0].entity_id
with pytest.raises(HomeAssistantError, match="Imou rejected the arming change"):
await hass.services.async_call(
ALARM_DOMAIN,
SERVICE_ALARM_ARM_HOME,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize("imou_mock_devices", [alarm_mock_devices], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_arm_home_invalid_auth_starts_reauth(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_imou_ha_device_manager: MagicMock,
) -> None:
"""Rejected credentials while arming start reauthentication."""
mock_imou_ha_device_manager.async_set_alarm_mode.side_effect = (
InvalidAppIdOrSecretException("fail")
)
entity_id = hass.states.async_all(ALARM_DOMAIN)[0].entity_id
with pytest.raises(
HomeAssistantError, match="Imou rejected the App ID and App secret"
):
await hass.services.async_call(
ALARM_DOMAIN,
SERVICE_ALARM_ARM_HOME,
{ATTR_ENTITY_ID: entity_id},
blocking=True,
)
await hass.async_block_till_done()
assert mock_config_entry.state is ConfigEntryState.LOADED
assert any(mock_config_entry.async_get_active_flows(hass, {SOURCE_REAUTH}))
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize("imou_mock_devices", [alarm_mock_devices], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_alarm_state_from_device(
hass: HomeAssistant,
) -> None:
"""Alarm state reflects the device panel mode."""
entity_id = hass.states.async_all(ALARM_DOMAIN)[0].entity_id
assert hass.states.get(entity_id).state == AlarmControlPanelState.DISARMED
@pytest.mark.parametrize("platforms", [[Platform.ALARM_CONTROL_PANEL]], indirect=True)
@pytest.mark.parametrize(
"imou_mock_devices",
[
alarm_panel_without_modes_devices,
alarm_panel_disarm_only_devices,
alarm_panel_home_without_disarm_devices,
],
indirect=True,
)
@pytest.mark.usefixtures("init_integration")
async def test_skips_incomplete_alarm_panel(hass: HomeAssistant) -> None:
"""Devices without disarm and an arm mode do not get an arming entity."""
assert not hass.states.async_all(ALARM_DOMAIN)