Enhancement: Sensors for audio detections

custom_components/frigate/__init__.py

@@ -115,6 +115,17 @@ def get_cameras_and_objects(
     return camera_objects
 
 
+def get_cameras_and_audio(config: dict[str, Any]) -> set[tuple[str, str]]:
+    """Get cameras and audio tuples."""
+    camera_audio = set()
+    for cam_name, cam_config in config["cameras"].items():
+        if cam_config.get("audio", {}).get("enabled_in_config", False):
+            for audio in cam_config.get("audio", {}).get("listen", []):
+                camera_audio.add((cam_name, audio))
+
+    return camera_audio
+
+
 def get_cameras_zones_and_objects(config: dict[str, Any]) -> set[tuple[str, str]]:
     """Get cameras/zones and tracking object tuples."""
     camera_objects = get_cameras_and_objects(config)

custom_components/frigate/binary_sensor.py

@@ -17,6 +17,7 @@
     FrigateMQTTEntity,
     ReceiveMessage,
     get_cameras,
+    get_cameras_and_audio,
     get_cameras_zones_and_objects,
     get_friendly_name,
     get_frigate_device_identifier,
@@ -45,6 +46,14 @@ async def async_setup_entry(
         ]
     )
 
+    # add audio sensors for cameras
+    entities.extend(
+        [
+            FrigateAudioSensor(entry, frigate_config, cam_name, audio)
+            for cam_name, audio in get_cameras_and_audio(frigate_config)
+        ]
+    )
+
     # add generic motion sensors for cameras
     entities.extend(
         [
@@ -141,6 +150,87 @@ def icon(self) -> str:
         return get_dynamic_icon_from_type(self._obj_name, self._is_on)
 
 
+class FrigateAudioSensor(FrigateMQTTEntity, BinarySensorEntity):  # type: ignore[misc]
+    """Frigate Audio Sensor class."""
+
+    def __init__(
+        self,
+        config_entry: ConfigEntry,
+        frigate_config: dict[str, Any],
+        cam_name: str,
+        audio_name: str,
+    ) -> None:
+        """Construct a new FrigateAudioSensor."""
+        self._cam_name = cam_name
+        self._audio_name = audio_name
+        self._is_on = False
+        self._frigate_config = frigate_config
+
+        super().__init__(
+            config_entry,
+            frigate_config,
+            {
+                "state_topic": {
+                    "msg_callback": self._state_message_received,
+                    "qos": 0,
+                    "topic": (
+                        f"{self._frigate_config['mqtt']['topic_prefix']}"
+                        f"/{self._cam_name}/audio/{self._audio_name}"
+                    ),
+                },
+            },
+        )
+
+    @callback  # type: ignore[misc]
+    def _state_message_received(self, msg: ReceiveMessage) -> None:
+        """Handle a new received MQTT state message."""
+        self._is_on = msg.payload == "ON"
+        self.async_write_ha_state()
+
+    @property
+    def unique_id(self) -> str:
+        """Return a unique ID for this entity."""
+        return get_frigate_entity_unique_id(
+            self._config_entry.entry_id,
+            "audio_sensor",
+            f"{self._cam_name}_{self._audio_name}",
+        )
+
+    @property
+    def device_info(self) -> dict[str, Any]:
+        """Return device information."""
+        return {
+            "identifiers": {
+                get_frigate_device_identifier(self._config_entry, self._cam_name)
+            },
+            "via_device": get_frigate_device_identifier(self._config_entry),
+            "name": get_friendly_name(self._cam_name),
+            "model": self._get_model(),
+            "configuration_url": f"{self._config_entry.data.get(CONF_URL)}/cameras/{self._cam_name}",
+            "manufacturer": NAME,
+        }
+
+    @property
+    def name(self) -> str:
+        """Return the name of the sensor."""
+        return f"{self._audio_name} sound"
+
+    @property
+    def is_on(self) -> bool:
+        """Return true if the binary sensor is on."""
+        return self._is_on
+
+    @property
+    def device_class(self) -> str:
+        """Return the device class."""
+        return cast(str, BinarySensorDeviceClass.SOUND)
+
+    @property
+    def icon(self) -> str:
+        """Return the icon of the sensor."""
+        return get_dynamic_icon_from_type("sound", self._is_on)
+
+
 class FrigateMotionSensor(FrigateMQTTEntity, BinarySensorEntity):  # type: ignore[misc]
     """Frigate Motion Sensor class."""
 

custom_components/frigate/icons.py

@@ -1,6 +1,7 @@
 """Handles icons for different entity types."""
 
 ICON_AUDIO = "mdi:ear-hearing"
+ICON_AUDIO_OFF = "mdi:ear-hearing-off"
 ICON_BICYCLE = "mdi:bicycle"
 ICON_CAR = "mdi:car"
 ICON_CAT = "mdi:cat"
@@ -32,6 +33,8 @@ def get_dynamic_icon_from_type(obj_type: str, is_on: bool) -> str:
         return ICON_CAR if is_on else ICON_CAR_OFF
     if obj_type == "dog":
         return ICON_DOG if is_on else ICON_DOG_OFF
+    if obj_type == "sound":
+        return ICON_AUDIO if is_on else ICON_AUDIO_OFF
 
     return ICON_DEFAULT_ON if is_on else ICON_DEFAULT_OFF
 

tests/__init__.py

@@ -22,6 +22,7 @@
 TEST_BINARY_SENSOR_FRONT_DOOR_PERSON_OCCUPANCY_ENTITY_ID = (
     "binary_sensor.front_door_person_occupancy"
 )
+TEST_BINARY_SENSOR_FRONT_DOOR_SPEECH_ENTITY_ID = "binary_sensor.front_door_speech_sound"
 TEST_BINARY_SENSOR_FRONT_DOOR_ALL_OCCUPANCY_ENTITY_ID = (
     "binary_sensor.front_door_all_occupancy"
 )

tests/test_binary_sensor.py

@@ -17,6 +17,7 @@
     TEST_BINARY_SENSOR_FRONT_DOOR_ALL_OCCUPANCY_ENTITY_ID,
     TEST_BINARY_SENSOR_FRONT_DOOR_MOTION_ENTITY_ID,
     TEST_BINARY_SENSOR_FRONT_DOOR_PERSON_OCCUPANCY_ENTITY_ID,
+    TEST_BINARY_SENSOR_FRONT_DOOR_SPEECH_ENTITY_ID,
     TEST_BINARY_SENSOR_STEPS_ALL_OCCUPANCY_ENTITY_ID,
     TEST_BINARY_SENSOR_STEPS_PERSON_OCCUPANCY_ENTITY_ID,
     TEST_CONFIG_ENTRY_ID,
@@ -78,6 +79,38 @@ async def test_occupancy_binary_sensor_setup(hass: HomeAssistant) -> None:
     assert entity_state.state == "unavailable"
 
 
+async def test_audio_binary_sensor_setup(hass: HomeAssistant) -> None:
+    """Verify a successful audio binary sensor setup."""
+    await setup_mock_frigate_config_entry(hass)
+
+    entity_state = hass.states.get(TEST_BINARY_SENSOR_FRONT_DOOR_SPEECH_ENTITY_ID)
+    assert entity_state
+    assert entity_state.state == "unavailable"
+
+    async_fire_mqtt_message(hass, "frigate/available", "online")
+    await hass.async_block_till_done()
+    entity_state = hass.states.get(TEST_BINARY_SENSOR_FRONT_DOOR_SPEECH_ENTITY_ID)
+    assert entity_state
+    assert entity_state.state == "off"
+
+    async_fire_mqtt_message(hass, "frigate/front_door/audio/speech", "ON")
+    await hass.async_block_till_done()
+    entity_state = hass.states.get(TEST_BINARY_SENSOR_FRONT_DOOR_SPEECH_ENTITY_ID)
+    assert entity_state
+    assert entity_state.state == "on"
+
+    async_fire_mqtt_message(hass, "frigate/front_door/audio/speech", "not_valid")
+    await hass.async_block_till_done()
+    entity_state = hass.states.get(TEST_BINARY_SENSOR_FRONT_DOOR_SPEECH_ENTITY_ID)
+    assert entity_state
+    assert entity_state.state == "off"
+
+    async_fire_mqtt_message(hass, "frigate/available", "offline")
+    entity_state = hass.states.get(TEST_BINARY_SENSOR_FRONT_DOOR_SPEECH_ENTITY_ID)
+    assert entity_state
+    assert entity_state.state == "unavailable"
+
+
 async def test_motion_binary_sensor_setup(hass: HomeAssistant) -> None:
     """Verify a successful motion binary sensor setup."""
     await setup_mock_frigate_config_entry(hass)