Spaces:
Running
Running
| import logging | |
| import re | |
| import io | |
| import wave | |
| # WARNING: Holding 60 seconds of uncompressed ambient audio in active memory | |
| # has privacy and memory implications. This is an explicit tradeoff to enable | |
| # the instant "emergency replay" feature. | |
| EMERGENCY_BUFFER_WINDOW_SECONDS = 60 | |
| # Assuming 16000Hz, 16-bit PCM, Mono | |
| # 16000 * 2 bytes = 32000 bytes per second | |
| BYTES_PER_SECOND = 32000 | |
| MAX_BUFFER_BYTES = EMERGENCY_BUFFER_WINDOW_SECONDS * BYTES_PER_SECOND | |
| class LiveAudioBuffer: | |
| def __init__(self): | |
| self.buffer = bytearray() | |
| def append(self, data: bytes): | |
| self.buffer.extend(data) | |
| if len(self.buffer) > MAX_BUFFER_BYTES: | |
| # Slice to keep only the most recent bytes | |
| self.buffer = self.buffer[-MAX_BUFFER_BYTES:] | |
| async def get_ring_buffer_tail(self, seconds_back: int) -> bytes: | |
| bytes_needed = seconds_back * BYTES_PER_SECOND | |
| if len(self.buffer) >= bytes_needed: | |
| return bytes(self.buffer[-bytes_needed:]) | |
| return bytes(self.buffer) | |
| # Global active recording session buffer | |
| active_recording_session = LiveAudioBuffer() | |
| async def get_live_audio_buffer(seconds_back: int = 30) -> bytes: | |
| """Pulls the last N seconds directly from the in-memory ring buffer.""" | |
| return await active_recording_session.get_ring_buffer_tail(seconds_back) | |
| def matches_emergency_replay_intent(transcript: str) -> bool: | |
| """Uses lightweight keyword spotting to trap emergency playback requests.""" | |
| if not transcript: | |
| return False | |
| target = transcript.lower() | |
| phrases = ["emergency replay", "play back the last", "what just happened", "play that back"] | |
| return any(phrase in target for phrase in phrases) | |
| def extract_duration(transcript: str, default: int = 30) -> int: | |
| """Extracts requested seconds from the transcript, e.g., 'play back the last 15 seconds'.""" | |
| if not transcript: | |
| return default | |
| match = re.search(r"(?:last|past)\s+(\d+)\s+seconds", transcript.lower()) | |
| if match: | |
| try: | |
| return int(match.group(1)) | |
| except ValueError: | |
| pass | |
| return default | |
| async def play_audio_immediately(audio_bytes: bytes): | |
| """ | |
| Plays back the RAW 16kHz PCM audio immediately through the native pygame mixer. | |
| Bypasses TTS entirely. | |
| """ | |
| from backend.voice.activation import _play_tts_bytes | |
| if not audio_bytes: | |
| logging.warning("No audio to playback in emergency replay.") | |
| return | |
| wav_io = io.BytesIO() | |
| with wave.open(wav_io, 'wb') as wf: | |
| wf.setnchannels(1) | |
| wf.setsampwidth(2) | |
| wf.setframerate(16000) | |
| wf.writeframes(audio_bytes) | |
| wav_bytes = wav_io.getvalue() | |
| logging.info(f"Playing back {len(audio_bytes) / BYTES_PER_SECOND} seconds of RAW emergency audio.") | |
| # _play_tts_bytes expects a WAV payload and handles the pygame playback safely | |
| await _play_tts_bytes(wav_bytes) | |