From f333afe08d83c083f8d504110d8627b89c6c8712 Mon Sep 17 00:00:00 2001 From: Robert Lippmann Date: Wed, 11 Oct 2023 05:49:52 -0400 Subject: [PATCH] move common functions to utils.py --- custom_components/adtpulse/__init__.py | 57 ++-------------- .../adtpulse/alarm_control_panel.py | 6 +- custom_components/adtpulse/binary_sensor.py | 6 +- custom_components/adtpulse/utils.py | 65 +++++++++++++++++++ 4 files changed, 75 insertions(+), 59 deletions(-) create mode 100644 custom_components/adtpulse/utils.py diff --git a/custom_components/adtpulse/__init__.py b/custom_components/adtpulse/__init__.py index 86a70fe..81f305c 100644 --- a/custom_components/adtpulse/__init__.py +++ b/custom_components/adtpulse/__init__.py @@ -18,22 +18,16 @@ CONF_USERNAME, EVENT_HOMEASSISTANT_STOP, ) -from homeassistant.core import HomeAssistant, callback +from homeassistant.core import HomeAssistant from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady -from homeassistant.helpers import entity_registry from homeassistant.helpers.config_entry_flow import FlowResult from homeassistant.helpers.typing import ConfigType -from homeassistant.util import slugify from pyadtpulse import PyADTPulse from pyadtpulse.const import ( ADT_DEFAULT_KEEPALIVE_INTERVAL, ADT_DEFAULT_POLL_INTERVAL, ADT_DEFAULT_RELOGIN_INTERVAL, - STATE_OK, - STATE_ONLINE, ) -from pyadtpulse.site import ADTPulseSite -from pyadtpulse.zones import ADTPulseZoneData from .const import ( ADTPULSE_DOMAIN, @@ -49,52 +43,9 @@ SUPPORTED_PLATFORMS = ["alarm_control_panel", "binary_sensor"] -def get_gateway_unique_id(site: ADTPulseSite) -> str: - """Get unique ID for gateway.""" - return f"adt_pulse_gateway_{site.id}" - - -def get_alarm_unique_id(site: ADTPulseSite) -> str: - """Get unique ID for alarm.""" - return f"adt_pulse_alarm_{site.id}" - - -def zone_open(zone: ADTPulseZoneData) -> bool: - """Determine if a zone is opened.""" - return not zone.state == STATE_OK - - -def zone_trouble(zone: ADTPulseZoneData) -> bool: - """Determine if a zone is in trouble state.""" - return not zone.status == STATE_ONLINE - - -@callback -def migrate_entity_name( - hass: HomeAssistant, site: ADTPulseSite, platform_name: str, entity_uid: str -) -> None: - """Migrate old entity names.""" - registry = entity_registry.async_get(hass) - if registry is None: - return - # this seems backwards - entity_id = registry.async_get_entity_id( - platform_name, - ADTPULSE_DOMAIN, - entity_uid, - ) - if entity_id is not None: - # change has_entity_name to True and set name to None for devices - registry.async_update_entity(entity_id, has_entity_name=True, name=None) - # rename site name to site id for entities which have site name - slugified_site_name = slugify(site.name) - if slugified_site_name in entity_id: - registry.async_update_entity( - entity_id, new_entity_id=entity_id.replace(slugified_site_name, site.id) - ) - - -async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: +async def async_setup( + hass: HomeAssistant, config: ConfigType # pylint: disable=unused-argument +) -> bool: """Start up the ADT Pulse HA integration. Args: diff --git a/custom_components/adtpulse/alarm_control_panel.py b/custom_components/adtpulse/alarm_control_panel.py index d0c1664..29d77d4 100644 --- a/custom_components/adtpulse/alarm_control_panel.py +++ b/custom_components/adtpulse/alarm_control_panel.py @@ -34,15 +34,15 @@ ) from pyadtpulse.site import ADTPulseSite -from . import ( +from .const import ADTPULSE_DATA_ATTRIBUTION, ADTPULSE_DOMAIN +from .coordinator import ADTPulseDataUpdateCoordinator +from .utils import ( get_alarm_unique_id, get_gateway_unique_id, migrate_entity_name, zone_open, zone_trouble, ) -from .const import ADTPULSE_DATA_ATTRIBUTION, ADTPULSE_DOMAIN -from .coordinator import ADTPulseDataUpdateCoordinator LOG = getLogger(__name__) diff --git a/custom_components/adtpulse/binary_sensor.py b/custom_components/adtpulse/binary_sensor.py index 1d88478..a96f410 100644 --- a/custom_components/adtpulse/binary_sensor.py +++ b/custom_components/adtpulse/binary_sensor.py @@ -25,15 +25,15 @@ from pyadtpulse.site import ADTPulseSite from pyadtpulse.zones import ADTPulseZoneData -from . import ( +from .const import ADTPULSE_DATA_ATTRIBUTION, ADTPULSE_DOMAIN +from .coordinator import ADTPulseDataUpdateCoordinator +from .utils import ( get_alarm_unique_id, get_gateway_unique_id, migrate_entity_name, zone_open, zone_trouble, ) -from .const import ADTPULSE_DATA_ATTRIBUTION, ADTPULSE_DOMAIN -from .coordinator import ADTPulseDataUpdateCoordinator LOG = getLogger(__name__) diff --git a/custom_components/adtpulse/utils.py b/custom_components/adtpulse/utils.py new file mode 100644 index 0000000..513f8e8 --- /dev/null +++ b/custom_components/adtpulse/utils.py @@ -0,0 +1,65 @@ +"""ADT Pulse utility functions.""" +from __future__ import annotations + +from homeassistant.core import HomeAssistant +from homeassistant.helpers import entity_registry +from pyadtpulse.const import STATE_OK, STATE_ONLINE +from pyadtpulse.site import ADTPulseSite +from pyadtpulse.zones import ADTPulseZoneData + +from .const import ADTPULSE_DOMAIN + + +def migrate_entity_name( + hass: HomeAssistant, site: ADTPulseSite, platform_name: str, entity_uid: str +) -> None: + """Migrate old entity names.""" + registry = entity_registry.async_get(hass) + if registry is None: + return + # this seems backwards + entity_id = registry.async_get_entity_id( + platform_name, + ADTPULSE_DOMAIN, + entity_uid, + ) + if entity_id is not None: + # change has_entity_name to True and set name to None for devices + registry.async_update_entity(entity_id, has_entity_name=True, name=None) + # rename site name to site id for entities which have site name + slugified_site_name = slugify(site.name) + if slugified_site_name in entity_id: + registry.async_update_entity( + entity_id, new_entity_id=entity_id.replace(slugified_site_name, site.id) + ) + + +def get_gateway_unique_id(site: ADTPulseSite) -> str: + """Get entity unique id for the gateway.""" + return f"adt_pulse_gateway_{site.id}" + + +def get_alarm_unique_id(site: ADTPulseSite) -> str: + """Get entity unique ID for alarm.""" + return f"adt_pulse_alarm_{site.id}" + + +def zone_open(zone: ADTPulseZoneData) -> bool: + """Determine if a zone is opened.""" + return not zone.state == STATE_OK + + +def zone_trouble(zone: ADTPulseZoneData) -> bool: + """Determine if a zone is in trouble state.""" + return not zone.status == STATE_ONLINE + + +def system_can_be_armed(site: ADTPulseSite) -> bool: + """Determine is the system is able to be armed without being forced.""" + zones = site.zones_as_dict + if zones is None: + return False + for zone in zones: + if zone_open(zone) or zone_trouble(zone): + return False + return True