orchestrata_ultimate.session.audio_alarm

Reference › orchestrata_ultimate › session › audio_alarm

The attention bridge for goals (SPEC-023, user story 4): an audible signal when a goal is achieved or blocked. select_backend returns the first audio backend found on the PATH, or nothing, in which case the terminal bell is used. AudioAlarm plays an ascending success tone sequence (play_success) or a low attention tone (play_attention) once, and start_repeat repeats the alarm at an interval until stop is called, for example when the user sends a message. GoalMixin calls play_success when goal_evaluate finds every exit criterion met; the --no-alarm option of the CLI evaluate command and the play_alarm argument skip it.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

AudioAlarm

Decorators: traces(spec='023', story='US-004', reqs=['REQ-023-017', 'REQ-023-018', 'REQ-023-019', 'REQ-023-020', 'REQ-023-021']) · Line 148

Attention bridge for goal achievement and blocking (US-004).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/audio_alarm.py

"""Audio alarm for goal achievement / blocking (SPEC-023 US-004).

Non-blocking by design (REQ-023-021): every tone is a detached subprocess
and the repeating alarm runs on a daemon thread, so the agent can always
keep producing output. Backend chain (REQ-023-019): ``ffplay``, then
``paplay``, then ``aplay``, then ``mpv``, then the terminal bell. The
alarm repeats every 30 seconds until ``stop()`` is called (REQ-023-020),
which the session entrypoints do when the user sends a message.

Tones are synthesized with the stdlib ``wave`` module for the file-based
backends (paplay/aplay/mpv) and with ffmpeg's ``lavfi`` sine generator
for ffplay — no audio assets, no extra dependencies.
"""

from __future__ import annotations

import io
import math
import os
import shutil
import struct
import subprocess  # noqa: S404
import tempfile
import threading
import wave
from collections.abc import Mapping

from orchestrata_ultimate.core.base.traces import traces

# Backend chain order (REQ-023-019): ffplay, paplay, aplay, mpv, bell.
_BACKENDS: tuple[str, ...] = ("ffplay", "paplay", "aplay", "mpv")

# US-004: success = ascending tones; attention = low repeating tones.
_SUCCESS_FREQUENCIES: tuple[float, ...] = (660.0, 880.0)
_ATTENTION_FREQUENCY: float = 330.0
_TONE_SECONDS: float = 0.4
_SAMPLE_RATE = 22050
DEFAULT_REPEAT_SECONDS: float = 30.0


def select_backend(env: Mapping[str, str] | None = None) -> str | None:
    """Return the first audio backend available on PATH, else None (bell).

    Args:
        env: Optional environment mapping for PATH resolution (test seam).
            Defaults to the process environment.
    """
    path = (env or os.environ).get("PATH", "")
    for backend in _BACKENDS:
        if shutil.which(backend, path=path):
            return backend
    return None


def _sine_wav_bytes(frequency: float, seconds: float) -> bytes:
    """Synthesize a mono 16-bit PCM sine tone into WAV bytes."""
    frame_count = int(_SAMPLE_RATE * seconds)
    buffer = io.BytesIO()
    with wave.open(buffer, "wb") as wav:
        wav.setnchannels(1)
        wav.setsampwidth(2)
        wav.setframerate(_SAMPLE_RATE)
        frames = bytearray()
        for i in range(frame_count):
            sample = 0.5 * math.sin(2.0 * math.pi * frequency * (i / _SAMPLE_RATE))
            frames += struct.pack("<h", int(sample * 32767))
        wav.writeframes(bytes(frames))
    return buffer.getvalue()


def _ffplay_args(frequency: float, seconds: float) -> list[str]:
    """ffplay command for one sine tone via the lavfi filter."""
    return [
        "ffplay",
        "-nodisp",
        "-autoexit",
        "-loglevel",
        "quiet",
        "-f",
        "lavfi",
        f"sine=frequency={int(frequency)}:duration={seconds}",
    ]


def _play_tone(
    frequency: float, seconds: float, env: Mapping[str, str] | None = None
) -> None:
    """Play one tone on the first available backend; never raises."""
    backend = select_backend(env)
    # env is an OVERLAY on the process environment (e.g. a test-scoped
    # PATH), never a replacement — audio backends legitimately need
    # HOME/DISPLAY/etc from the real environment.
    resolved_env = {**os.environ, **(dict(env))} if env is not None else None
    if backend is None:
        # Terminal bell fallback — the last entry in the REQ-023-019 chain.
        print("\a", end="", flush=True)
        return
    try:
        if backend == "ffplay":
            subprocess.Popen(  # noqa: S603
                _ffplay_args(frequency, seconds),
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                env=resolved_env,
            )
            return
        # File-based backends (paplay/aplay/mpv): a real synthesized WAV.
        handle = tempfile.NamedTemporaryFile(  # noqa: SIM115
            prefix="orchestrata-alarm-", suffix=".wav", delete=False
        )
        handle.write(_sine_wav_bytes(frequency, seconds))
        handle.close()
        wav_path = handle.name
        process = subprocess.Popen(  # noqa: S603
            [backend, wav_path],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            env=resolved_env,
        )

        def _cleanup() -> None:
            """Remove the temp WAV once playback finished."""
            try:
                process.wait()
            finally:
                try:
                    os.unlink(wav_path)
                except OSError:
                    pass

        threading.Thread(target=_cleanup, daemon=True).start()
    except OSError:
        # A backend vanished between selection and spawn: fall back to bell.
        print("\a", end="", flush=True)


@traces(
    spec="023",
    story="US-004",
    reqs=[
        "REQ-023-017",  # success alarm: ascending tones
        "REQ-023-018",  # attention alarm: low repeating tones
        "REQ-023-019",  # backend chain ffplay/paplay/aplay/mpv/bell
        "REQ-023-020",  # repeat every 30s until the user sends a message
        "REQ-023-021",  # non-blocking
    ],
)
class AudioAlarm:
    """Attention bridge for goal achievement and blocking (US-004)."""

    def __init__(
        self,
        repeat_seconds: float = DEFAULT_REPEAT_SECONDS,
        env: Mapping[str, str] | None = None,
    ) -> None:
        self._repeat_seconds = repeat_seconds
        self._env = env
        self._stop = threading.Event()
        self._thread: threading.Thread | None = None

    def play_success(self) -> None:
        """Play the ascending success tone sequence once."""
        for frequency in _SUCCESS_FREQUENCIES:
            _play_tone(frequency, _TONE_SECONDS, env=self._env)

    def play_attention(self) -> None:
        """Play the low attention tone once."""
        _play_tone(_ATTENTION_FREQUENCY, _TONE_SECONDS, env=self._env)

    def start_repeat(self, attention: bool = True) -> None:
        """Repeat the alarm every ``repeat_seconds`` until ``stop()``.

        Non-blocking: the loop runs on a daemon thread.
        """
        self._stop.clear()
        if self._thread is not None and self._thread.is_alive():
            return
        self._thread = threading.Thread(
            target=self._repeat_loop, args=(attention,), daemon=True
        )
        self._thread.start()

    def _repeat_loop(self, attention: bool) -> None:
        while not self._stop.is_set():
            if attention:
                self.play_attention()
            else:
                self.play_success()
            self._stop.wait(self._repeat_seconds)

    def stop(self) -> None:
        """Stop the repeating alarm (e.g. the user sent a message)."""
        self._stop.set()