Separate max and mean volume sensors

haffmpeg/sensor.py

@@ -119,14 +119,15 @@ async def _worker_process(self) -> None:
             _LOGGER.warning("Unknown data from queue!")
 
 
-class SensorVolume(HAFFmpegWorker):
+class SensorVolumeBase(HAFFmpegWorker):
     """Implement volume sensor on audio stream."""
 
-    def __init__(self, ffmpeg_bin: str, callback: Callable):
+    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."""
@@ -153,14 +154,12 @@ def open_sensor(
         )
 
     async def _worker_process(self) -> None:
-        """This function processes stream data into a mean dB and max dB sensor."""
+        """This function extracts mean/max dB into the state sensor."""
         state: Decimal = -100
         timeout = self._time_duration
 
         self._loop.call_soon(self._callback, False)
 
-        re_mean = re.compile(r"mean_volume: (\-\d{1,3}(\.\d{1,2})?) dB")
-
         # process queue data
         while True:
             try:
@@ -177,15 +176,33 @@ async def _worker_process(self) -> None:
                 timeout = None
                 continue
 
-            mean_matches = re_mean.search(data)
-            if mean_matches:
-                state = Decimal(mean_matches[1])
+            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_volume.py → test/sensor_max_volume.py

@@ -3,7 +3,7 @@
 
 import click
 
-from haffmpeg.sensor import SensorVolume
+from haffmpeg.sensor import MaxSensorVolume
 
 logging.basicConfig(level=logging.DEBUG)
 
@@ -20,16 +20,16 @@
     help="Time duration to detect volume",
 )
 @click.option("--extra", "-e", help="Extra ffmpeg command line arguments")
-def cli(ffmpeg, source, output, duration, reset, peak, extra):
-    """FFMPEG volume detection."""
+def cli(ffmpeg, source, output, duration, extra):
+    """FFMPEG max volume detection."""
 
     def callback(state):
-        print("Volume is: %s" % str(state))
+        print("Max volume is: %s" % str(state))
 
     async def run():
 
-        sensor = SensorVolume(ffmpeg_bin=ffmpeg, callback=callback)
-        sensor.set_options(time_duration=duration, time_reset=reset, peak=peak)
+        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
         )

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()