feat: device, notifications, outputs, status

custom_components/amc_alarm/alarm_control_panel.py

@@ -17,7 +17,7 @@
 from .amc_alarm_api.amc_proto import CentralDataSections
 from .amc_alarm_api.api import AmcStatesParser
 from .const import DOMAIN
-from .entity import AmcBaseEntity
+from .entity import AmcBaseEntity, device_info
 
 
 async def async_setup_entry(
@@ -42,6 +42,7 @@ def _area(_central_id, _amc_id):
         alarms.extend(
             AmcAreaGroup(
                 coordinator=coordinator,
+                device_info=device_info(states, central_id),
                 amc_entry=x,
                 attributes_fn=_group(central_id, x.Id),
             )
@@ -50,6 +51,7 @@ def _area(_central_id, _amc_id):
         alarms.extend(
             AmcAreaGroup(
                 coordinator=coordinator,
+                device_info=device_info(states, central_id),
                 amc_entry=x,
                 attributes_fn=_area(central_id, x.Id),
             )
@@ -58,13 +60,14 @@ def _area(_central_id, _amc_id):
         alarms.extend(
             AmcZone(
                 coordinator=coordinator,
+                device_info=device_info(states, central_id),
                 amc_entry=x,
                 attributes_fn=_zone(central_id, x.Id),
             )
             for x in states.zones(central_id).list
         )
 
-    async_add_entities(alarms, True)
+    async_add_entities(alarms, False)
 
 
 class AmcZone(AmcBaseEntity, AlarmControlPanelEntity):

custom_components/amc_alarm/amc_alarm_api/amc_proto.py

@@ -22,6 +22,7 @@ class AmcState(BaseModel):
     bit_opened: int
     bit_notReady: int
     remote: bool
+    progress: Optional[int]
 
 
 class AmcEntry(BaseModel):

custom_components/amc_alarm/amc_alarm_api/api.py

@@ -15,6 +15,8 @@
     CentralDataSections,
     AmcData,
     AmcEntry,
+    AmcNotificationEntry,
+    AmcNotification,
 )
 from .exceptions import AmcException, ConnectionFailed, AuthenticationFailed
 
@@ -33,13 +35,13 @@ class SimplifiedAmcApi:
     MAX_RETRY_DELAY = 600  # 10 min
 
     def __init__(
-            self,
-            login_email,
-            password,
-            central_id,
-            central_username,
-            central_password,
-            async_state_updated_callback=None,
+        self,
+        login_email,
+        password,
+        central_id,
+        central_username,
+        central_password,
+        async_state_updated_callback=None,
     ):
         self._raw_states: dict[str, AmcCentralResponse] = {}
 
@@ -72,7 +74,7 @@ async def connect(self):
                 continue
 
             if self._listen_task.done() and issubclass(
-                    self._listen_task.exception().__class__, AmcException
+                self._listen_task.exception().__class__, AmcException
             ):
                 raise self._listen_task.exception()  # Something known happened in the listener
 
@@ -98,7 +100,7 @@ async def _running(self) -> None:
             try:
                 _LOGGER.debug("Logging into %s" % self._ws_url)
                 async with session.ws_connect(
-                        self._ws_url, heartbeat=15, autoping=True
+                    self._ws_url, heartbeat=15, autoping=True
                 ) as ws_client:
                     self._ws_state = ConnectionState.CONNECTED
                     self._websocket = ws_client
@@ -125,7 +127,7 @@ async def _running(self) -> None:
                 _LOGGER.error("Unexpected response received from server : %s", error)
                 self._ws_state = ConnectionState.STOPPED
             except (aiohttp.ClientConnectionError, asyncio.TimeoutError) as error:
-                retry_delay = min(2 ** self._failed_attempts * 30, self.MAX_RETRY_DELAY)
+                retry_delay = min(2**self._failed_attempts * 30, self.MAX_RETRY_DELAY)
                 self._failed_attempts += 1
                 _LOGGER.error(
                     "Websocket connection failed, retrying in %ds: %s",
@@ -233,7 +235,7 @@ def __init__(self, states: dict[str, AmcCentralResponse]):
     def raw_states(self) -> dict[str, AmcCentralResponse]:
         return self._raw_states
 
-    def _get_section(self, central_id, section_index) -> AmcData:
+    def _get_section(self, central_id, section_index) -> AmcData | AmcNotification:
         central = self._raw_states[central_id]
         zones = next(x for x in central.data if x.index == section_index)
         return zones
@@ -265,7 +267,20 @@ def output(self, central_id: str, entry_id: int) -> AmcEntry:
     def system_statuses(self, central_id: str) -> AmcData:
         return self._get_section(central_id, CentralDataSections.SYSTEM_STATUS)
 
-    def system_status(self, central_id: str, entry_id: int) -> AmcEntry:
+    def system_status(self, central_id: str, entry_index: int) -> AmcEntry:
         return next(
-            x for x in self.system_statuses(central_id).list if x.Id == entry_id
+            x for x in self.system_statuses(central_id).list if x.index == entry_index
         )
+
+    def notifications(self, central_id: str) -> list[AmcNotificationEntry]:
+        return self._get_section(central_id, CentralDataSections.NOTIFICATIONS).list
+
+    def real_name(self, central_id: str) -> str:
+        return self._raw_states[central_id].realName
+
+    def status(self, central_id: str) -> str:
+        return self._raw_states[central_id].status
+
+    def model(self, central_id: str) -> str:
+        # Assuming from status
+        return self._raw_states[central_id].status.split(" ")[-1]

custom_components/amc_alarm/binary_sensor.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from homeassistant.components.binary_sensor import (
+    BinarySensorEntity,
+    BinarySensorDeviceClass,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+from .amc_alarm_api.amc_proto import CentralDataSections
+from .amc_alarm_api.api import AmcStatesParser
+from .const import DOMAIN
+from .entity import device_info, AmcBaseEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    coordinator: DataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+    states = AmcStatesParser(coordinator.data)
+    sensors: list[BinarySensorEntity] = []
+
+    def _system_status(_central_id, index):
+        return lambda raw_state: AmcStatesParser(raw_state).system_status(
+            _central_id, index
+        )
+
+    for central_id in states.raw_states():
+        sensors.extend(
+            AmcTamperSensor(
+                coordinator=coordinator,
+                device_info=device_info(states, central_id),
+                amc_entry=x,
+                attributes_fn=_system_status(central_id, x.index),
+            )
+            for x in states.system_statuses(central_id).list
+        )
+
+    async_add_entities(sensors, False)
+
+
+class AmcTamperSensor(AmcBaseEntity, BinarySensorEntity):
+    _amc_group_id = CentralDataSections.SYSTEM_STATUS
+    _attr_device_class = BinarySensorDeviceClass.TAMPER
+
+    def _handle_coordinator_update(self) -> None:
+        super()._handle_coordinator_update()
+        self._attr_is_on = self._amc_entry.states.anomaly == 1

custom_components/amc_alarm/const.py

@@ -5,7 +5,12 @@
 NAME = "AMC Alarm"
 
 # PLATFORMS SUPPORTED
-PLATFORMS = [Platform.ALARM_CONTROL_PANEL]
+PLATFORMS = [
+    Platform.ALARM_CONTROL_PANEL,
+    Platform.SENSOR,
+    Platform.BINARY_SENSOR,
+    Platform.SWITCH,
+]
 
 # DATA COORDINATOR ATTRIBUTES
 LAST_UPDATED = "last_updated"

custom_components/amc_alarm/entity.py

@@ -3,11 +3,23 @@
 from typing import Optional, Any, Callable
 
 from homeassistant.core import callback
+from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.update_coordinator import (
     CoordinatorEntity,
     DataUpdateCoordinator,
 )
 from .amc_alarm_api.amc_proto import AmcCentralResponse, AmcEntry
+from .amc_alarm_api.api import AmcStatesParser
+from .const import DOMAIN
+
+
+def device_info(states: AmcStatesParser, central_id: str) -> DeviceInfo:
+    return DeviceInfo(
+        identifiers={(DOMAIN, central_id)},
+        manufacturer="AMC Elettronica",
+        model=states.model(central_id),
+        name=states.real_name(central_id),
+    )
 
 
 class AmcBaseEntity(CoordinatorEntity):
@@ -16,6 +28,7 @@ class AmcBaseEntity(CoordinatorEntity):
     def __init__(
         self,
         coordinator: DataUpdateCoordinator,
+        device_info: DeviceInfo,
         amc_entry: AmcEntry,
         attributes_fn: Callable[[dict[str, AmcCentralResponse]], AmcEntry],
     ) -> None:
@@ -26,6 +39,7 @@ def __init__(
 
         self._attr_name = amc_entry.name
         self._attr_unique_id = str(amc_entry.Id)
+        self._attr_device_info = device_info
 
     @callback
     def _handle_coordinator_update(self) -> None:

custom_components/amc_alarm/sensor.py

@@ -0,0 +1,138 @@
+from __future__ import annotations
+
+from typing import Callable
+
+from homeassistant.const import PERCENTAGE, SIGNAL_STRENGTH_DECIBELS
+from homeassistant.components.sensor import SensorEntity, SensorDeviceClass
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.device_registry import DeviceInfo
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import (
+    DataUpdateCoordinator,
+    CoordinatorEntity,
+)
+from .amc_alarm_api.amc_proto import (
+    CentralDataSections,
+    AmcNotificationEntry,
+    SystemStatusDataSections,
+)
+from .amc_alarm_api.api import AmcStatesParser
+from .const import DOMAIN
+from .entity import device_info, AmcBaseEntity
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    coordinator: DataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id]
+    states = AmcStatesParser(coordinator.data)
+    sensors: list[SensorEntity] = []
+
+    def _notifications(_central_id):
+        return lambda raw_state: AmcStatesParser(raw_state).notifications(_central_id)
+
+    def _system_status(_central_id, index):
+        return lambda raw_state: AmcStatesParser(raw_state).system_status(
+            _central_id, index
+        )
+
+    for central_id in states.raw_states():
+        sensors.append(
+            AmcSignalSensor(
+                coordinator=coordinator,
+                device_info=device_info(states, central_id),
+                amc_entry=states.system_status(
+                    central_id, SystemStatusDataSections.GSM_SIGNAL
+                ),
+                attributes_fn=_system_status(
+                    central_id, SystemStatusDataSections.GSM_SIGNAL
+                ),
+            )
+        )
+        sensors.append(
+            AmcBatterySensor(
+                coordinator=coordinator,
+                device_info=device_info(states, central_id),
+                amc_entry=states.system_status(
+                    central_id, SystemStatusDataSections.BATTERY_STATUS
+                ),
+                attributes_fn=_system_status(
+                    central_id, SystemStatusDataSections.BATTERY_STATUS
+                ),
+            )
+        )
+
+        sensors.append(
+            AmcNotification(
+                coordinator=coordinator,
+                device_info=device_info(states, central_id),
+                amc_notifications=states.notifications(central_id),
+                attributes_fn=_notifications(central_id),
+            )
+        )
+
+    async_add_entities(sensors, False)
+
+
+class AmcBatterySensor(AmcBaseEntity, SensorEntity):
+    _attr_device_class = SensorDeviceClass.BATTERY
+    _attr_native_unit_of_measurement = PERCENTAGE
+
+    def _handle_coordinator_update(self) -> None:
+        super()._handle_coordinator_update()
+        self._attr_native_value = int(
+            self._amc_entry.states.progress / 15 * 100
+        )  # 15 is max
+
+
+class AmcSignalSensor(AmcBaseEntity, SensorEntity):
+    _attr_device_class = SensorDeviceClass.SIGNAL_STRENGTH
+    _attr_native_unit_of_measurement = SIGNAL_STRENGTH_DECIBELS
+
+    def _handle_coordinator_update(self) -> None:
+        super()._handle_coordinator_update()
+        self._attr_native_value = self._amc_entry.states.progress
+
+
+class AmcNotification(CoordinatorEntity, SensorEntity):
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: DataUpdateCoordinator,
+        device_info: DeviceInfo,
+        amc_notifications: list[AmcNotificationEntry],
+        attributes_fn: Callable,
+    ) -> None:
+        super().__init__(coordinator)
+
+        self._attributes_fn = attributes_fn
+        self._amc_notifications = amc_notifications
+
+        self._attr_name = "Notifications"
+        self._attr_unique_id = str(CentralDataSections.NOTIFICATIONS)
+        self._attr_device_info = device_info
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        self._amc_notifications: list[AmcNotificationEntry] = self._attributes_fn(
+            self.coordinator.data
+        )
+        if self._amc_notifications:
+            notification = self._amc_notifications[0]
+            self._attr_native_value = notification.name
+
+            self._attr_extra_state_attributes = {
+                x.serverDate: x.name for x in self._amc_notifications
+            }
+
+        super()._handle_coordinator_update()
+
+    async def async_added_to_hass(self) -> None:
+        """When entity is added to hass."""
+        self._handle_coordinator_update()
+        await super().async_added_to_hass()