""" Stream controller - manages input sources and orchestrates the render stream. """ from engine.config import Config, get_config from engine.effects.controller import handle_effects_command from engine.eventbus import EventBus from engine.events import EventType, StreamEvent from engine.mic import MicMonitor from engine.ntfy import NtfyPoller from engine.scroll import stream from engine.websocket_display import WebSocketDisplay def _get_display(config: Config): """Get the appropriate display based on config.""" if config.websocket: ws = WebSocketDisplay(host="0.0.0.0", port=config.websocket_port) ws.start_server() ws.start_http_server() return ws return None class StreamController: """Controls the stream lifecycle - initializes sources and runs the stream.""" _topics_warmed = False def __init__(self, config: Config | None = None, event_bus: EventBus | None = None): self.config = config or get_config() self.event_bus = event_bus self.mic: MicMonitor | None = None self.ntfy: NtfyPoller | None = None self.ntfy_cc: NtfyPoller | None = None @classmethod def warmup_topics(cls) -> None: """Warm up ntfy topics lazily (creates them if they don't exist).""" if cls._topics_warmed: return import urllib.request topics = [ "https://ntfy.sh/klubhaus_terminal_mainline_cc_cmd", "https://ntfy.sh/klubhaus_terminal_mainline_cc_resp", "https://ntfy.sh/klubhaus_terminal_mainline", ] for topic in topics: try: req = urllib.request.Request( topic, data=b"init", headers={ "User-Agent": "mainline/0.1", "Content-Type": "text/plain", }, method="POST", ) urllib.request.urlopen(req, timeout=5) except Exception: pass cls._topics_warmed = True def initialize_sources(self) -> tuple[bool, bool]: """Initialize microphone and ntfy sources. Returns: (mic_ok, ntfy_ok) - success status for each source """ self.mic = MicMonitor(threshold_db=self.config.mic_threshold_db) mic_ok = self.mic.start() if self.mic.available else False self.ntfy = NtfyPoller( self.config.ntfy_topic, reconnect_delay=self.config.ntfy_reconnect_delay, display_secs=self.config.message_display_secs, ) ntfy_ok = self.ntfy.start() self.ntfy_cc = NtfyPoller( self.config.ntfy_cc_cmd_topic, reconnect_delay=self.config.ntfy_reconnect_delay, display_secs=5, ) self.ntfy_cc.subscribe(self._handle_cc_message) ntfy_cc_ok = self.ntfy_cc.start() return bool(mic_ok), ntfy_ok and ntfy_cc_ok def _handle_cc_message(self, event) -> None: """Handle incoming C&C message - like a serial port control interface.""" import urllib.request cmd = event.body.strip() if hasattr(event, "body") else str(event).strip() if not cmd.startswith("/"): return response = handle_effects_command(cmd) topic_url = self.config.ntfy_cc_resp_topic.replace("/json", "") data = response.encode("utf-8") req = urllib.request.Request( topic_url, data=data, headers={"User-Agent": "mainline/0.1", "Content-Type": "text/plain"}, method="POST", ) try: urllib.request.urlopen(req, timeout=5) except Exception: pass def run(self, items: list) -> None: """Run the stream with initialized sources.""" if self.mic is None or self.ntfy is None: self.initialize_sources() if self.event_bus: self.event_bus.publish( EventType.STREAM_START, StreamEvent( event_type=EventType.STREAM_START, headline_count=len(items), ), ) display = _get_display(self.config) stream(items, self.ntfy, self.mic, display) if display: display.cleanup() if self.event_bus: self.event_bus.publish( EventType.STREAM_END, StreamEvent( event_type=EventType.STREAM_END, headline_count=len(items), ), ) def cleanup(self) -> None: """Clean up resources.""" if self.mic: self.mic.stop()