Add a server status sensor (#278)

* Update coordinator keeps server state based on stats polling

* Add frigate status sensor

* Formatting

* Cast to str

* Add test

* Update tests

* Test unknown state

* Simplify state logic

* Update the statuses to use consts

* Remove unknown status

* Add error test for sensor

* Adjust test

* Adjust test

* Wait time for failed stats to take effect

custom_components/frigate/__init__.py

@@ -44,6 +44,9 @@
     NAME,
     PLATFORMS,
     STARTUP_MESSAGE,
+    STATUS_ERROR,
+    STATUS_RUNNING,
+    STATUS_STARTING,
 )
 from .views import (
     JSMPEGProxyView,
@@ -292,13 +295,17 @@ class FrigateDataUpdateCoordinator(DataUpdateCoordinator):  # type: ignore[misc]
     def __init__(self, hass: HomeAssistant, client: FrigateApiClient):
         """Initialize."""
         self._api = client
+        self.server_status: str = STATUS_STARTING
         super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=SCAN_INTERVAL)
 
     async def _async_update_data(self) -> dict[str, Any]:
         """Update data via library."""
         try:
-            return await self._api.async_get_stats()
+            stats = await self._api.async_get_stats()
+            self.server_status = STATUS_RUNNING
+            return stats
         except FrigateApiClientError as exc:
+            self.server_status = STATUS_ERROR
             raise UpdateFailed from exc
 
 

custom_components/frigate/const.py

@@ -16,6 +16,7 @@
 ICON_MOTION_SENSOR = "hass:motion-sensor"
 ICON_OTHER = "mdi:shield-alert"
 ICON_PERSON = "mdi:shield-account"
+ICON_SERVER = "mdi:server"
 ICON_SPEEDOMETER = "mdi:speedometer"
 
 # Platforms
@@ -63,3 +64,8 @@
 # States
 STATE_DETECTED = "active"
 STATE_IDLE = "idle"
+
+# Statuses
+STATUS_ERROR = "error"
+STATUS_RUNNING = "running"
+STATUS_STARTING = "starting"

custom_components/frigate/sensor.py

@@ -32,6 +32,7 @@
     ICON_DOG,
     ICON_OTHER,
     ICON_PERSON,
+    ICON_SERVER,
     ICON_SPEEDOMETER,
     MS,
     NAME,
@@ -71,6 +72,7 @@ async def async_setup_entry(
             for cam_name, obj in get_cameras_zones_and_objects(frigate_config)
         ]
     )
+    entities.append(FrigateStatusSensor(coordinator, entry))
     async_add_entities(entities)
 
 
@@ -133,6 +135,53 @@ def icon(self) -> str:
         return ICON_SPEEDOMETER
 
 
+class FrigateStatusSensor(FrigateEntity, CoordinatorEntity):  # type: ignore[misc]
+    """Frigate Status Sensor class."""
+
+    _attr_entity_category = EntityCategory.DIAGNOSTIC
+
+    def __init__(
+        self, coordinator: FrigateDataUpdateCoordinator, config_entry: ConfigEntry
+    ) -> None:
+        """Construct a FrigateStatusSensor."""
+        FrigateEntity.__init__(self, config_entry)
+        CoordinatorEntity.__init__(self, coordinator)
+        self._attr_entity_registry_enabled_default = False
+
+    @property
+    def unique_id(self) -> str:
+        """Return a unique ID to use for this entity."""
+        return get_frigate_entity_unique_id(
+            self._config_entry.entry_id, "sensor_status", "frigate"
+        )
+
+    @property
+    def device_info(self) -> DeviceInfo:
+        """Get device information."""
+        return {
+            "identifiers": {get_frigate_device_identifier(self._config_entry)},
+            "name": NAME,
+            "model": self._get_model(),
+            "configuration_url": self._config_entry.data.get(CONF_URL),
+            "manufacturer": NAME,
+        }
+
+    @property
+    def name(self) -> str:
+        """Return the name of the sensor."""
+        return "Frigate Status"
+
+    @property
+    def state(self) -> str:
+        """Return the state of the sensor."""
+        return str(self.coordinator.server_status)
+
+    @property
+    def icon(self) -> str:
+        """Return the icon of the sensor."""
+        return ICON_SERVER
+
+
 class DetectorSpeedSensor(FrigateEntity, CoordinatorEntity):  # type: ignore[misc]
     """Frigate Detector Speed class."""
 

