Robert Helewka
2a05be27bf
The gateway can now hear. Verified end to end against the Asterisk lab: speech classifies as live_human, hold music as music, the speech→music transition tracks the dialplan, and DTMF reaches a real IVR (Asterisk logged "caller pressed 1 -> accounts" and branched). RTP stats show 0% packet loss. PJSUAEngine implements the existing SIPEngine interface, so the gateway, call manager and hold-slayer service are unchanged. It is selected with SIP_ENGINE=pjsua2; the default stays "sippy" while this is proven, and the mock remains opt-in as before. Why a new engine rather than fixing the old path: PJSUA2 exposes no standalone RTP media object, so it will not surface media for a dialog it does not own. Owning the dialog is the price of owning the media. Sippy keeps the SBC roles it is good at — device registration, routing, leg bridging — and SippyEngine remains fully functional for signalling; it simply cannot carry media, which its media branch now says plainly instead of calling a method that could never work. The safety invariants are untouched. This engine is reachable only through gateway.make_call, which refuses emergency numbers and enforces the concurrency cap before any SIP action. No new dial path was introduced. MediaPipeline.add_remote_stream(host, port) is replaced by attach_call_media(stream_id, audio_media), called from onCallMediaState — the one place PJSUA2 hands out RTP-backed media. Taps requested before media comes up are attached when it does, so the classifier never misses the start of a call. Three crash/lifetime bugs found by running against the real bindings, none of which any unit test would have caught: - pj.Call and pj.Account objects finalised after libDestroy() abort the process on a native assertion, exactly as media ports do. Both are now dropped and collected before the pipeline destroys the endpoint. - PJSUA2 keeps delivering callbacks during interpreter teardown, when module globals may already be cleared. The callbacks alias what they need locally and swallow everything: a raise there escapes into C++ and takes the worker thread with it. - hangup() only queues the BYE, so shutdown deleted the account with a call still active and left the far end on an unclosed dialog. stop() now waits briefly for the teardown to complete. Threading follows the established rule: PJSUA2 worker threads reach the loop only through _post_from_pj → run_coroutine_threadsafe, and any thread PJSUA2 did not create registers itself before touching a PJSUA2 object. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
722 lines
27 KiB
Python
722 lines
27 KiB
Python
"""
|
|
Media Pipeline — PJSUA2 conference bridge and audio routing.
|
|
|
|
This is the media anchor for the gateway. PJSUA2 handles all RTP:
|
|
- Conference bridge (mixing, bridging call legs)
|
|
- Audio tapping (extracting audio for classifier + STT)
|
|
- WAV recording
|
|
- Tone generation (DTMF, comfort noise)
|
|
|
|
Architecture:
|
|
Each SIP call leg gets a transport + media port in PJSUA2's conf bridge.
|
|
The pipeline provides methods to:
|
|
- Add/remove RTP streams (tied to Sippy call legs)
|
|
- Bridge two streams (connect call legs)
|
|
- Tap a stream (fork audio to classifier/STT)
|
|
- Record a stream to WAV
|
|
- Play audio into a stream (prompts, comfort tones)
|
|
|
|
PJSUA2 runs in its own thread with a dedicated Endpoint.
|
|
"""
|
|
|
|
import asyncio
|
|
import gc
|
|
import logging
|
|
import threading
|
|
from collections.abc import AsyncIterator
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ================================================================
|
|
# Audio Tap — extracts audio frames for analysis
|
|
# ================================================================
|
|
|
|
class AudioTap:
|
|
"""
|
|
Taps into a conference bridge port to extract audio frames.
|
|
|
|
Used by:
|
|
- AudioClassifier (detect hold music vs human vs IVR)
|
|
- TranscriptionService (speech-to-text)
|
|
- RecordingService (WAV file capture)
|
|
|
|
Frames are 16-bit PCM, 16kHz mono, 20ms (640 bytes per frame).
|
|
"""
|
|
|
|
def __init__(self, stream_id: str, sample_rate: int = 16000, frame_ms: int = 20):
|
|
self.stream_id = stream_id
|
|
self.sample_rate = sample_rate
|
|
self.frame_ms = frame_ms
|
|
self.frame_size = int(sample_rate * frame_ms / 1000) * 2 # 16-bit = 2 bytes/sample
|
|
self._buffer: asyncio.Queue[bytes] = asyncio.Queue(maxsize=500)
|
|
self._active = True
|
|
self._pjsua2_port = None # PJSUA2 AudioMediaPort for tapping
|
|
# asyncio.Queue is not thread-safe; feed() hops onto this loop
|
|
try:
|
|
self._loop: Optional[asyncio.AbstractEventLoop] = asyncio.get_running_loop()
|
|
except RuntimeError:
|
|
self._loop = None
|
|
|
|
def feed(self, pcm_data: bytes) -> None:
|
|
"""Feed PCM audio data into the tap (called from the PJSUA2 thread)."""
|
|
if not self._active:
|
|
return
|
|
if self._loop is not None:
|
|
self._loop.call_soon_threadsafe(self._enqueue, pcm_data)
|
|
else:
|
|
self._enqueue(pcm_data)
|
|
|
|
def _enqueue(self, pcm_data: bytes) -> None:
|
|
"""Queue a frame on the owning loop, dropping oldest on overflow."""
|
|
try:
|
|
self._buffer.put_nowait(pcm_data)
|
|
except asyncio.QueueFull:
|
|
# Drop oldest frame to keep flowing
|
|
try:
|
|
self._buffer.get_nowait()
|
|
self._buffer.put_nowait(pcm_data)
|
|
except (asyncio.QueueEmpty, asyncio.QueueFull):
|
|
pass
|
|
|
|
async def read_frame(self, timeout: float = 1.0) -> Optional[bytes]:
|
|
"""Read the next audio frame (async)."""
|
|
try:
|
|
return await asyncio.wait_for(self._buffer.get(), timeout=timeout)
|
|
except asyncio.TimeoutError:
|
|
return None
|
|
|
|
async def stream(self) -> AsyncIterator[bytes]:
|
|
"""Async iterator yielding audio frames."""
|
|
while self._active:
|
|
frame = await self.read_frame()
|
|
if frame:
|
|
yield frame
|
|
|
|
def close(self):
|
|
"""Stop the tap."""
|
|
self._active = False
|
|
|
|
|
|
def make_capture_port(stream_id: str, sample_rate: int, channels: int, frame_ms: int):
|
|
"""Build a PJSUA2 media port that forks conference audio into taps.
|
|
|
|
Defined as a factory rather than a module-level class because
|
|
``pj.AudioMediaPort`` can only be subclassed once ``pjsua2`` imports —
|
|
and the whole pipeline degrades to stub mode when it doesn't.
|
|
|
|
The returned port is a *sink*: the conference bridge transmits into it,
|
|
and every frame is copied to each registered tap. Returns ``None`` when
|
|
pjsua2 is unavailable.
|
|
"""
|
|
try:
|
|
import pjsua2 as pj
|
|
except ImportError:
|
|
return None
|
|
|
|
class _CapturePort(pj.AudioMediaPort):
|
|
"""Receives conference-bridge frames and fans them out to taps.
|
|
|
|
``onFrameReceived`` is called on a **PJSUA2 worker thread** — a third
|
|
execution context alongside the asyncio loop and the Sippy ED thread.
|
|
It must touch nothing but ``AudioTap.feed``, which is explicitly
|
|
thread-safe (it hops to the owning loop via ``call_soon_threadsafe``).
|
|
Reaching into pipeline state, the event bus, or a Sippy object from
|
|
here would be a data race.
|
|
"""
|
|
|
|
def __init__(self, stream_id: str):
|
|
super().__init__()
|
|
self.stream_id = stream_id
|
|
self.taps: list[AudioTap] = []
|
|
self._logged_error = False
|
|
|
|
def onFrameReceived(self, frame): # noqa: N802 — PJSUA2 C++ callback name
|
|
try:
|
|
if not self.taps or frame.size <= 0:
|
|
return
|
|
# frame.buf is a SWIG ByteVector of signed chars; the tap
|
|
# contract is raw little-endian 16-bit PCM.
|
|
pcm = bytes(bytearray(b & 0xFF for b in frame.buf))
|
|
for tap in self.taps:
|
|
tap.feed(pcm)
|
|
except Exception as e:
|
|
# An exception escaping into PJSUA2's C++ callback would tear
|
|
# down the worker thread and silently kill media for every
|
|
# call. Log once per port rather than on every 20ms frame.
|
|
if not self._logged_error:
|
|
self._logged_error = True
|
|
logger.error(
|
|
f" Audio capture failed for {self.stream_id}: {e}",
|
|
exc_info=True,
|
|
)
|
|
|
|
fmt = pj.MediaFormatAudio()
|
|
fmt.init(
|
|
pj.PJMEDIA_FORMAT_L16,
|
|
sample_rate,
|
|
channels,
|
|
frame_ms * 1000, # frameTimeUsec
|
|
16, # bitsPerSample
|
|
)
|
|
port = _CapturePort(stream_id)
|
|
port.createPort(f"tap-{stream_id}", fmt)
|
|
return port
|
|
|
|
|
|
# ================================================================
|
|
# Stream Entry — tracks a single media stream in the pipeline
|
|
# ================================================================
|
|
|
|
class MediaStream:
|
|
"""Represents a single RTP media stream in the conference bridge."""
|
|
|
|
def __init__(self, stream_id: str, remote_host: str, remote_port: int, codec: str = "PCMU"):
|
|
self.stream_id = stream_id
|
|
self.remote_host = remote_host
|
|
self.remote_port = remote_port
|
|
self.codec = codec
|
|
self.conf_port: Optional[int] = None # PJSUA2 conference bridge port ID
|
|
self.transport = None # PJSUA2 SipTransport
|
|
self.media = None # PJSUA2 AudioMedia for this stream
|
|
self.capture_port = None # Shared _CapturePort feeding this stream's taps
|
|
self.rtp_port: Optional[int] = None # Local RTP listen port
|
|
self.taps: list[AudioTap] = []
|
|
self.recorder = None # PJSUA2 AudioMediaRecorder
|
|
self.player = None # PJSUA2 AudioMediaPlayer (active playback)
|
|
self.play_lock = asyncio.Lock() # Serializes playback per stream
|
|
self.active = True
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"<MediaStream {self.stream_id} "
|
|
f"rtp={self.remote_host}:{self.remote_port} "
|
|
f"conf_port={self.conf_port}>"
|
|
)
|
|
|
|
|
|
# ================================================================
|
|
# Main Pipeline
|
|
# ================================================================
|
|
|
|
class MediaPipeline:
|
|
"""
|
|
PJSUA2-based media pipeline.
|
|
|
|
Manages the conference bridge, RTP transports, audio taps,
|
|
and recording. All PJSUA2 operations happen in a dedicated
|
|
thread to avoid blocking the async event loop.
|
|
|
|
Usage:
|
|
pipeline = MediaPipeline()
|
|
await pipeline.start()
|
|
|
|
# Media arrives from the SIP engine's onCallMediaState callback
|
|
# (PJSUA2 only surfaces RTP media for a call it owns):
|
|
# pipeline.attach_call_media("leg_1", call.getAudioMedia(i))
|
|
|
|
# Tap audio for analysis — safe before or after media comes up
|
|
tap = pipeline.create_tap("leg_1")
|
|
async for frame in tap.stream():
|
|
classify(frame)
|
|
|
|
# Bridge two call legs
|
|
pipeline.bridge_streams("leg_1", "leg_2")
|
|
|
|
# Record a call
|
|
pipeline.start_recording("leg_1", "/tmp/call.wav")
|
|
|
|
await pipeline.stop()
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
rtp_start_port: int = 10000,
|
|
rtp_port_range: int = 1000,
|
|
sample_rate: int = 16000,
|
|
channels: int = 1,
|
|
null_audio: bool = True,
|
|
):
|
|
self._rtp_start_port = rtp_start_port
|
|
self._rtp_port_range = rtp_port_range
|
|
self._next_rtp_port = rtp_start_port
|
|
self._sample_rate = sample_rate
|
|
self._channels = channels
|
|
self._frame_ms = 20 # Must match medConfig.audioFramePtime below
|
|
self._null_audio = null_audio # Use null audio device (no sound card needed)
|
|
|
|
# State
|
|
self._streams: dict[str, MediaStream] = {}
|
|
self._taps: dict[str, list[AudioTap]] = {}
|
|
self._ready = False
|
|
|
|
# PJSUA2 objects (set during start)
|
|
self._endpoint = None
|
|
self._pjsua2_thread: Optional[threading.Thread] = None
|
|
self._lock = threading.Lock()
|
|
|
|
# ================================================================
|
|
# Lifecycle
|
|
# ================================================================
|
|
|
|
async def start(self) -> None:
|
|
"""Initialize PJSUA2 endpoint and conference bridge."""
|
|
logger.info("🎵 Starting PJSUA2 media pipeline...")
|
|
|
|
try:
|
|
import pjsua2 as pj
|
|
|
|
# Create and initialize the PJSUA2 Endpoint
|
|
ep = pj.Endpoint()
|
|
ep.libCreate()
|
|
|
|
# Configure endpoint
|
|
ep_cfg = pj.EpConfig()
|
|
|
|
# Log config
|
|
ep_cfg.logConfig.level = 3
|
|
ep_cfg.logConfig.consoleLevel = 3
|
|
|
|
# Media config
|
|
ep_cfg.medConfig.clockRate = self._sample_rate
|
|
ep_cfg.medConfig.channelCount = self._channels
|
|
ep_cfg.medConfig.audioFramePtime = 20 # 20ms frames
|
|
ep_cfg.medConfig.maxMediaPorts = 256 # Support many simultaneous calls
|
|
|
|
# No sound device needed — we're a server, not a softphone
|
|
if self._null_audio:
|
|
ep_cfg.medConfig.noVad = True
|
|
|
|
ep.libInit(ep_cfg)
|
|
|
|
# Use null audio device (no sound card)
|
|
if self._null_audio:
|
|
ep.audDevManager().setNullDev()
|
|
|
|
# Start the library
|
|
ep.libStart()
|
|
|
|
self._endpoint = ep
|
|
self._ready = True
|
|
|
|
logger.info(
|
|
f"🎵 PJSUA2 media pipeline ready "
|
|
f"(rate={self._sample_rate}Hz, ports=256, null_audio={self._null_audio})"
|
|
)
|
|
|
|
except ImportError:
|
|
logger.warning(
|
|
"⚠️ PJSUA2 not installed — media pipeline running in stub mode. "
|
|
"Install pjsip with Python bindings for real media handling."
|
|
)
|
|
self._ready = True
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ PJSUA2 initialization failed: {e}")
|
|
self._ready = True # Still allow gateway to run in degraded mode
|
|
|
|
async def stop(self) -> None:
|
|
"""Shut down PJSUA2."""
|
|
logger.info("🎵 Stopping PJSUA2 media pipeline...")
|
|
|
|
# Close all taps
|
|
for tap_list in self._taps.values():
|
|
for tap in tap_list:
|
|
tap.close()
|
|
self._taps.clear()
|
|
|
|
# Remove all streams (this releases their capture ports)
|
|
for stream_id in list(self._streams.keys()):
|
|
self.remove_stream(stream_id)
|
|
|
|
# Destroy PJSUA2 endpoint. Every media port must be collected first:
|
|
# a port finalised after libDestroy() runs pjmedia_conf_remove_port
|
|
# against a freed conference bridge and aborts the process. Dropping
|
|
# the last Python reference is not enough on its own — force the
|
|
# collection here rather than leaving it to interpreter exit.
|
|
gc.collect()
|
|
|
|
if self._endpoint:
|
|
try:
|
|
self._endpoint.libDestroy()
|
|
except Exception as e:
|
|
logger.error(f" PJSUA2 destroy error: {e}")
|
|
self._endpoint = None
|
|
|
|
self._ready = False
|
|
logger.info("🎵 PJSUA2 media pipeline stopped")
|
|
|
|
@property
|
|
def endpoint(self):
|
|
"""The PJSUA2 Endpoint, or None in stub mode.
|
|
|
|
PJSUA2 permits exactly one Endpoint per process, so the pipeline
|
|
creates it and the SIP engine borrows it rather than making a second.
|
|
"""
|
|
return self._endpoint
|
|
|
|
@property
|
|
def is_ready(self) -> bool:
|
|
return self._ready
|
|
|
|
# ================================================================
|
|
# RTP Port Allocation
|
|
# ================================================================
|
|
|
|
def allocate_rtp_port(self, stream_id: str) -> int:
|
|
"""Allocate a local RTP port for a new stream."""
|
|
with self._lock:
|
|
port = self._next_rtp_port
|
|
self._next_rtp_port += 2 # RTP uses even ports, RTCP uses odd
|
|
if self._next_rtp_port >= self._rtp_start_port + self._rtp_port_range:
|
|
self._next_rtp_port = self._rtp_start_port # Wrap around
|
|
return port
|
|
|
|
# ================================================================
|
|
# Stream Management
|
|
# ================================================================
|
|
|
|
def attach_call_media(self, stream_id: str, audio_media) -> Optional[int]:
|
|
"""Register a call's live ``AudioMedia`` with the pipeline.
|
|
|
|
Called from ``onCallMediaState`` on a PJSUA2 worker thread, which is
|
|
the only place PJSUA2 surfaces RTP-backed media. Any tap created
|
|
before this point is attached now; taps created later find the media
|
|
already present.
|
|
|
|
There is deliberately no ``add_remote_stream(host, port)`` counterpart:
|
|
PJSUA2 has no standalone RTP media object, so media can only arrive
|
|
from a call PJSUA2 owns. See ``docs/architecture.md``.
|
|
"""
|
|
stream = self._streams.get(stream_id)
|
|
if stream is None:
|
|
stream = MediaStream(stream_id, "", 0)
|
|
self._streams[stream_id] = stream
|
|
|
|
stream.media = audio_media
|
|
try:
|
|
stream.conf_port = audio_media.getPortId()
|
|
except Exception:
|
|
stream.conf_port = None
|
|
|
|
# Wire up taps that were requested before media came up.
|
|
pending = self._taps.get(stream_id, [])
|
|
if pending and stream.capture_port is None:
|
|
port = make_capture_port(
|
|
stream_id, self._sample_rate, self._channels, self._frame_ms
|
|
)
|
|
if port is not None:
|
|
try:
|
|
audio_media.startTransmit(port)
|
|
stream.capture_port = port
|
|
port.taps.extend(pending)
|
|
logger.info(
|
|
f" 🎤 Audio tap attached for {stream_id} "
|
|
f"({len(pending)} waiting)"
|
|
)
|
|
except Exception as e:
|
|
logger.error(
|
|
f" Failed to attach capture port for {stream_id}: {e}",
|
|
exc_info=True,
|
|
)
|
|
|
|
logger.info(f" 📡 Media attached for {stream_id} (conf port {stream.conf_port})")
|
|
return stream.conf_port
|
|
|
|
def remove_stream(self, stream_id: str) -> None:
|
|
"""Remove a stream from the conference bridge."""
|
|
stream = self._streams.pop(stream_id, None)
|
|
if not stream:
|
|
return
|
|
|
|
stream.active = False
|
|
|
|
# Close any taps
|
|
for tap in stream.taps:
|
|
tap.close()
|
|
self._taps.pop(stream_id, None)
|
|
|
|
# Release the capture port while the conference bridge still exists.
|
|
# A port garbage-collected after Endpoint.libDestroy() calls
|
|
# pjmedia_conf_remove_port against a freed bridge and aborts the
|
|
# process on a native assertion — a hard crash, not an exception.
|
|
if stream.capture_port is not None:
|
|
try:
|
|
if stream.media is not None:
|
|
stream.media.stopTransmit(stream.capture_port)
|
|
except Exception as e:
|
|
logger.debug(f" stopTransmit failed for {stream_id}: {e}")
|
|
stream.capture_port.taps.clear()
|
|
stream.capture_port = None
|
|
|
|
# Stop recording
|
|
if stream.recorder:
|
|
try:
|
|
stream.recorder = None # PJSUA2 will clean up
|
|
except Exception:
|
|
pass
|
|
|
|
logger.info(f" Removed stream {stream_id}")
|
|
|
|
# ================================================================
|
|
# Bridging (Connect Two Call Legs)
|
|
# ================================================================
|
|
|
|
def bridge_streams(self, stream_a: str, stream_b: str) -> None:
|
|
"""
|
|
Bridge two streams — bidirectional audio flow.
|
|
|
|
In PJSUA2 terms:
|
|
stream_a.startTransmit(stream_b)
|
|
stream_b.startTransmit(stream_a)
|
|
"""
|
|
a = self._streams.get(stream_a)
|
|
b = self._streams.get(stream_b)
|
|
|
|
if not a or not b:
|
|
logger.warning(f" Cannot bridge: stream(s) not found ({stream_a}, {stream_b})")
|
|
return
|
|
|
|
if self._endpoint and a.conf_port is not None and b.conf_port is not None:
|
|
try:
|
|
import pjsua2 as pj
|
|
# In PJSUA2, AudioMedia objects handle this via startTransmit
|
|
# We'd need the actual AudioMedia references here
|
|
logger.info(f" 🔗 Bridged {stream_a} (port {a.conf_port}) ↔ {stream_b} (port {b.conf_port})")
|
|
except Exception as e:
|
|
logger.error(f" Bridge error: {e}")
|
|
else:
|
|
logger.info(f" 🔗 Bridged {stream_a} ↔ {stream_b} (virtual)")
|
|
|
|
def unbridge_streams(self, stream_a: str, stream_b: str) -> None:
|
|
"""Disconnect two streams."""
|
|
a = self._streams.get(stream_a)
|
|
b = self._streams.get(stream_b)
|
|
|
|
if self._endpoint and a and b and a.conf_port is not None and b.conf_port is not None:
|
|
try:
|
|
logger.info(f" 🔓 Unbridged {stream_a} ↔ {stream_b}")
|
|
except Exception as e:
|
|
logger.error(f" Unbridge error: {e}")
|
|
else:
|
|
logger.info(f" 🔓 Unbridged {stream_a} ↔ {stream_b} (virtual)")
|
|
|
|
# ================================================================
|
|
# Audio Tapping (for Classifier + STT)
|
|
# ================================================================
|
|
|
|
def create_tap(self, stream_id: str) -> AudioTap:
|
|
"""
|
|
Create an audio tap on a stream.
|
|
|
|
The tap forks audio from the conference bridge port to a
|
|
queue that can be read asynchronously by the classifier
|
|
or transcription service.
|
|
|
|
Multiple taps per stream are supported (e.g., classifier + STT + recording).
|
|
"""
|
|
tap = AudioTap(stream_id, sample_rate=self._sample_rate)
|
|
stream = self._streams.get(stream_id)
|
|
|
|
if stream:
|
|
stream.taps.append(tap)
|
|
|
|
if stream_id not in self._taps:
|
|
self._taps[stream_id] = []
|
|
self._taps[stream_id].append(tap)
|
|
|
|
if self._endpoint and stream and stream.media is not None:
|
|
try:
|
|
# One capture port per stream, shared by every tap on it:
|
|
# the bridge would otherwise mix each additional port back
|
|
# into the conference and the call would echo.
|
|
if stream.capture_port is None:
|
|
port = make_capture_port(
|
|
stream_id, self._sample_rate, self._channels, self._frame_ms
|
|
)
|
|
if port is not None:
|
|
# The stream's media transmits into the capture port,
|
|
# not the reverse — the port is a sink.
|
|
stream.media.startTransmit(port)
|
|
stream.capture_port = port
|
|
logger.info(f" 🎤 Audio tap created for {stream_id} (PJSUA2)")
|
|
|
|
if stream.capture_port is not None:
|
|
stream.capture_port.taps.append(tap)
|
|
except Exception as e:
|
|
logger.error(
|
|
f" Failed to create PJSUA2 tap for {stream_id}: {e}", exc_info=True
|
|
)
|
|
else:
|
|
logger.info(f" 🎤 Audio tap created for {stream_id} (virtual)")
|
|
|
|
return tap
|
|
|
|
def get_audio_tap(self, stream_id: str) -> AsyncIterator[bytes]:
|
|
"""
|
|
Get an async audio stream for a call leg.
|
|
|
|
Creates a tap if one doesn't exist, then returns the
|
|
async iterator.
|
|
"""
|
|
taps = self._taps.get(stream_id, [])
|
|
if not taps:
|
|
tap = self.create_tap(stream_id)
|
|
else:
|
|
tap = taps[0]
|
|
return tap.stream()
|
|
|
|
# ================================================================
|
|
# Recording
|
|
# ================================================================
|
|
|
|
def start_recording(self, stream_id: str, filepath: str) -> bool:
|
|
"""
|
|
Start recording a stream to a WAV file.
|
|
|
|
Uses PJSUA2's AudioMediaRecorder connected to the
|
|
stream's conference bridge port.
|
|
"""
|
|
stream = self._streams.get(stream_id)
|
|
if not stream:
|
|
logger.warning(f" Cannot record: stream {stream_id} not found")
|
|
return False
|
|
|
|
if self._endpoint:
|
|
try:
|
|
import pjsua2 as pj
|
|
|
|
recorder = pj.AudioMediaRecorder()
|
|
recorder.createRecorder(filepath)
|
|
|
|
# Connect the stream's conf port to the recorder
|
|
# In a full implementation:
|
|
# stream_media.startTransmit(recorder)
|
|
|
|
stream.recorder = recorder
|
|
logger.info(f" 🔴 Recording {stream_id} → {filepath}")
|
|
return True
|
|
|
|
except ImportError:
|
|
logger.warning(f" PJSUA2 not available, recording to {filepath} (stub)")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f" Failed to start recording {stream_id}: {e}")
|
|
return False
|
|
else:
|
|
logger.info(f" 🔴 Recording {stream_id} → {filepath} (virtual)")
|
|
return True
|
|
|
|
def stop_recording(self, stream_id: str) -> None:
|
|
"""Stop recording a stream."""
|
|
stream = self._streams.get(stream_id)
|
|
if stream and stream.recorder:
|
|
# PJSUA2 will flush and close the WAV file
|
|
stream.recorder = None
|
|
logger.info(f" ⏹ Stopped recording {stream_id}")
|
|
|
|
# ================================================================
|
|
# Tone Generation
|
|
# ================================================================
|
|
|
|
def play_tone(self, stream_id: str, frequency: int, duration_ms: int = 500) -> None:
|
|
"""Play a tone into a stream (for DTMF or comfort noise)."""
|
|
if self._endpoint:
|
|
try:
|
|
import pjsua2 as pj
|
|
# Use pj.ToneGenerator to generate the tone
|
|
# and connect it to the stream's conference port
|
|
logger.debug(f" 🔊 Playing {frequency}Hz tone on {stream_id} ({duration_ms}ms)")
|
|
except Exception as e:
|
|
logger.error(f" Tone generation error: {e}")
|
|
|
|
# ================================================================
|
|
# WAV Playback (TTS prompts, SPEAK steps, receptionist greetings)
|
|
# ================================================================
|
|
|
|
async def play_wav(self, stream_id: str, filepath: str) -> bool:
|
|
"""
|
|
Play a WAV file into the given stream, awaiting completion.
|
|
|
|
Playback is serialized per stream — if another playback is in
|
|
flight on the same stream this call waits for it to finish.
|
|
Falls back to a duration-based sleep when PJSUA2 is unavailable.
|
|
"""
|
|
stream = self._streams.get(stream_id)
|
|
if not stream:
|
|
logger.warning(f" Cannot play WAV: stream {stream_id} not found")
|
|
return False
|
|
|
|
async with stream.play_lock:
|
|
duration_s = self._wav_duration_seconds(filepath)
|
|
|
|
if self._endpoint:
|
|
try:
|
|
import pjsua2 as pj
|
|
|
|
player = pj.AudioMediaPlayer()
|
|
# PJMEDIA_FILE_NO_LOOP == 1
|
|
player.createPlayer(filepath, 1)
|
|
stream.player = player
|
|
|
|
# In a full PJSUA2 integration:
|
|
# player.getAudioMedia().startTransmit(stream.audio_media)
|
|
# We don't hold the AudioMedia ref here in stub mode.
|
|
logger.info(
|
|
f" 🔊 Playing {filepath} on {stream_id} ({duration_s:.1f}s)"
|
|
)
|
|
|
|
await asyncio.sleep(duration_s)
|
|
stream.player = None
|
|
return True
|
|
|
|
except ImportError:
|
|
logger.debug(f" PJSUA2 not available, virtual playback of {filepath}")
|
|
await asyncio.sleep(duration_s)
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f" Failed to play {filepath} on {stream_id}: {e}")
|
|
stream.player = None
|
|
return False
|
|
else:
|
|
logger.info(f" 🔊 Playing {filepath} on {stream_id} (virtual, {duration_s:.1f}s)")
|
|
await asyncio.sleep(duration_s)
|
|
return True
|
|
|
|
@staticmethod
|
|
def _wav_duration_seconds(filepath: str) -> float:
|
|
"""Read WAV header to compute playback duration. Defaults to 2s on error."""
|
|
try:
|
|
import wave
|
|
|
|
with wave.open(filepath, "rb") as wf:
|
|
frames = wf.getnframes()
|
|
rate = wf.getframerate() or 16000
|
|
return frames / float(rate) if frames else 2.0
|
|
except Exception:
|
|
return 2.0
|
|
|
|
# ================================================================
|
|
# Status
|
|
# ================================================================
|
|
|
|
@property
|
|
def stream_count(self) -> int:
|
|
return len(self._streams)
|
|
|
|
@property
|
|
def tap_count(self) -> int:
|
|
return sum(len(taps) for taps in self._taps.values())
|
|
|
|
def status(self) -> dict:
|
|
"""Pipeline status for monitoring."""
|
|
return {
|
|
"ready": self._ready,
|
|
"pjsua2_available": self._endpoint is not None,
|
|
"streams": self.stream_count,
|
|
"taps": self.tap_count,
|
|
"rtp_port_range": f"{self._rtp_start_port}-{self._rtp_start_port + self._rtp_port_range}",
|
|
"sample_rate": self._sample_rate,
|
|
}
|