Now outputs mp3s

.gitignore

@@ -1,3 +1,6 @@
+outputs
+*.mp3
+
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

dogbarking/__main__.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 from typing import Annotated
 from datetime import datetime
 import pyaudio
@@ -32,9 +33,15 @@ def nogui(
     sample_freq: Annotated[
         int, typer.Argument(help="The sample rate in Hz.", min=0)
     ] = 44100,
+    keep_historical_seconds: Annotated[
+        int, typer.Argument(help="The number of seconds to save to audio.", min=0)
+    ] = 10,
     seconds_per_buffer: Annotated[
         float, typer.Argument(help="The number of seconds per buffer.", min=0.0)
     ] = 0.1,
+    save_path: Annotated[
+        Path, typer.Argument(help="The path to save the audio file to.")
+    ] = "./outputs",
     # email: Annotated[Optional[str], "The email to send the alert to."]=None
 ):
     logger.warning("Remember to turn your volume all the way up!")
@@ -50,6 +57,7 @@ def nogui(
     )
     r = Recorder(
         audio=audio,
+        keep_historical_buffers=int(keep_historical_seconds / seconds_per_buffer),
         sample_freq=sample_freq,
         frames_per_buffer=int(sample_freq * seconds_per_buffer),
     )
@@ -65,6 +73,10 @@ def nogui(
             # Stop the recording, don't want to record the sound we are playing
             r.stop()
 
+            # Save the recording
+            filename = save_path / f"{datetime.now().isoformat()}.mp3"
+            r.save(filename)
+
             # Play the sound
             p.play_sound()
 

dogbarking/audio.py

@@ -1,12 +1,15 @@
 import functools
-from typing import Generator, Optional
+from pathlib import Path
+from typing import Generator, List, Optional
 import pyaudio
 import numpy as np
 import numpy.typing as npt
 from pyaudio import Stream
 from pydantic import BaseModel, PrivateAttr
 import numpy
 from loguru import logger
+import soundfile as sf
+import os
 
 # REF: https://gist.github.com/mabdrabo/8678538
 FORMAT = pyaudio.paFloat32
@@ -16,12 +19,15 @@ class Recorder(BaseModel):
     audio: pyaudio.PyAudio
     frames_per_buffer: int
     sample_freq: int
+    keep_historical_buffers: int
     _stream: Optional[Stream] = PrivateAttr()
+    _waveform: Optional[List[npt.NDArray[np.float32]]] = PrivateAttr()
 
     class Config:
         arbitrary_types_allowed = True
 
     def start(self):
+        self._waveform = []
         self._stream = self.audio.open(
             format=FORMAT,
             channels=1,
@@ -50,7 +56,11 @@ def device_index(self) -> int:
 
     def __iter__(self) -> Generator[npt.NDArray[np.float32], None, None]:  # type: ignore
         while True:
-            yield self.get_buffer()
+            out = self.get_buffer()
+            if len(self._waveform) > self.keep_historical_buffers:
+                self._waveform.pop(0)
+            self._waveform.append(out)
+            yield out
 
     def get_buffer(self) -> npt.NDArray[np.float32]:
         """REF: https://stackoverflow.com/questions/19629496/get-an-audio-sample-as-float-number-from-pyaudio-stream # noqa"""
@@ -59,6 +69,17 @@ def get_buffer(self) -> npt.NDArray[np.float32]:
         decoded = numpy.fromstring(data, np.float32)  # type: ignore
         return decoded
 
+    def save(self, filepath: Path) -> None:
+        """Save the recorded waveform to an mp3 file."""
+        # If directory doesn't exist, create it
+        directory = filepath.parent
+        if not directory.exists():
+            os.makedirs(directory)
+
+        waveform = np.concatenate(self._waveform)
+        sf.write(filepath, waveform, self.sample_freq)
+        logger.info(f"Saved {filepath}")
+
 
 class Player(BaseModel):
     audio: pyaudio.PyAudio

pyproject.toml

@@ -17,6 +17,7 @@ pyaudio = "^0.2.14"
 typer-config = "^1.4.0"
 ruff = "^0.3.3"
 loguru = "^0.7.2"
+soundfile = "^0.12.1"
 
 [tool.poetry.group.dev.dependencies]
 mypy = "^1.9.0"