More experimenting

.pylintrc

@@ -1,2 +1,5 @@
 [DESIGN]
 max-args=6
+
+[MESSAGES CONTROL]
+disable=too-few-public-methods

local/configs/package.yaml

@@ -19,3 +19,6 @@ commands:
 mypy_local: |
   [mypy-scipy.*]
   ignore_missing_imports = True
+
+  [mypy-pyaudio.*]
+  ignore_missing_imports = True

mypy.ini

@@ -17,3 +17,6 @@ strict_equality = False
 # quasimoto-specific configurations.
 [mypy-scipy.*]
 ignore_missing_imports = True
+
+[mypy-pyaudio.*]
+ignore_missing_imports = True

tasks/default.yaml

@@ -0,0 +1,10 @@
+---
+app:
+  - tasks.dev.main
+  - runtimepy.net.apps.wait_for_stop
+
+factories:
+  - {name: tasks.dev.Stereo}
+
+tasks:
+  - {name: stereo, factory: stereo, period_s: 0.1}

tasks/dev.py

@@ -0,0 +1,100 @@
+"""
+A module implementing developmental runtimepy interfaces.
+"""
+
+# built-in
+from contextlib import contextmanager
+from typing import Iterator
+
+# third-party
+import pyaudio
+from runtimepy.net.arbiter import AppInfo
+from runtimepy.net.arbiter.task import ArbiterTask, TaskFactory
+
+# internal
+from tasks.stereo import StereoInterface
+
+
+@contextmanager
+def get_pyaudio() -> Iterator[pyaudio.PyAudio]:
+    """Get a PyAudio instance."""
+
+    audio = pyaudio.PyAudio()
+    try:
+        yield audio
+    finally:
+        audio.terminate()
+
+
+class StereoTask(ArbiterTask):
+    """A task for logging metrics."""
+
+    auto_finalize = True
+
+    audio: pyaudio.PyAudio
+    stream: pyaudio.Stream
+
+    def _init_state(self) -> None:
+        """Add channels to this instance's channel environment."""
+
+        # Add channels from this here.
+        self.stereo = StereoInterface()
+
+        # Make this a channel.
+        self.buffer_depth_scalar = 10.0
+
+    @staticmethod
+    @contextmanager
+    def get_stream(
+        audio: pyaudio.PyAudio, stereo: StereoInterface
+    ) -> Iterator[pyaudio.Stream]:
+        """Get a pyaudio stream."""
+
+        try:
+            stream = audio.open(
+                format=audio.get_format_from_width(stereo.left.num_bits // 8),
+                channels=stereo.num_channels,
+                rate=stereo.left.sample_rate,
+                stream_callback=stereo.callback,
+                output=True,
+            )
+            yield stream
+        finally:
+            stream.close()
+
+    async def init(self, app: AppInfo) -> None:
+        """Initialize this task with application information."""
+
+        await super().init(app)
+
+        self.audio = app.stack.enter_context(get_pyaudio())
+
+        self.stream = app.stack.enter_context(
+            StereoTask.get_stream(self.audio, self.stereo)
+        )
+
+    async def dispatch(self) -> bool:
+        """Dispatch an iteration of this task."""
+
+        result: bool = self.stream.is_active()
+        if result:
+            # Populate 10x our period
+            self.stereo.buffer_to_duration(
+                self.period_s.value * self.buffer_depth_scalar
+            )
+
+        return result
+
+
+class Stereo(TaskFactory[StereoTask]):
+    """A factory for the stereo task."""
+
+    kind = StereoTask
+
+
+async def main(app: AppInfo) -> int:
+    """Waits for the stop signal to be set."""
+
+    del app
+
+    return 0

tasks/stereo/__init__.py

@@ -0,0 +1,75 @@
+"""
+A module implementing some stereo-audio interfaces.
+"""
+
+# built-in
+from io import BytesIO
+from queue import SimpleQueue
+
+# third-party
+import pyaudio
+
+# internal
+from quasimoto.sampler import Sampler
+from quasimoto.wave import WaveWriter
+
+
+class StereoInterface:
+    """An interface for managing stereo sound output."""
+
+    num_channels = 2
+
+    def __init__(self) -> None:
+        """Initialize this instance."""
+
+        self.left = Sampler()
+        self.right = self.left.copy(harmonic=-1)
+
+        self.sample_queue: SimpleQueue[tuple[int, int]] = SimpleQueue()
+
+    def buffer_to_duration(self, duration_s: float) -> None:
+        """Fill sample-queue buffer to the specified duration."""
+
+        for _ in range(
+            int(duration_s * self.left.sample_rate) - self.sample_queue.qsize()
+        ):
+            self.sample_queue.put_nowait((next(self.left), next(self.right)))
+
+    def frames(self, frame_count: int) -> bytes:
+        """Get sample frames in a single chunk of bytes."""
+
+        with BytesIO() as stream:
+            # Get pre-computed samples from queue.
+            from_queue = min(self.sample_queue.qsize(), frame_count)
+            for _ in range(from_queue):
+                left_sample, right_sample = self.sample_queue.get_nowait()
+                WaveWriter.to_stream(stream, left_sample)
+                WaveWriter.to_stream(stream, right_sample)
+
+            # Get new samples if necessary.
+            frame_count -= from_queue
+            for _ in range(frame_count):
+                WaveWriter.to_stream(stream, next(self.left))
+                WaveWriter.to_stream(stream, next(self.right))
+
+            return stream.getvalue()
+
+    def callback(self, in_data, frame_count, time_info, status):
+        """Called when stream needs more data in raw bytes."""
+
+        # Not an input callback (is None).
+        del in_data
+
+        # Example Contents:
+        #
+        # {
+        #   'input_buffer_adc_time': 0.8380952380952347,
+        #   'current_time': 16208.699135654,
+        #   'output_buffer_dac_time': 16208.709929304794
+        # }
+        del time_info
+
+        # Always 0?
+        del status
+
+        return (self.frames(frame_count), pyaudio.paContinue)