"""Persistent laptop prompt and manual voice transcription bar.""" from __future__ import annotations import json from pathlib import Path import re from typing import Any from PyQt5.QtCore import ( QEvent, QProcess, QTimer, Qt, ) from PyQt5.QtWidgets import ( QFrame, QHBoxLayout, QLabel, QLineEdit, QPushButton, QSizePolicy, ) from core.voice_privacy import voice_privacy_foundation ROOT = Path(__file__).resolve().parent.parent VOICE_PYTHON = ( Path.home() / ".venvs" / "shire-voice" / "bin" / "python" ) VOICE_MODEL = ( Path.home() / ".local" / "share" / "armor" / "voice" / "vosk-model-small-en-us-0.15" ) TRANSCRIBER = ( ROOT / "scripts" / "shire_live_transcribe.py" ) class GlobalPromptBar(QFrame): """Global keyboard and manual voice prompt entry.""" def __init__( self, stack, shire_page, parent=None, ): super().__init__(parent) self.stack = stack self.shire_page = shire_page self._stdout_buffer = "" self._voice_segments: list[str] = [] self._voice_partial = "" self._last_complete = "" self.setObjectName("globalPromptBar") self.setMinimumHeight(66) self.setMaximumHeight(72) self.setSizePolicy( QSizePolicy.Expanding, QSizePolicy.Fixed, ) self.setStyleSheet(""" QFrame#globalPromptBar { background-color: #03030c; border-bottom: 2px solid #9b4dff; } QLabel { background: transparent; border: none; font-family: "DejaVu Sans Mono", monospace; } QLineEdit { background-color: #070715; color: #f4efff; border: 2px solid #9b4dff; border-radius: 12px; font-family: "DejaVu Sans Mono", monospace; font-size: 15px; font-weight: 800; padding: 8px 14px; selection-background-color: #9b4dff; } QLineEdit:focus { border: 2px solid #00f58a; } QPushButton { background-color: #0b071c; color: #c89cff; border: 1px solid #9b4dff; border-radius: 11px; font-family: "DejaVu Sans Mono", monospace; font-size: 12px; font-weight: 900; padding: 8px 14px; } QPushButton:hover { background-color: #251044; color: #ffffff; } QPushButton:pressed { background-color: #073322; color: #00f58a; border-color: #00f58a; } """) layout = QHBoxLayout(self) layout.setContentsMargins(16, 9, 16, 9) layout.setSpacing(10) self.state_label = QLabel("● READY") self.state_label.setMinimumWidth(122) self.state_label.setAlignment( Qt.AlignLeft | Qt.AlignVCenter ) self.input = QLineEdit() self.input.setPlaceholderText( "Ask SHiRE… type here or press MIC and speak" ) self.input.setClearButtonEnabled(True) self.input.installEventFilter(self) self.input.returnPressed.connect( self.submit_prompt ) self.microphone_button = QPushButton("🎙 MIC") self.microphone_button.setMinimumWidth(104) self.microphone_button.clicked.connect( self.toggle_voice ) self.send_button = QPushButton("▶ SEND") self.send_button.setMinimumWidth(104) self.send_button.clicked.connect( self.submit_prompt ) self.home_button = QPushButton("⌂ HOME") self.home_button.setObjectName("globalHomeButton") self.home_button.setMinimumWidth(96) self.home_button.setToolTip( "Return to the SHiRE home page" ) self.home_button.clicked.connect( self.return_home ) layout.addWidget(self.home_button) layout.addWidget(self.state_label) layout.addWidget(self.input, 1) layout.addWidget(self.microphone_button) layout.addWidget(self.send_button) self.voice_process = QProcess(self) self.voice_process.setProcessChannelMode( QProcess.SeparateChannels ) self.voice_process.readyReadStandardOutput.connect( self._read_voice_output ) self.voice_process.readyReadStandardError.connect( self._read_voice_error ) self.voice_process.finished.connect( self._voice_finished ) self.brain_timer = QTimer(self) self.brain_timer.timeout.connect( self._refresh_brain_state ) self.brain_timer.start(500) self._set_state( "READY", "#00f58a", ) def return_home(self) -> None: """Return from any ARMOR station to SHiRE.""" self.stack.setCurrentIndex(1) self.input.setFocus() def eventFilter(self, watched, event): if ( watched is self.input and event.type() == QEvent.KeyPress ): if event.key() == Qt.Key_Escape: if self.voice_running(): self.stop_voice() else: self.input.clear() return True if event.key() == Qt.Key_F2: self.toggle_voice() return True return super().eventFilter( watched, event, ) def voice_running(self) -> bool: return ( self.voice_process.state() != QProcess.NotRunning ) def _set_state( self, text: str, colour: str, ) -> None: self.state_label.setText( f"● {str(text).upper()}" ) self.state_label.setStyleSheet( f"color:{colour}; " "font-size:11px; " "font-weight:900;" ) def toggle_voice(self) -> None: if self.voice_running(): self.stop_voice() else: self.start_voice() def start_voice(self) -> None: if not VOICE_PYTHON.is_file(): self._set_state( "VOICE ERROR", "#ff5c70", ) self.input.setPlaceholderText( "SHiRE voice environment is missing" ) return if not TRANSCRIBER.is_file(): self._set_state( "VOICE ERROR", "#ff5c70", ) self.input.setPlaceholderText( "ARMOR transcription helper is missing" ) return if not ( VOICE_MODEL / "am" / "final.mdl" ).is_file(): self._set_state( "VOICE ERROR", "#ff5c70", ) self.input.setPlaceholderText( "SHiRE speech model is missing" ) return self._stdout_buffer = "" self._voice_segments = [] self._voice_partial = "" self._last_complete = "" self._voice_prefix = self.input.text().strip() self.input.setPlaceholderText( "Listening… press STOP when finished" ) self.microphone_button.setText( "■ STOP" ) self._set_state( "LOADING", "#18e6d2", ) voice_privacy_foundation.set_runtime_state( "PROCESSING" ) self.voice_process.start( str(VOICE_PYTHON), [ str(TRANSCRIBER), "--model", str(VOICE_MODEL), "--max-seconds", "30", ], ) if not self.voice_process.waitForStarted( 3000 ): voice_privacy_foundation.set_runtime_state( "ERROR", "Unable to start local voice process.", ) self._set_state( "VOICE ERROR", "#ff5c70", ) self.microphone_button.setText( "🎙 MIC" ) def stop_voice(self) -> None: if not self.voice_running(): return self._set_state( "FINALISING", "#18e6d2", ) voice_privacy_foundation.set_runtime_state( "PROCESSING" ) self.voice_process.terminate() QTimer.singleShot( 2500, self._force_stop_voice, ) def _force_stop_voice(self) -> None: if self.voice_running(): self.voice_process.kill() def _read_voice_output(self) -> None: raw = bytes( self.voice_process.readAllStandardOutput() ).decode( "utf-8", errors="replace", ) self._stdout_buffer += raw while "\n" in self._stdout_buffer: line, self._stdout_buffer = ( self._stdout_buffer.split( "\n", 1, ) ) self._handle_voice_message( line.strip() ) def _read_voice_error(self) -> None: raw = bytes( self.voice_process.readAllStandardError() ).decode( "utf-8", errors="replace", ).strip() if raw: self.input.setToolTip(raw) def _handle_voice_message( self, line: str, ) -> None: if not line: return try: message: dict[str, Any] = ( json.loads(line) ) except json.JSONDecodeError: return message_type = str( message.get("type", "") ).lower() if message_type == "status": state = str( message.get("state", "") ).upper() if state == "LISTENING": voice_privacy_foundation.set_runtime_state( "LISTENING" ) self._set_state( "LISTENING", "#00f58a", ) elif state in { "LOADING", "PROCESSING", }: voice_privacy_foundation.set_runtime_state( "PROCESSING" ) self._set_state( state, "#18e6d2", ) return if message_type == "partial": transcript = str( message.get( "transcript", message.get("text", ""), ) ) self._voice_partial = transcript self.input.setText( " ".join( part for part in ( self._voice_prefix, self.normalise_transcript(transcript).strip(), ) if part ).strip() ) self.input.setCursorPosition( len(self.input.text()) ) return if message_type == "final": transcript = str( message.get( "transcript", message.get("text", ""), ) ).strip() if transcript: self._voice_segments = [ transcript ] self.input.setText( " ".join( part for part in ( self._voice_prefix, self.normalise_transcript(transcript).strip(), ) if part ).strip() ) self.input.setCursorPosition( len(self.input.text()) ) return if message_type == "complete": transcript = str( message.get( "transcript", "", ) ).strip() self._last_complete = transcript if transcript: self.input.setText( " ".join( part for part in ( self._voice_prefix, self.normalise_transcript(transcript).strip(), ) if part ).strip() ) self.input.setCursorPosition( len(self.input.text()) ) self._set_state( "READY TO SEND", "#00f58a", ) else: self._set_state( "NO WORDS", "#ffaa33", ) voice_privacy_foundation.set_runtime_state( "READY" ) return if message_type == "error": error = str( message.get( "message", "Unknown voice error", ) ) self.input.setToolTip(error) voice_privacy_foundation.set_runtime_state( "ERROR", error, ) self._set_state( "VOICE ERROR", "#ff5c70", ) def _voice_finished( self, exit_code: int, exit_status, ) -> None: del exit_status if self._stdout_buffer.strip(): self._handle_voice_message( self._stdout_buffer.strip() ) self._stdout_buffer = "" self.microphone_button.setText( "🎙 MIC" ) self.input.setPlaceholderText( "Ask SHiRE… type here or press MIC and speak" ) if exit_code != 0 and not self.input.text(): voice_privacy_foundation.set_runtime_state( "ERROR", f"Voice process exited with code {exit_code}.", ) self._set_state( "VOICE ERROR", "#ff5c70", ) elif self.input.text().strip(): voice_privacy_foundation.set_runtime_state( "READY" ) self._set_state( "READY TO SEND", "#00f58a", ) else: voice_privacy_foundation.set_runtime_state( "READY" ) self._set_state( "READY", "#00f58a", ) self.input.setFocus() def submit_prompt(self) -> None: if self.voice_running(): self.stop_voice() return prompt = self.input.text().strip() if not prompt: self._set_state( "PROMPT EMPTY", "#ffaa33", ) return self.stack.setCurrentIndex(1) try: self.shire_page.shire_append( f"\n> {prompt}" ) self.shire_page.ask_laptop_brain( prompt ) except Exception as exc: self._set_state( "SUBMIT ERROR", "#ff5c70", ) self.input.setToolTip( f"{type(exc).__name__}: {exc}" ) return self.input.clear() self._set_state( "THINKING", "#18e6d2", ) def _refresh_brain_state(self) -> None: if self.voice_running(): return worker = getattr( self.shire_page, "brain_worker", None, ) if ( worker is not None and worker.isRunning() ): self._set_state( "THINKING", "#18e6d2", ) return if ( "THINKING" in self.state_label.text() ): self._set_state( "READY", "#00f58a", ) @staticmethod def normalise_transcript( text: str, ) -> str: clean = " ".join( str(text).split() ).strip() clean = re.sub( r"^(hey|hello|hi|oi)\s+" r"(shire|share)\b[\s,]*", "", clean, flags=re.IGNORECASE, ) clean = re.sub( r"\b(shire|share)\b", "SHiRE", clean, flags=re.IGNORECASE, ) clean = re.sub( r"\barmor\b", "ARMOR", clean, flags=re.IGNORECASE, ) clean = re.sub( r"\barmor os\b", "ARMOR OS", clean, flags=re.IGNORECASE, ) return clean.strip()