Add metronome to library

1 files changed, 73 insertions, 0 deletions

diff --git a/solo-tool-project/src/solo_tool/metronome.py b/solo-tool-project/src/solo_tool/metronome.py
new file mode 100644
index 0000000..d99e0b8
--- /dev/null
+++ b/solo-tool-project/src/solo_tool/metronome.py
@@ -0,0 +1,73 @@
+import pyaudio as pa
+from array import array
+
+def _createClickSample(clickFrequency: int, samplingRate: int) -> array:
+    clickDuration = 0.1 # seconds
+
+    sampleLength = int(clickDuration * samplingRate)
+    fadeSamples = 10
+    attackEnv = [i/fadeSamples for i in range(1, fadeSamples, 1)]
+    releaseEnv = [i/fadeSamples for i in range(fadeSamples, 0, -1)]
+    amplitudeEnv = attackEnv + [1.0] * (sampleLength - len(attackEnv) - len(releaseEnv)) + releaseEnv
+    cycleLength = (1 / clickFrequency) * samplingRate
+
+    from math import sin, pi
+    return array('h', [int(sin(((i % cycleLength) / cycleLength) * 2 * pi) * amplitudeEnv[i] * 32767) for i in range(sampleLength)])
+
+def _createOutputBuffer(bpm: int, clickFrequency: int, samplingRate: int) -> array:
+    clickSample = _createClickSample(clickFrequency, samplingRate)
+    samplesPerBeat = int((60 / bpm) * samplingRate)
+    assert len(clickSample) <= samplesPerBeat
+    return clickSample + array('h', [0] * (samplesPerBeat - len(clickSample)))
+
+class Metronome:
+    def __init__(self, bufferSize: int, samplingRate: int):
+        self._bufferSize = bufferSize
+        self._samplingRate = samplingRate
+        self._sampleFormat = pa.paInt16
+        self._channels = 1
+        self._pa = pa.PyAudio()
+        self._stream = None
+        self._bufferPointer = 0
+        self._outputBuffer = array('h')
+
+    def __del__(self):
+        if self.running:
+            self._stream.stop_stream()
+            self._stream.close()
+
+    def _callback(self, inData, frameCount, timeInfo, statusFlags):
+        leftover = len(self._outputBuffer) - self._bufferPointer
+        if frameCount > leftover:
+            outData = self._outputBuffer[self._bufferPointer:] + self._outputBuffer[0:frameCount - leftover]
+            self._bufferPointer = frameCount - leftover
+        else:
+            outData = self._outputBuffer[self._bufferPointer:self._bufferPointer + frameCount]
+            self._bufferPointer += frameCount
+        assert self._bufferPointer >= 0 and self._bufferPointer <= len(self._outputBuffer)
+        return (outData.tobytes(), pa.paContinue)
+
+    def start(self, bpm: int, clickFrequency: int) -> None:
+        if self.running:
+            return
+        self._bufferPointer = 0
+        self._outputBuffer = _createOutputBuffer(bpm, clickFrequency, self._samplingRate)
+        self._stream = self._pa.open(format=self._sampleFormat,
+                                     channels=self._channels,
+                                     rate=self._samplingRate,
+                                     frames_per_buffer=self._bufferSize,
+                                     output=True,
+                                     stream_callback=self._callback)
+
+    def stop(self) -> None:
+        if not self.running:
+            return None
+
+        self._stream.stop_stream()
+        self._stream.close()
+        self._stream = None
+
+    @property
+    def running(self) -> bool:
+        return self._stream is not None
+