Add Risco system binary sensors (#114062)
* Add Risco system binary sensors * Remove leading underscore * Address code review commments
This commit is contained in:
parent
d75315f225
commit
c661622332
7 changed files with 221 additions and 5 deletions
|
@ -17,7 +17,7 @@ from pyrisco import (
|
|||
)
|
||||
from pyrisco.cloud.alarm import Alarm
|
||||
from pyrisco.cloud.event import Event
|
||||
from pyrisco.common import Partition, Zone
|
||||
from pyrisco.common import Partition, System, Zone
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import (
|
||||
|
@ -42,6 +42,7 @@ from .const import (
|
|||
DEFAULT_SCAN_INTERVAL,
|
||||
DOMAIN,
|
||||
EVENTS_COORDINATOR,
|
||||
SYSTEM_UPDATE_SIGNAL,
|
||||
TYPE_LOCAL,
|
||||
)
|
||||
|
||||
|
@ -122,6 +123,12 @@ async def _async_setup_local_entry(hass: HomeAssistant, entry: ConfigEntry) -> b
|
|||
|
||||
entry.async_on_unload(risco.add_partition_handler(_partition))
|
||||
|
||||
async def _system(system: System) -> None:
|
||||
_LOGGER.debug("Risco system update")
|
||||
async_dispatcher_send(hass, SYSTEM_UPDATE_SIGNAL)
|
||||
|
||||
entry.async_on_unload(risco.add_system_handler(_system))
|
||||
|
||||
entry.async_on_unload(entry.add_update_listener(_update_listener))
|
||||
|
||||
hass.data.setdefault(DOMAIN, {})
|
||||
|
|
|
@ -3,23 +3,71 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from itertools import chain
|
||||
from typing import Any
|
||||
|
||||
from pyrisco.cloud.zone import Zone as CloudZone
|
||||
from pyrisco.common import System
|
||||
from pyrisco.local.zone import Zone as LocalZone
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
BinarySensorDeviceClass,
|
||||
BinarySensorEntity,
|
||||
BinarySensorEntityDescription,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.dispatcher import async_dispatcher_connect
|
||||
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
||||
|
||||
from . import LocalData, RiscoDataUpdateCoordinator, is_local
|
||||
from .const import DATA_COORDINATOR, DOMAIN
|
||||
from .const import DATA_COORDINATOR, DOMAIN, SYSTEM_UPDATE_SIGNAL
|
||||
from .entity import RiscoCloudZoneEntity, RiscoLocalZoneEntity
|
||||
|
||||
SYSTEM_ENTITY_DESCRIPTIONS = [
|
||||
BinarySensorEntityDescription(
|
||||
key="low_battery_trouble",
|
||||
translation_key="low_battery_trouble",
|
||||
device_class=BinarySensorDeviceClass.BATTERY,
|
||||
),
|
||||
BinarySensorEntityDescription(
|
||||
key="ac_trouble",
|
||||
translation_key="ac_trouble",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
),
|
||||
BinarySensorEntityDescription(
|
||||
key="monitoring_station_1_trouble",
|
||||
translation_key="monitoring_station_1_trouble",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
),
|
||||
BinarySensorEntityDescription(
|
||||
key="monitoring_station_2_trouble",
|
||||
translation_key="monitoring_station_2_trouble",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
),
|
||||
BinarySensorEntityDescription(
|
||||
key="monitoring_station_3_trouble",
|
||||
translation_key="monitoring_station_3_trouble",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
),
|
||||
BinarySensorEntityDescription(
|
||||
key="phone_line_trouble",
|
||||
translation_key="phone_line_trouble",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
),
|
||||
BinarySensorEntityDescription(
|
||||
key="clock_trouble",
|
||||
translation_key="clock_trouble",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
),
|
||||
BinarySensorEntityDescription(
|
||||
key="box_tamper",
|
||||
translation_key="box_tamper",
|
||||
device_class=BinarySensorDeviceClass.TAMPER,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
|
@ -29,7 +77,7 @@ async def async_setup_entry(
|
|||
"""Set up the Risco alarm control panel."""
|
||||
if is_local(config_entry):
|
||||
local_data: LocalData = hass.data[DOMAIN][config_entry.entry_id]
|
||||
async_add_entities(
|
||||
zone_entities = (
|
||||
entity
|
||||
for zone_id, zone in local_data.system.zones.items()
|
||||
for entity in (
|
||||
|
@ -38,6 +86,15 @@ async def async_setup_entry(
|
|||
RiscoLocalArmedBinarySensor(local_data.system.id, zone_id, zone),
|
||||
)
|
||||
)
|
||||
|
||||
system_entities = (
|
||||
RiscoSystemBinarySensor(
|
||||
local_data.system.id, local_data.system.system, entity_description
|
||||
)
|
||||
for entity_description in SYSTEM_ENTITY_DESCRIPTIONS
|
||||
)
|
||||
|
||||
async_add_entities(chain(system_entities, zone_entities))
|
||||
else:
|
||||
coordinator: RiscoDataUpdateCoordinator = hass.data[DOMAIN][
|
||||
config_entry.entry_id
|
||||
|
@ -128,3 +185,40 @@ class RiscoLocalArmedBinarySensor(RiscoLocalZoneEntity, BinarySensorEntity):
|
|||
def is_on(self) -> bool | None:
|
||||
"""Return true if sensor is on."""
|
||||
return self._zone.armed
|
||||
|
||||
|
||||
class RiscoSystemBinarySensor(BinarySensorEntity):
|
||||
"""Risco local system binary sensor class."""
|
||||
|
||||
_attr_should_poll = False
|
||||
_attr_has_entity_name = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
system_id: str,
|
||||
system: System,
|
||||
entity_description: BinarySensorEntityDescription,
|
||||
) -> None:
|
||||
"""Init the sensor."""
|
||||
self._system = system
|
||||
self._property = entity_description.key
|
||||
self._attr_unique_id = f"{system_id}_{self._property}"
|
||||
self._attr_device_info = DeviceInfo(
|
||||
identifiers={(DOMAIN, system_id)},
|
||||
manufacturer="Risco",
|
||||
name=system.name,
|
||||
)
|
||||
self.entity_description = entity_description
|
||||
|
||||
async def async_added_to_hass(self) -> None:
|
||||
"""Subscribe to updates."""
|
||||
self.async_on_remove(
|
||||
async_dispatcher_connect(
|
||||
self.hass, SYSTEM_UPDATE_SIGNAL, self.async_write_ha_state
|
||||
)
|
||||
)
|
||||
|
||||
@property
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return true if sensor is on."""
|
||||
return getattr(self._system, self._property)
|
||||
|
|
|
@ -19,6 +19,7 @@ TYPE_LOCAL = "local"
|
|||
|
||||
MAX_COMMUNICATION_DELAY = 3
|
||||
|
||||
SYSTEM_UPDATE_SIGNAL = "risco_system_update"
|
||||
CONF_CODE_ARM_REQUIRED = "code_arm_required"
|
||||
CONF_CODE_DISARM_REQUIRED = "code_disarm_required"
|
||||
CONF_RISCO_STATES_TO_HA = "risco_states_to_ha"
|
||||
|
|
|
@ -72,6 +72,30 @@
|
|||
},
|
||||
"armed": {
|
||||
"name": "Armed"
|
||||
},
|
||||
"low_battery_trouble": {
|
||||
"name": "Low battery trouble"
|
||||
},
|
||||
"ac_trouble": {
|
||||
"name": "A/C trouble"
|
||||
},
|
||||
"monitoring_station_1_trouble": {
|
||||
"name": "Monitoring station 1 trouble"
|
||||
},
|
||||
"monitoring_station_2_trouble": {
|
||||
"name": "Monitoring station 2 trouble"
|
||||
},
|
||||
"monitoring_station_3_trouble": {
|
||||
"name": "Monitoring station 3 trouble"
|
||||
},
|
||||
"phone_line_trouble": {
|
||||
"name": "Phone line trouble"
|
||||
},
|
||||
"clock_trouble": {
|
||||
"name": "Clock trouble"
|
||||
},
|
||||
"box_tamper": {
|
||||
"name": "Box tamper"
|
||||
}
|
||||
},
|
||||
"switch": {
|
||||
|
|
|
@ -14,7 +14,7 @@ from homeassistant.const import (
|
|||
CONF_USERNAME,
|
||||
)
|
||||
|
||||
from .util import TEST_SITE_NAME, TEST_SITE_UUID, zone_mock
|
||||
from .util import TEST_SITE_NAME, TEST_SITE_UUID, system_mock, zone_mock
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
@ -63,6 +63,7 @@ def two_zone_cloud():
|
|||
def two_zone_local():
|
||||
"""Fixture to mock alarm with two zones."""
|
||||
zone_mocks = {0: zone_mock(), 1: zone_mock()}
|
||||
system = system_mock()
|
||||
with patch.object(
|
||||
zone_mocks[0], "id", new_callable=PropertyMock(return_value=0)
|
||||
), patch.object(
|
||||
|
@ -83,12 +84,17 @@ def two_zone_local():
|
|||
zone_mocks[1], "bypassed", new_callable=PropertyMock(return_value=False)
|
||||
), patch.object(
|
||||
zone_mocks[1], "armed", new_callable=PropertyMock(return_value=False)
|
||||
), patch.object(
|
||||
system, "name", new_callable=PropertyMock(return_value=TEST_SITE_NAME)
|
||||
), patch(
|
||||
"homeassistant.components.risco.RiscoLocal.partitions",
|
||||
new_callable=PropertyMock(return_value={}),
|
||||
), patch(
|
||||
"homeassistant.components.risco.RiscoLocal.zones",
|
||||
new_callable=PropertyMock(return_value=zone_mocks),
|
||||
), patch(
|
||||
"homeassistant.components.risco.RiscoLocal.system",
|
||||
new_callable=PropertyMock(return_value=system),
|
||||
):
|
||||
yield zone_mocks
|
||||
|
||||
|
|
|
@ -11,7 +11,7 @@ from homeassistant.core import HomeAssistant
|
|||
from homeassistant.helpers import device_registry as dr, entity_registry as er
|
||||
from homeassistant.helpers.entity_component import async_update_entity
|
||||
|
||||
from .util import TEST_SITE_UUID
|
||||
from .util import TEST_SITE_NAME, TEST_SITE_UUID, system_mock
|
||||
|
||||
FIRST_ENTITY_ID = "binary_sensor.zone_0"
|
||||
SECOND_ENTITY_ID = "binary_sensor.zone_1"
|
||||
|
@ -116,6 +116,10 @@ async def test_local_setup(
|
|||
assert device is not None
|
||||
assert device.manufacturer == "Risco"
|
||||
|
||||
device = registry.async_get_device(identifiers={(DOMAIN, TEST_SITE_UUID)})
|
||||
assert device is not None
|
||||
assert device.manufacturer == "Risco"
|
||||
|
||||
|
||||
async def _check_local_state(
|
||||
hass, zones, property, value, entity_id, zone_id, callback
|
||||
|
@ -204,3 +208,68 @@ async def test_armed_local_states(
|
|||
await _check_local_state(
|
||||
hass, two_zone_local, "armed", False, SECOND_ARMED_ENTITY_ID, 1, callback
|
||||
)
|
||||
|
||||
|
||||
async def _check_system_state(hass, system, property, value, callback):
|
||||
with patch.object(
|
||||
system,
|
||||
property,
|
||||
new_callable=PropertyMock(return_value=value),
|
||||
):
|
||||
await callback(system)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
expected_value = STATE_ON if value else STATE_OFF
|
||||
if property == "ac_trouble":
|
||||
property = "a_c_trouble"
|
||||
entity_id = f"binary_sensor.test_site_name_{property}"
|
||||
assert hass.states.get(entity_id).state == expected_value
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_system_handler():
|
||||
"""Create a mock for add_system_handler."""
|
||||
with patch("homeassistant.components.risco.RiscoLocal.add_system_handler") as mock:
|
||||
yield mock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def system_only_local():
|
||||
"""Fixture to mock a system with no zones or partitions."""
|
||||
system = system_mock()
|
||||
with patch.object(
|
||||
system, "name", new_callable=PropertyMock(return_value=TEST_SITE_NAME)
|
||||
), patch(
|
||||
"homeassistant.components.risco.RiscoLocal.zones",
|
||||
new_callable=PropertyMock(return_value={}),
|
||||
), patch(
|
||||
"homeassistant.components.risco.RiscoLocal.partitions",
|
||||
new_callable=PropertyMock(return_value={}),
|
||||
), patch(
|
||||
"homeassistant.components.risco.RiscoLocal.system",
|
||||
new_callable=PropertyMock(return_value=system),
|
||||
):
|
||||
yield system
|
||||
|
||||
|
||||
async def test_system_states(
|
||||
hass: HomeAssistant, system_only_local, mock_system_handler, setup_risco_local
|
||||
) -> None:
|
||||
"""Test the various zone states."""
|
||||
callback = mock_system_handler.call_args.args[0]
|
||||
|
||||
assert callback is not None
|
||||
|
||||
properties = [
|
||||
"low_battery_trouble",
|
||||
"ac_trouble",
|
||||
"monitoring_station_1_trouble",
|
||||
"monitoring_station_2_trouble",
|
||||
"monitoring_station_3_trouble",
|
||||
"phone_line_trouble",
|
||||
"clock_trouble",
|
||||
"box_tamper",
|
||||
]
|
||||
for property in properties:
|
||||
await _check_system_state(hass, system_only_local, property, True, callback)
|
||||
await _check_system_state(hass, system_only_local, property, False, callback)
|
||||
|
|
|
@ -11,3 +11,18 @@ def zone_mock():
|
|||
return MagicMock(
|
||||
triggered=False, bypassed=False, bypass=AsyncMock(return_value=True)
|
||||
)
|
||||
|
||||
|
||||
def system_mock():
|
||||
"""Return a mocked system."""
|
||||
return MagicMock(
|
||||
low_battery_trouble=False,
|
||||
ac_trouble=False,
|
||||
monitoring_station_1_trouble=False,
|
||||
monitoring_station_2_trouble=False,
|
||||
monitoring_station_3_trouble=False,
|
||||
phone_line_trouble=False,
|
||||
clock_trouble=False,
|
||||
box_tamper=False,
|
||||
programming_mode=False,
|
||||
)
|
||||
|
|
Loading…
Add table
Reference in a new issue