Implement dynamic audio rate control

Author: vxgmichel

gambaterm/audio.py

@@ -1,8 +1,11 @@
 from __future__ import annotations
 
+import time
 from typing import Generator, Iterator, TYPE_CHECKING
-from queue import Queue, Empty, Full
+from queue import Queue, Empty
 from contextlib import contextmanager
+from threading import Lock
+from collections import deque
 
 import numpy as np
 import numpy.typing as npt
@@ -17,6 +20,10 @@
 class AudioOut:
     output_rate: float = 48000.0  # Hz
     buffersize: float = 0.05  # seconds
+    pi_kp: float = 0.002
+    pi_ki: float = 0.003
+    pi_deadband: float = 0.01
+    pi_i_clamp: float = 0.005
 
     input_rate: float
     speed: float
@@ -30,12 +37,49 @@ def __init__(
         self.input_rate = input_rate
         self.speed = speed
         self.resampler = resampler
-        self.queue = Queue(maxsize=100)
+        self.queue = Queue()
+        self.queued_frames = 0
         self.offset = 0
+        self.lock = Lock()
+
+        # Record last time `self.send()` was called
+        self.last_send = time.perf_counter()
+
+        # Calculate initial ratio and target frames in buffer
+        self.base_ratio = self.output_rate / self.input_rate / self.speed
+        self.pi_ratio = self.base_ratio
+        self.pi_target = int(round(self.output_rate * self.buffersize * 2))
+        self.pi_integral = 0.0
+        self.pi_deque = deque[int](maxlen=3)
+
+    def adapt_ratio(self, extra_frames: int) -> None:
+        # Get consistent view of time and queued frames
+        with self.lock:
+            delta = time.perf_counter() - self.last_send
+            queued_frames = self.queued_frames
+
+        # Compute pending frames
+        frames_predicted = int(round(delta * self.output_rate))
+        pending_frames = extra_frames + queued_frames + frames_predicted
+
+        # Average pending frames over last few calls to smooth out noise
+        self.pi_deque.appendleft(pending_frames)
+        pending_frames = sum(self.pi_deque) // len(self.pi_deque)
+
+        # Compute PI correction
+        error = self.pi_target - pending_frames
+        normalized_error = error / self.pi_target
+        if abs(normalized_error) >= self.pi_deadband:
+            self.pi_integral += normalized_error
+        p_term = self.pi_kp * normalized_error
+        i_term = np.clip(
+            self.pi_ki * self.pi_integral, -self.pi_i_clamp, self.pi_i_clamp
+        )
+        self.pi_integral = i_term / self.pi_ki
+        correction = p_term + i_term
 
-    @property
-    def ratio(self) -> float:
-        return self.output_rate / self.input_rate / self.speed
+        # Update ratio
+        self.pi_ratio = self.base_ratio * (1.0 + correction)
 
     @contextmanager
     def run(self) -> Iterator[AudioOut]:
@@ -51,33 +95,40 @@ def run(self) -> Iterator[AudioOut]:
             buffersize_msec=int(round(self.buffersize * 1000)),
         )
         device.start(stream)
-        try:
+        with device:
             yield self
-        finally:
-            device.stop()
 
     def send(self, audio: npt.NDArray[np.int16]) -> None:
         # Resample to output rate
-        data = self.resampler.process(audio, self.ratio)
+        data = self.resampler.process(audio, self.pi_ratio)
         # Reshape
         data = data.astype(np.int16).reshape(-1, 2)
         # Reduce volume
         data //= 4
-        # Send without blocking if possible
-        try:
+        # Send without blocking
+        with self.lock:
             self.queue.put_nowait(data)
-        # Synchronization issue, let it regulate itself
-        except Full:
-            pass
+            self.queued_frames += len(data)
+            self.last_send = time.perf_counter()
 
     def _audio_stream(self) -> Generator[bytes, int, None]:
-        # Get first required frames
+        # Initialize state
         extra_frames_start = 0
         buffer = np.empty((0, 2), dtype=np.int16)
+
+        # Get first required frames
         required_frames = yield b""
+        result = np.zeros((required_frames, 2), dtype=np.int16)
+
+        # Wait until we have enough frames to fill the first request
+        while self.queued_frames < 1.5 * required_frames:
+            required_frames = yield result.tobytes()
 
         # Loop over requested frames
         while True:
+            # Adapt ratio
+            self.adapt_ratio(len(buffer) - extra_frames_start)
+
             # Initialize result buffer
             result = np.zeros((required_frames, 2), dtype=np.int16)
 
@@ -96,7 +147,9 @@ def _audio_stream(self) -> Generator[bytes, int, None]:
                 # Get more frames until we have enough
                 while True:
                     try:
-                        buffer = self.queue.get_nowait()
+                        with self.lock:
+                            buffer = self.queue.get_nowait()
+                            self.queued_frames -= len(buffer)
                     except Empty:
                         extra_frames_start = len(buffer)
                         break

gambaterm/run.py

@@ -84,10 +84,6 @@ def run(
     data_length: Deque[int] = deque(maxlen=average_over)
     start = time.time()
 
-    # Create a 100 ms time shift to fill up audio buffer
-    if audio_out:
-        start -= 0.1
-
     # Prepare state
     new_frame = False
     screen_ready = True