tests/__init__.py

@@ -49,6 +49,7 @@
 TEST_SENSOR_FRONT_DOOR_DETECTION_FPS_ENTITY_ID = "sensor.front_door_detection_fps"
 TEST_SENSOR_FRONT_DOOR_PROCESS_FPS_ENTITY_ID = "sensor.front_door_process_fps"
 TEST_SENSOR_FRONT_DOOR_SKIPPED_FPS_ENTITY_ID = "sensor.front_door_skipped_fps"
+TEST_SENSOR_FRIGATE_STATUS_ENTITY_ID = "sensor.frigate_status"
 TEST_UPDATE_FRIGATE_CONTAINER_ENTITY_ID = "update.frigate_server"
 
 TEST_SERVER_VERSION = "0.9.0-09a4d6d"

tests/test_sensor.py

@@ -13,6 +13,7 @@
 )
 
 from custom_components.frigate import SCAN_INTERVAL
+from custom_components.frigate.api import FrigateApiClientError
 from custom_components.frigate.const import (
     DOMAIN,
     FPS,
@@ -21,6 +22,7 @@
     ICON_DOG,
     ICON_OTHER,
     ICON_PERSON,
+    ICON_SERVER,
     ICON_SPEEDOMETER,
     MS,
     NAME,
@@ -35,6 +37,7 @@
     TEST_SENSOR_CPU1_INTFERENCE_SPEED_ENTITY_ID,
     TEST_SENSOR_CPU2_INTFERENCE_SPEED_ENTITY_ID,
     TEST_SENSOR_DETECTION_FPS_ENTITY_ID,
+    TEST_SENSOR_FRIGATE_STATUS_ENTITY_ID,
     TEST_SENSOR_FRONT_DOOR_ALL_ENTITY_ID,
     TEST_SENSOR_FRONT_DOOR_CAMERA_FPS_ENTITY_ID,
     TEST_SENSOR_FRONT_DOOR_DETECTION_FPS_ENTITY_ID,
@@ -208,6 +211,36 @@ async def test_fps_sensor(hass: HomeAssistant) -> None:
     assert entity_state.state == "unknown"
 
 
+async def test_status_sensor_success(hass: HomeAssistant) -> None:
+    """Test FrigateStatusSensor expected state."""
+
+    client = create_mock_frigate_client()
+    await setup_mock_frigate_config_entry(hass, client=client)
+    await enable_and_load_entity(hass, client, TEST_SENSOR_FRIGATE_STATUS_ENTITY_ID)
+
+    entity_state = hass.states.get(TEST_SENSOR_FRIGATE_STATUS_ENTITY_ID)
+    assert entity_state
+    assert entity_state.state == "running"
+    assert entity_state.attributes["icon"] == ICON_SERVER
+
+
+async def test_status_sensor_error(hass: HomeAssistant) -> None:
+    """Test FrigateStatusSensor unexpected state."""
+
+    client: AsyncMock = create_mock_frigate_client()
+    await setup_mock_frigate_config_entry(hass, client=client)
+    await enable_and_load_entity(hass, client, TEST_SENSOR_FRIGATE_STATUS_ENTITY_ID)
+
+    client.async_get_stats = AsyncMock(side_effect=FrigateApiClientError)
+    async_fire_time_changed(hass, dt_util.utcnow() + SCAN_INTERVAL)
+    await hass.async_block_till_done()
+
+    entity_state = hass.states.get(TEST_SENSOR_FRIGATE_STATUS_ENTITY_ID)
+    assert entity_state
+    assert entity_state.state == "error"
+    assert entity_state.attributes["icon"] == ICON_SERVER
+
+
 async def test_per_entry_device_info(hass: HomeAssistant) -> None:
     """Verify switch device information."""
     config_entry = await setup_mock_frigate_config_entry(hass)