Feature: volume detection

README.md

@@ -3,7 +3,7 @@
 A Python library to control ffmepg from asyncio for [Home Assistant](https://www.home-assistant.io).
 
 - Emulate webcam from any video input source for HA
-- Analyse a video/audio stream for noise or motion detection
+- Analyse a video/audio stream for volume, noise or motion detection
 - Grab image from a stream
 
 Be carfull that you protect function calls to this library with `asyncio.shield`.

haffmpeg/sensor.py

@@ -4,6 +4,9 @@
 import re
 from time import time
 from typing import Callable, Coroutine, Optional
+from decimal import Decimal
+
+import async_timeout
 
 from .core import FFMPEG_STDOUT, HAFFmpegWorker
 from .timeout import asyncio_timeout
@@ -12,7 +15,7 @@
 
 
 class SensorNoise(HAFFmpegWorker):
-    """Implement a noise detection on a autio stream."""
+    """Implement a noise detection on a audio stream."""
 
     STATE_NONE = 0
     STATE_NOISE = 1
@@ -42,7 +45,7 @@ def open_sensor(
         output_dest: Optional[str] = None,
         extra_cmd: Optional[str] = None,
     ) -> Coroutine:
-        """Open FFmpeg process for read autio stream.
+        """Open FFmpeg process for read audio stream.
 
         Return a coroutine.
         """
@@ -116,6 +119,90 @@ async def _worker_process(self) -> None:
             _LOGGER.warning("Unknown data from queue!")
 
 
+class SensorVolumeBase(HAFFmpegWorker):
+    """Implement volume sensor on audio stream."""
+
+    def __init__(self, ffmpeg_bin: str, callback: Callable, re_state: re.Pattern):
+        """Init volume sensor."""
+        super().__init__(ffmpeg_bin)
+        self._callback = callback
+        self._time_duration = 1
+        self._re_state: re.Pattern = re_state
+
+    def set_options(self, time_duration: int = 1) -> None:
+        """Set option parameters for volume sensor."""
+        self._time_duration = time_duration
+
+    def open_sensor(
+        self,
+        input_source: str,
+        output_dest: Optional[str] = None,
+        extra_cmd: Optional[str] = None,
+    ) -> Coroutine:
+        """Open FFmpeg process for read audio stream.
+
+        Return a coroutine.
+        """
+        command = ["-vn", "-filter:a", "volumedetect"]
+
+        # run ffmpeg, read output
+        return self.start_worker(
+            cmd=command,
+            input_source=input_source,
+            output=output_dest,
+            extra_cmd=extra_cmd,
+        )
+
+    async def _worker_process(self) -> None:
+        """This function extracts mean/max dB into the state sensor."""
+        state: Decimal = -100
+        timeout = self._time_duration
+
+        self._loop.call_soon(self._callback, False)
+
+        # process queue data
+        while True:
+            try:
+                _LOGGER.debug("Reading mean: %ddB state with timeout: %s", state, timeout)
+                with async_timeout.timeout(timeout):
+                    data = await self._queue.get()
+                timeout = None
+                if data is None:
+                    self._loop.call_soon(self._callback, None)
+                    return
+            except asyncio.TimeoutError:
+                _LOGGER.debug("Blocking timeout")
+                self._loop.call_soon(self._callback, None)
+                timeout = None
+                continue
+
+            match = self._re_state.search(data)
+            if match:
+                state = Decimal(match[1])
+                self._loop.call_soon(self._callback, state)
+                continue
+
+            _LOGGER.warning("Unknown data from queue!")
+
+
+class MeanSensorVolume(SensorVolumeBase):
+    """Implement a mean volume sensor."""
+
+    def __init__(self, ffmpeg_bin: str, callback: Callable):
+        """Init mean volume sensor."""
+        re_state = re.compile(r"mean_volume: (\-\d{1,3}(\.\d{1,2})?) dB")
+        super().__init__(ffmpeg_bin, callback, re_state)
+
+
+class MaxSensorVolume(SensorVolumeBase):
+    """Implement a max volume sensor."""
+
+    def __init__(self, ffmpeg_bin: str, callback: Callable):
+        """Init max volume sensor."""
+        re_state = re.compile(r"max_volume: (\-\d{1,3}(\.\d{1,2})?) dB")
+        super().__init__(ffmpeg_bin, callback, re_state)
+
+
 class SensorMotion(HAFFmpegWorker):
     """Implement motion detection with ffmpeg scene detection."""
 

test/sensor_max_volume.py

@@ -0,0 +1,46 @@
+import asyncio
+import logging
+
+import click
+
+from haffmpeg.sensor import MaxSensorVolume
+
+logging.basicConfig(level=logging.DEBUG)
+
+
+@click.command()
+@click.option("--ffmpeg", "-f", default="ffmpeg", help="FFmpeg binary")
+@click.option("--source", "-s", help="Input file for ffmpeg")
+@click.option("--output", "-o", default=None, help="Output ffmpeg target")
+@click.option(
+    "--duration",
+    "-d",
+    default=1,
+    type=int,
+    help="Time duration to detect volume",
+)
+@click.option("--extra", "-e", help="Extra ffmpeg command line arguments")
+def cli(ffmpeg, source, output, duration, extra):
+    """FFMPEG max volume detection."""
+
+    def callback(state):
+        print("Max volume is: %s" % str(state))
+
+    async def run():
+
+        sensor = MaxSensorVolume(ffmpeg_bin=ffmpeg, callback=callback)
+        sensor.set_options(time_duration=duration)
+        await sensor.open_sensor(
+            input_source=source, output_dest=output, extra_cmd=extra
+        )
+        try:
+            while True:
+                await asyncio.sleep(0.1)
+        finally:
+            await sensor.close()
+
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    cli()

test/sensor_mean_volume.py

@@ -0,0 +1,46 @@
+import asyncio
+import logging
+
+import click
+
+from haffmpeg.sensor import MeanSensorVolume
+
+logging.basicConfig(level=logging.DEBUG)
+
+
+@click.command()
+@click.option("--ffmpeg", "-f", default="ffmpeg", help="FFmpeg binary")
+@click.option("--source", "-s", help="Input file for ffmpeg")
+@click.option("--output", "-o", default=None, help="Output ffmpeg target")
+@click.option(
+    "--duration",
+    "-d",
+    default=1,
+    type=int,
+    help="Time duration to detect volume",
+)
+@click.option("--extra", "-e", help="Extra ffmpeg command line arguments")
+def cli(ffmpeg, source, output, duration, extra):
+    """FFMPEG mean volume detection."""
+
+    def callback(state):
+        print("Mean volume is: %s" % str(state))
+
+    async def run():
+
+        sensor = MeanSensorVolume(ffmpeg_bin=ffmpeg, callback=callback)
+        sensor.set_options(time_duration=duration)
+        await sensor.open_sensor(
+            input_source=source, output_dest=output, extra_cmd=extra
+        )
+        try:
+            while True:
+                await asyncio.sleep(0.1)
+        finally:
+            await sensor.close()
+
+    asyncio.run(run())
+
+
+if __name__ == "__main__":
+    cli()