jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-24 17:27:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add Connectivity sensor to SIA (#64305)

Browse Source 
* implemented connectivity sensor

* further cleanup off update code

* cleanup and tighter behaviour for attributes

* added seperate connectivity class to binary sensor

* callbacks and keys

* redid name and unique_id logic, non-breaking result

* using entry more in inits

* Fix import

* fix ping_interval in sia_entity_base

* added ping_interval default to next

* fixed next

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
This commit is contained in:
Eduard van Valkenburg 2022-02-15 15:53:38 +01:00 committed by GitHub
parent 430162fa5f
commit af4e37339a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 143 additions and 140 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
homeassistant/components/sia/alarm_control_panel.py
View File

@ -12,7 +12,6 @@ from homeassistant.components.alarm_control_panel import (
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
CONF_PORT,
STATE_ALARM_ARMED_AWAY,
STATE_ALARM_ARMED_CUSTOM_BYPASS,
STATE_ALARM_ARMED_NIGHT,
@ -24,16 +23,7 @@ from homeassistant.core import HomeAssistant, State
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from .const import (
CONF_ACCOUNT,
CONF_ACCOUNTS,
CONF_PING_INTERVAL,
CONF_ZONES,
KEY_ALARM,
PREVIOUS_STATE,
SIA_NAME_FORMAT,
SIA_UNIQUE_ID_FORMAT_ALARM,
)
from .const import CONF_ACCOUNT, CONF_ACCOUNTS, CONF_ZONES, KEY_ALARM, PREVIOUS_STATE
from .sia_entity_base import SIABaseEntity, SIAEntityDescription
_LOGGER = logging.getLogger(__name__)
@ -86,17 +76,7 @@ async def async_setup_entry(
"""Set up SIA alarm_control_panel(s) from a config entry."""
async_add_entities(
SIAAlarmControlPanel(
port=entry.data[CONF_PORT],
account=account_data[CONF_ACCOUNT],
zone=zone,
ping_interval=account_data[CONF_PING_INTERVAL],
entity_description=ENTITY_DESCRIPTION_ALARM,
unique_id=SIA_UNIQUE_ID_FORMAT_ALARM.format(
entry.entry_id, account_data[CONF_ACCOUNT], zone
),
name=SIA_NAME_FORMAT.format(
entry.data[CONF_PORT], account_data[CONF_ACCOUNT], zone, "alarm"
),
entry, account_data[CONF_ACCOUNT], zone, ENTITY_DESCRIPTION_ALARM
)
for account_data in entry.data[CONF_ACCOUNTS]
for zone in range(
@ -114,23 +94,17 @@ class SIAAlarmControlPanel(SIABaseEntity, AlarmControlPanelEntity):
def __init__(
self,
port: int,
entry: ConfigEntry,
account: str,
zone: int | None,
ping_interval: int,
zone: int,
entity_description: SIAAlarmControlPanelEntityDescription,
unique_id: str,
name: str,
) -> None:
"""Create SIAAlarmControlPanel object."""
super().__init__(
port,
entry,
account,
zone,
ping_interval,
entity_description,
unique_id,
name,
)
self._attr_state: StateType = None
@ -144,7 +118,10 @@ class SIAAlarmControlPanel(SIABaseEntity, AlarmControlPanelEntity):
self._attr_available = False
def update_state(self, sia_event: SIAEvent) -> bool:
"""Update the state of the alarm control panel."""
"""Update the state of the alarm control panel.
Return True if the event was relevant for this entity.
"""
new_state = self.entity_description.code_consequences.get(sia_event.code)
if new_state is None:
return False

97
homeassistant/components/sia/binary_sensor.py
View File

@ -13,23 +13,20 @@ from homeassistant.components.binary_sensor import (
BinarySensorEntityDescription,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PORT, STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant, State
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant, State, callback
from homeassistant.helpers.entity import EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
CONF_ACCOUNT,
CONF_ACCOUNTS,
CONF_PING_INTERVAL,
CONF_ZONES,
KEY_CONNECTIVITY,
KEY_MOISTURE,
KEY_POWER,
KEY_SMOKE,
SIA_HUB_ZONE,
SIA_NAME_FORMAT,
SIA_NAME_FORMAT_HUB,
SIA_UNIQUE_ID_FORMAT_BINARY,
)
from .sia_entity_base import SIABaseEntity, SIAEntityDescription
@ -78,72 +75,31 @@ ENTITY_DESCRIPTION_MOISTURE = SIABinarySensorEntityDescription(
entity_registry_enabled_default=False,
)
ENTITY_DESCRIPTION_CONNECTIVITY = SIABinarySensorEntityDescription(
key=KEY_CONNECTIVITY,
device_class=BinarySensorDeviceClass.CONNECTIVITY,
entity_category=EntityCategory.DIAGNOSTIC,
code_consequences={"RP": True},
)
def generate_binary_sensors(entry) -> Iterable[SIABinarySensor]:
def generate_binary_sensors(entry: ConfigEntry) -> Iterable[SIABinarySensor]:
"""Generate binary sensors.
For each Account there is one power sensor with zone == 0.
For each Zone in each Account there is one smoke and one moisture sensor.
"""
for account_data in entry.data[CONF_ACCOUNTS]:
yield SIABinarySensor(
port=entry.data[CONF_PORT],
account=account_data[CONF_ACCOUNT],
zone=SIA_HUB_ZONE,
ping_interval=account_data[CONF_PING_INTERVAL],
entity_description=ENTITY_DESCRIPTION_POWER,
unique_id=SIA_UNIQUE_ID_FORMAT_BINARY.format(
entry.entry_id,
account_data[CONF_ACCOUNT],
SIA_HUB_ZONE,
ENTITY_DESCRIPTION_POWER.device_class,
),
name=SIA_NAME_FORMAT_HUB.format(
entry.data[CONF_PORT],
account_data[CONF_ACCOUNT],
ENTITY_DESCRIPTION_POWER.device_class,
),
account = account_data[CONF_ACCOUNT]
zones = entry.options[CONF_ACCOUNTS][account][CONF_ZONES]
yield SIABinarySensorConnectivity(
entry, account, SIA_HUB_ZONE, ENTITY_DESCRIPTION_CONNECTIVITY
)
zones = entry.options[CONF_ACCOUNTS][account_data[CONF_ACCOUNT]][CONF_ZONES]
yield SIABinarySensor(entry, account, SIA_HUB_ZONE, ENTITY_DESCRIPTION_POWER)
for zone in range(1, zones + 1):
yield SIABinarySensor(
port=entry.data[CONF_PORT],
account=account_data[CONF_ACCOUNT],
zone=zone,
ping_interval=account_data[CONF_PING_INTERVAL],
entity_description=ENTITY_DESCRIPTION_SMOKE,
unique_id=SIA_UNIQUE_ID_FORMAT_BINARY.format(
entry.entry_id,
account_data[CONF_ACCOUNT],
zone,
ENTITY_DESCRIPTION_SMOKE.device_class,
),
name=SIA_NAME_FORMAT.format(
entry.data[CONF_PORT],
account_data[CONF_ACCOUNT],
zone,
ENTITY_DESCRIPTION_SMOKE.device_class,
),
)
yield SIABinarySensor(
port=entry.data[CONF_PORT],
account=account_data[CONF_ACCOUNT],
zone=zone,
ping_interval=account_data[CONF_PING_INTERVAL],
entity_description=ENTITY_DESCRIPTION_MOISTURE,
unique_id=SIA_UNIQUE_ID_FORMAT_BINARY.format(
entry.entry_id,
account_data[CONF_ACCOUNT],
zone,
ENTITY_DESCRIPTION_MOISTURE.device_class,
),
name=SIA_NAME_FORMAT.format(
entry.data[CONF_PORT],
account_data[CONF_ACCOUNT],
zone,
ENTITY_DESCRIPTION_MOISTURE.device_class,
),
)
yield SIABinarySensor(entry, account, zone, ENTITY_DESCRIPTION_SMOKE)
yield SIABinarySensor(entry, account, zone, ENTITY_DESCRIPTION_MOISTURE)
async def async_setup_entry(
@ -171,10 +127,23 @@ class SIABinarySensor(SIABaseEntity, BinarySensorEntity):
self._attr_available = False
def update_state(self, sia_event: SIAEvent) -> bool:
"""Update the state of the binary sensor."""
"""Update the state of the binary sensor.
Return True if the event was relevant for this entity.
"""
new_state = self.entity_description.code_consequences.get(sia_event.code)
if new_state is None:
return False
_LOGGER.debug("New state will be %s", new_state)
self._attr_is_on = bool(new_state)
return True
class SIABinarySensorConnectivity(SIABinarySensor):
"""Class for Connectivity Sensor."""
@callback
def async_post_interval_update(self, _) -> None:
"""Update state after a ping interval. Overwritten from sia entity base."""
self._attr_is_on = False
self.async_write_ha_state()

8
homeassistant/components/sia/const.py
View File

@ -24,18 +24,14 @@ CONF_IGNORE_TIMESTAMPS: Final = "ignore_timestamps"
CONF_PING_INTERVAL: Final = "ping_interval"
CONF_ZONES: Final = "zones"
SIA_NAME_FORMAT: Final = "{} - {} - zone {} - {}"
SIA_NAME_FORMAT_HUB: Final = "{} - {} - {}"
SIA_UNIQUE_ID_FORMAT_ALARM: Final = "{}_{}_{}"
SIA_UNIQUE_ID_FORMAT_BINARY: Final = "{}_{}_{}_{}"
SIA_UNIQUE_ID_FORMAT_HUB: Final = "{}_{}_{}"
SIA_HUB_ZONE: Final = 0
SIA_EVENT: Final = "sia_event_{}_{}"
KEY_ALARM: Final = "alarm_control_panel"
KEY_ALARM: Final = "alarm"
KEY_SMOKE: Final = "smoke"
KEY_MOISTURE: Final = "moisture"
KEY_POWER: Final = "power"
KEY_CONNECTIVITY: Final = "connectivity"
PREVIOUS_STATE: Final = "previous_state"
AVAILABILITY_EVENT_CODE: Final = "RP"

105
homeassistant/components/sia/sia_entity_base.py
View File

@ -7,6 +7,8 @@ import logging
from pysiaalarm import SIAEvent
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PORT
from homeassistant.core import CALLBACK_TYPE, State, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo, EntityDescription
@ -14,8 +16,20 @@ from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import StateType
from .const import AVAILABILITY_EVENT_CODE, DOMAIN, SIA_EVENT, SIA_HUB_ZONE
from .utils import get_attr_from_sia_event, get_unavailability_interval
from .const import (
AVAILABILITY_EVENT_CODE,
CONF_ACCOUNT,
CONF_ACCOUNTS,
CONF_PING_INTERVAL,
DOMAIN,
SIA_EVENT,
SIA_HUB_ZONE,
)
from .utils import (
get_attr_from_sia_event,
get_unavailability_interval,
get_unique_id_and_name,
)
_LOGGER = logging.getLogger(__name__)
@ -39,29 +53,32 @@ class SIABaseEntity(RestoreEntity):
def __init__(
self,
port: int,
entry: ConfigEntry,
account: str,
zone: int | None,
ping_interval: int,
zone: int,
entity_description: SIAEntityDescription,
unique_id: str,
name: str,
) -> None:
"""Create SIABaseEntity object."""
self.port = port
self.port = entry.data[CONF_PORT]
self.account = account
self.zone = zone
self.ping_interval = ping_interval
self.entity_description = entity_description
self._attr_unique_id = unique_id
self._attr_name = name
self.ping_interval: int = next(
acc[CONF_PING_INTERVAL]
for acc in entry.data[CONF_ACCOUNTS]
if acc[CONF_ACCOUNT] == account
)
self._attr_unique_id, self._attr_name = get_unique_id_and_name(
entry.entry_id, entry.data[CONF_PORT], account, zone, entity_description.key
)
self._attr_device_info = DeviceInfo(
name=name,
identifiers={(DOMAIN, unique_id)},
via_device=(DOMAIN, f"{port}_{account}"),
name=self._attr_name,
identifiers={(DOMAIN, self._attr_unique_id)},
via_device=(DOMAIN, f"{entry.data[CONF_PORT]}_{account}"),
)
self._cancel_availability_cb: CALLBACK_TYPE | None = None
self._post_interval_update_cb_canceller: CALLBACK_TYPE | None = None
self._attr_extra_state_attributes = {}
self._attr_should_poll = False
@ -83,7 +100,7 @@ class SIABaseEntity(RestoreEntity):
)
self.handle_last_state(await self.async_get_last_state())
if self._attr_available:
self.async_create_availability_cb()
self.async_create_post_interval_update_cb()
@abstractmethod
def handle_last_state(self, last_state: State | None) -> None:
@ -94,43 +111,57 @@ class SIABaseEntity(RestoreEntity):
Overridden from Entity.
"""
if self._cancel_availability_cb:
self._cancel_availability_cb()
self._cancel_post_interval_update_cb()
@callback
def async_handle_event(self, sia_event: SIAEvent) -> None:
"""Listen to dispatcher events for this port and account and update state and attributes."""
"""Listen to dispatcher events for this port and account and update state and attributes.
If the event is for either the zone or the 0 zone (hub zone), then handle it further.
If the event had a code that was relevant for the entity, then update the attributes.
If the event had a code that was relevant or it was a availability event then update the availability and schedule the next unavailability check.
"""
_LOGGER.debug("Received event: %s", sia_event)
if int(sia_event.ri) not in (self.zone, SIA_HUB_ZONE):
return
self._attr_extra_state_attributes.update(get_attr_from_sia_event(sia_event))
state_changed = self.update_state(sia_event)
if state_changed or sia_event.code == AVAILABILITY_EVENT_CODE:
self.async_reset_availability_cb()
relevant_event = self.update_state(sia_event)
if relevant_event:
self._attr_extra_state_attributes.update(get_attr_from_sia_event(sia_event))
if relevant_event or sia_event.code == AVAILABILITY_EVENT_CODE:
self._attr_available = True
self._cancel_post_interval_update_cb()
self.async_create_post_interval_update_cb()
self.async_write_ha_state()
@abstractmethod
def update_state(self, sia_event: SIAEvent) -> bool:
"""Do the entity specific state updates."""
"""Do the entity specific state updates.
Return True if the event was relevant for this entity.
"""
@callback
def async_reset_availability_cb(self) -> None:
"""Reset availability cb by cancelling the current and creating a new one."""
self._attr_available = True
if self._cancel_availability_cb:
self._cancel_availability_cb()
self.async_create_availability_cb()
def async_create_availability_cb(self) -> None:
"""Create a availability cb and return the callback."""
self._cancel_availability_cb = async_call_later(
def async_create_post_interval_update_cb(self) -> None:
"""Create a port interval update cb and store the callback."""
self._post_interval_update_cb_canceller = async_call_later(
self.hass,
get_unavailability_interval(self.ping_interval),
self.async_set_unavailable,
self.async_post_interval_update,
)
@callback
def async_set_unavailable(self, _) -> None:
"""Set unavailable."""
def async_post_interval_update(self, _) -> None:
"""Set unavailable after a ping interval."""
self._attr_available = False
self.async_write_ha_state()
@callback
def _cancel_post_interval_update_cb(self) -> None:
"""Cancel the callback."""
if self._post_interval_update_cb_canceller:
self._post_interval_update_cb_canceller()
self._post_interval_update_cb_canceller = None

32
homeassistant/components/sia/utils.py
View File

@ -8,11 +8,41 @@ from pysiaalarm import SIAEvent
from homeassistant.util.dt import utcnow
from .const import ATTR_CODE, ATTR_ID, ATTR_MESSAGE, ATTR_TIMESTAMP, ATTR_ZONE
from .const import (
ATTR_CODE,
ATTR_ID,
ATTR_MESSAGE,
ATTR_TIMESTAMP,
ATTR_ZONE,
KEY_ALARM,
SIA_HUB_ZONE,
)
PING_INTERVAL_MARGIN = 30
def get_unique_id_and_name(
entry_id: str,
port: int,
account: str,
zone: int,
entity_key: str,
) -> tuple[str, str]:
"""Return the unique_id and name for an entity."""
return (
(
f"{entry_id}_{account}_{zone}"
if entity_key == KEY_ALARM
else f"{entry_id}_{account}_{zone}_{entity_key}"
),
(
f"{port} - {account} - {entity_key}"
if zone == SIA_HUB_ZONE
else f"{port} - {account} - zone {zone} - {entity_key}"
),
)
def get_unavailability_interval(ping: int) -> float:
"""Return the interval to the next unavailability check."""
return timedelta(minutes=ping, seconds=PING_INTERVAL_MARGIN).total_seconds()