""" WebRTC Media Agent for Python This module provides WebRTC signaling server communication and peer connection management. Synthetic audio/video track creation is handled by the synthetic_media module. Test change to trigger reload - TESTING RELOAD NOW """ from __future__ import annotations import asyncio import json import websockets from typing import ( Dict, Optional, Callable, Awaitable, Protocol, AsyncIterator, cast, ) # test from dataclasses import dataclass, field from pydantic import ValidationError # types.SimpleNamespace removed — not used anymore after parsing candidates via aiortc.sdp import argparse import urllib.request import urllib.error import urllib.parse import ssl import sys import os # Import shared models sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from shared.models import ( SessionModel, LobbyCreateResponse, WebSocketMessageModel, JoinStatusModel, UserJoinedModel, LobbyStateModel, UpdateNameModel, AddPeerModel, RemovePeerModel, SessionDescriptionModel, IceCandidateModel, ICECandidateDictModel, SessionDescriptionTypedModel, ) from aiortc import ( RTCPeerConnection, RTCSessionDescription, RTCIceCandidate, MediaStreamTrack, ) from logger import logger from synthetic_media import create_synthetic_tracks, AnimatedVideoTrack # import debug_aioice # Generic message payload type MessageData = dict[str, object] class WebSocketProtocol(Protocol): def send(self, message: object, text: Optional[bool] = None) -> Awaitable[None]: ... def close(self, code: int = 1000, reason: str = "") -> Awaitable[None]: ... def __aiter__(self) -> AsyncIterator[str]: ... def _default_attributes() -> Dict[str, object]: return {} @dataclass class Peer: """Represents a WebRTC peer in the session""" session_id: str peer_name: str # Generic attributes bag. Values can be tracks or simple metadata. attributes: Dict[str, object] = field(default_factory=_default_attributes) muted: bool = False video_on: bool = True local: bool = False dead: bool = False connection: Optional[RTCPeerConnection] = None class WebRTCSignalingClient: """ WebRTC signaling client that communicates with the FastAPI signaling server. Handles peer-to-peer connection establishment and media streaming. """ def __init__( self, server_url: str, lobby_id: str, session_id: str, session_name: str, insecure: bool = False, ): self.server_url = server_url self.lobby_id = lobby_id self.session_id = session_id self.session_name = session_name self.insecure = insecure # WebSocket client protocol instance (typed as object to avoid Any) self.websocket: Optional[object] = None # Optional password to register or takeover a name self.name_password: Optional[str] = None self.peers: dict[str, Peer] = {} self.peer_connections: dict[str, RTCPeerConnection] = {} self.local_tracks: dict[str, MediaStreamTrack] = {} # State management self.is_negotiating: dict[str, bool] = {} self.making_offer: dict[str, bool] = {} self.initiated_offer: set[str] = set() self.pending_ice_candidates: dict[str, list[ICECandidateDictModel]] = {} # Event callbacks self.on_peer_added: Optional[Callable[[Peer], Awaitable[None]]] = None self.on_peer_removed: Optional[Callable[[Peer], Awaitable[None]]] = None self.on_track_received: Optional[ Callable[[Peer, MediaStreamTrack], Awaitable[None]] ] = None async def connect(self): """Connect to the signaling server""" ws_url = f"{self.server_url}/ws/lobby/{self.lobby_id}/{self.session_id}" logger.info(f"Connecting to signaling server: {ws_url}") # Log network information for debugging try: import socket hostname = socket.gethostname() local_ip = socket.gethostbyname(hostname) logger.info(f"Container hostname: {hostname}, local IP: {local_ip}") # Get all network interfaces import subprocess result = subprocess.run( ["ip", "addr", "show"], capture_output=True, text=True ) logger.info(f"Network interfaces:\n{result.stdout}") except Exception as e: logger.warning(f"Could not get network info: {e}") try: # If insecure (self-signed certs), create an SSL context for the websocket ws_ssl = None if self.insecure: ws_ssl = ssl.create_default_context() ws_ssl.check_hostname = False ws_ssl.verify_mode = ssl.CERT_NONE logger.info( f"Attempting websocket connection to {ws_url} with ssl={bool(ws_ssl)}" ) self.websocket = await websockets.connect(ws_url, ssl=ws_ssl) logger.info("Connected to signaling server") # Set up local media await self._setup_local_media() # Set name and join lobby name_payload: MessageData = {"name": self.session_name} if self.name_password: name_payload["password"] = self.name_password logger.info(f"Sending set_name: {name_payload}") await self._send_message("set_name", name_payload) logger.info("Sending join message") await self._send_message("join", {}) # Start message handling logger.info("Starting message handler loop") await self._handle_messages() except Exception as e: logger.error(f"Failed to connect to signaling server: {e}", exc_info=True) raise async def disconnect(self): """Disconnect from signaling server and cleanup""" if self.websocket: ws = cast(WebSocketProtocol, self.websocket) await ws.close() # Close all peer connections for pc in self.peer_connections.values(): await pc.close() # Stop local tracks for track in self.local_tracks.values(): track.stop() logger.info("Disconnected from signaling server") async def _setup_local_media(self): """Create local synthetic media tracks""" # Create synthetic tracks using the new module tracks = create_synthetic_tracks(self.session_name) self.local_tracks.update(tracks) # Add local peer to peers dict local_peer = Peer( session_id=self.session_id, peer_name=self.session_name, local=True, attributes={"tracks": self.local_tracks}, ) self.peers[self.session_id] = local_peer logger.info("Local synthetic media tracks created") async def _send_message( self, message_type: str, data: Optional[MessageData] = None ): """Send message to signaling server""" if not self.websocket: logger.error("No websocket connection") return # Build message with explicit type to avoid type narrowing message: dict[str, object] = {"type": message_type} if data is not None: message["data"] = data ws = cast(WebSocketProtocol, self.websocket) try: logger.debug(f"_send_message: Sending {message_type} with data: {data}") await ws.send(json.dumps(message)) logger.debug(f"_send_message: Sent message: {message_type}") except Exception as e: logger.error( f"_send_message: Failed to send {message_type}: {e}", exc_info=True ) async def _handle_messages(self): """Handle incoming messages from signaling server""" try: ws = cast(WebSocketProtocol, self.websocket) async for message in ws: logger.debug(f"_handle_messages: Received raw message: {message}") try: data = cast(MessageData, json.loads(message)) except Exception as e: logger.error( f"_handle_messages: Failed to parse message: {e}", exc_info=True ) continue await self._process_message(data) except websockets.exceptions.ConnectionClosed as e: logger.info(f"WebSocket connection closed: {e}") except Exception as e: logger.error(f"Error handling messages: {e}", exc_info=True) async def _process_message(self, message: MessageData): """Process incoming signaling messages""" try: # Validate the base message structure first validated_message = WebSocketMessageModel.model_validate(message) msg_type = validated_message.type data = validated_message.data except ValidationError as e: logger.error(f"Invalid message structure: {e}", exc_info=True) return logger.debug( f"_process_message: Received message type: {msg_type} with data: {data}" ) if msg_type == "addPeer": try: validated = AddPeerModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid addPeer payload: {e}", exc_info=True) return await self._handle_add_peer(validated) elif msg_type == "removePeer": try: validated = RemovePeerModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid removePeer payload: {e}", exc_info=True) return await self._handle_remove_peer(validated) elif msg_type == "sessionDescription": try: validated = SessionDescriptionModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid sessionDescription payload: {e}", exc_info=True) return await self._handle_session_description(validated) elif msg_type == "iceCandidate": try: validated = IceCandidateModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid iceCandidate payload: {e}", exc_info=True) return await self._handle_ice_candidate(validated) elif msg_type == "join_status": try: validated = JoinStatusModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid join_status payload: {e}", exc_info=True) return logger.info(f"Join status: {validated.status} - {validated.message}") elif msg_type == "user_joined": try: validated = UserJoinedModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid user_joined payload: {e}", exc_info=True) return logger.info( f"User joined: {validated.name} (session: {validated.session_id})" ) elif msg_type == "lobby_state": try: validated = LobbyStateModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid lobby_state payload: {e}", exc_info=True) return participants = validated.participants logger.info(f"Lobby state updated: {len(participants)} participants") elif msg_type == "update_name": try: validated = UpdateNameModel.model_validate(data) except ValidationError as e: logger.error(f"Invalid update payload: {e}", exc_info=True) return logger.info(f"Received update message: {validated}") else: logger.info(f"Unhandled message type: {msg_type} with data: {data}") async def _handle_add_peer(self, data: AddPeerModel): """Handle addPeer message - create new peer connection""" peer_id = data.peer_id peer_name = data.peer_name should_create_offer = data.should_create_offer logger.info( f"Adding peer: {peer_name} (should_create_offer: {should_create_offer})" ) logger.debug( f"_handle_add_peer: peer_id={peer_id}, peer_name={peer_name}, should_create_offer={should_create_offer}" ) # Check if peer already exists if peer_id in self.peer_connections: pc = self.peer_connections[peer_id] logger.debug( f"_handle_add_peer: Existing connection state: {pc.connectionState}" ) if pc.connectionState in ["new", "connected", "connecting"]: logger.info(f"Peer connection already exists for {peer_name}") return else: # Clean up stale connection logger.debug( f"_handle_add_peer: Closing stale connection for {peer_name}" ) await pc.close() del self.peer_connections[peer_id] # Create new peer peer = Peer(session_id=peer_id, peer_name=peer_name, local=False) self.peers[peer_id] = peer # Create RTCPeerConnection from aiortc.rtcconfiguration import RTCConfiguration, RTCIceServer config = RTCConfiguration( iceServers=[ RTCIceServer(urls="stun:ketrenos.com:3478"), RTCIceServer( urls="turns:ketrenos.com:5349", username="ketra", credential="ketran", ), # Add Google's public STUN server as fallback RTCIceServer(urls="stun:stun.l.google.com:19302"), ], ) logger.debug( f"_handle_add_peer: Creating RTCPeerConnection for {peer_name} with config: {config}" ) pc = RTCPeerConnection(configuration=config) # Add ICE gathering state change handler (explicit registration to satisfy static analyzers) def on_ice_gathering_state_change() -> None: logger.info(f"ICE gathering state: {pc.iceGatheringState}") # Debug: Check if we have any local candidates when gathering is complete if pc.iceGatheringState == "complete": logger.info( f"ICE gathering complete for {peer_name} - checking if candidates were generated..." ) pc.on("icegatheringstatechange")(on_ice_gathering_state_change) # Add connection state change handler (explicit registration to satisfy static analyzers) def on_connection_state_change() -> None: logger.info(f"Connection state: {pc.connectionState}") pc.on("connectionstatechange")(on_connection_state_change) self.peer_connections[peer_id] = pc peer.connection = pc # Set up event handlers def on_track(track: MediaStreamTrack) -> None: logger.info(f"Received {track.kind} track from {peer_name}") logger.info(f"on_track: {track.kind} from {peer_name}, track={track}") peer.attributes[f"{track.kind}_track"] = track if self.on_track_received: asyncio.ensure_future(self.on_track_received(peer, track)) pc.on("track")(on_track) def on_ice_candidate(candidate: Optional[RTCIceCandidate]) -> None: logger.info(f"on_ice_candidate: {candidate}") logger.info( f"on_ice_candidate CALLED for {peer_name}: candidate={candidate}" ) if not candidate: logger.info( f"on_ice_candidate: End of candidates signal for {peer_name}" ) return # Raw SDP fragment for the candidate raw = getattr(candidate, "candidate", None) # Try to infer candidate type from the SDP string (host/srflx/relay/prflx) def _parse_type(s: Optional[str]) -> str: if not s: return "eoc" import re m = re.search(r"\btyp\s+(host|srflx|relay|prflx)\b", s) return m.group(1) if m else "unknown" cand_type = _parse_type(raw) protocol = getattr(candidate, "protocol", "unknown") logger.info( f"ICE candidate outgoing for {peer_name}: type={cand_type} protocol={protocol} sdp={raw}" ) candidate_model = ICECandidateDictModel( candidate=raw, sdpMid=getattr(candidate, "sdpMid", None), sdpMLineIndex=getattr(candidate, "sdpMLineIndex", None), ) payload_model = IceCandidateModel( peer_id=peer_id, peer_name=peer_name, candidate=candidate_model ) logger.info( f"on_ice_candidate: Sending relayICECandidate for {peer_name}: {candidate_model}" ) asyncio.ensure_future( self._send_message("relayICECandidate", payload_model.model_dump()) ) pc.on("icecandidate")(on_ice_candidate) # Add local tracks for track in self.local_tracks.values(): logger.debug( f"_handle_add_peer: Adding local track {track.kind} to {peer_name}" ) pc.addTrack(track) # Create offer if needed if should_create_offer: self.initiated_offer.add(peer_id) self.making_offer[peer_id] = True self.is_negotiating[peer_id] = True try: logger.debug(f"_handle_add_peer: Creating offer for {peer_name}") offer = await pc.createOffer() logger.debug( f"_handle_add_peer: Offer created for {peer_name}: {offer}" ) await pc.setLocalDescription(offer) logger.debug(f"_handle_add_peer: Local description set for {peer_name}") # WORKAROUND for aiortc icecandidate event not firing (GitHub issue #1344) # Use Method 2: Complete SDP approach to extract ICE candidates logger.debug( f"_handle_add_peer: Waiting for ICE gathering to complete for {peer_name}" ) while pc.iceGatheringState != "complete": await asyncio.sleep(0.1) logger.debug( f"_handle_add_peer: ICE gathering complete, extracting candidates from SDP for {peer_name}" ) # Parse ICE candidates from the local SDP sdp_lines = pc.localDescription.sdp.split("\n") candidate_lines = [ line for line in sdp_lines if line.startswith("a=candidate:") ] # Track which media section we're in to determine sdpMid and sdpMLineIndex current_media_index = -1 current_mid = None for line in sdp_lines: if line.startswith("m="): # Media section current_media_index += 1 elif line.startswith("a=mid:"): # Media ID current_mid = line.split(":", 1)[1].strip() elif line.startswith("a=candidate:"): candidate_sdp = line[2:] # Remove 'a=' prefix candidate_model = ICECandidateDictModel( candidate=candidate_sdp, sdpMid=current_mid, sdpMLineIndex=current_media_index, ) payload_candidate = IceCandidateModel( peer_id=peer_id, peer_name=peer_name, candidate=candidate_model, ) logger.debug( f"_handle_add_peer: Sending extracted ICE candidate for {peer_name}: {candidate_sdp[:60]}..." ) await self._send_message( "relayICECandidate", payload_candidate.model_dump() ) # Send end-of-candidates signal (empty candidate) end_candidate_model = ICECandidateDictModel( candidate="", sdpMid=None, sdpMLineIndex=None, ) payload_end = IceCandidateModel( peer_id=peer_id, peer_name=peer_name, candidate=end_candidate_model ) logger.debug( f"_handle_add_peer: Sending end-of-candidates signal for {peer_name}" ) await self._send_message("relayICECandidate", payload_end.model_dump()) logger.debug( f"_handle_add_peer: Sent {len(candidate_lines)} ICE candidates to {peer_name}" ) session_desc_typed = SessionDescriptionTypedModel( type=offer.type, sdp=offer.sdp ) session_desc_model = SessionDescriptionModel( peer_id=peer_id, peer_name=peer_name, session_description=session_desc_typed, ) await self._send_message( "relaySessionDescription", session_desc_model.model_dump(), ) logger.info(f"Offer sent to {peer_name}") except Exception as e: logger.error( f"Failed to create/send offer to {peer_name}: {e}", exc_info=True ) finally: self.making_offer[peer_id] = False if self.on_peer_added: await self.on_peer_added(peer) async def _handle_remove_peer(self, data: RemovePeerModel): """Handle removePeer message""" peer_id = data.peer_id peer_name = data.peer_name logger.info(f"Removing peer: {peer_name}") # Close peer connection if peer_id in self.peer_connections: pc = self.peer_connections[peer_id] await pc.close() del self.peer_connections[peer_id] # Clean up state self.is_negotiating.pop(peer_id, None) self.making_offer.pop(peer_id, None) self.initiated_offer.discard(peer_id) self.pending_ice_candidates.pop(peer_id, None) # Remove peer peer = self.peers.pop(peer_id, None) if peer and self.on_peer_removed: await self.on_peer_removed(peer) async def _handle_session_description(self, data: SessionDescriptionModel): """Handle sessionDescription message""" peer_id = data.peer_id peer_name = data.peer_name session_description = data.session_description.model_dump() logger.info(f"Received {session_description['type']} from {peer_name}") pc = self.peer_connections.get(peer_id) if not pc: logger.error(f"No peer connection for {peer_name}") return desc = RTCSessionDescription( sdp=session_description["sdp"], type=session_description["type"] ) # Handle offer collision (polite peer pattern) making_offer = self.making_offer.get(peer_id, False) offer_collision = desc.type == "offer" and ( making_offer or pc.signalingState != "stable" ) we_initiated = peer_id in self.initiated_offer ignore_offer = we_initiated and offer_collision if ignore_offer: logger.info(f"Ignoring offer from {peer_name} due to collision") return try: await pc.setRemoteDescription(desc) self.is_negotiating[peer_id] = False logger.info(f"Remote description set for {peer_name}") # Process queued ICE candidates pending_candidates = self.pending_ice_candidates.pop(peer_id, []) from aiortc.sdp import candidate_from_sdp for candidate_data in pending_candidates: # candidate_data is an ICECandidateDictModel Pydantic model cand = candidate_data.candidate # handle end-of-candidates marker if not cand: await pc.addIceCandidate(None) logger.info(f"Added queued end-of-candidates for {peer_name}") continue # cand may be the full "candidate:..." string or the inner SDP part if cand and cand.startswith("candidate:"): sdp_part = cand.split(":", 1)[1] else: sdp_part = cand try: rtc_candidate = candidate_from_sdp(sdp_part) rtc_candidate.sdpMid = candidate_data.sdpMid rtc_candidate.sdpMLineIndex = candidate_data.sdpMLineIndex await pc.addIceCandidate(rtc_candidate) logger.info(f"Added queued ICE candidate for {peer_name}") except Exception as e: logger.error( f"Failed to add queued ICE candidate for {peer_name}: {e}" ) except Exception as e: logger.error(f"Failed to set remote description for {peer_name}: {e}") return # Create answer if this was an offer if session_description["type"] == "offer": try: answer = await pc.createAnswer() await pc.setLocalDescription(answer) # WORKAROUND for aiortc icecandidate event not firing (GitHub issue #1344) # Use Method 2: Complete SDP approach to extract ICE candidates logger.debug( f"_handle_session_description: Waiting for ICE gathering to complete for {peer_name} (answer)" ) while pc.iceGatheringState != "complete": await asyncio.sleep(0.1) logger.debug( f"_handle_session_description: ICE gathering complete, extracting candidates from SDP for {peer_name} (answer)" ) # Parse ICE candidates from the local SDP sdp_lines = pc.localDescription.sdp.split("\n") candidate_lines = [ line for line in sdp_lines if line.startswith("a=candidate:") ] # Track which media section we're in to determine sdpMid and sdpMLineIndex current_media_index = -1 current_mid = None for line in sdp_lines: if line.startswith("m="): # Media section current_media_index += 1 elif line.startswith("a=mid:"): # Media ID current_mid = line.split(":", 1)[1].strip() elif line.startswith("a=candidate:"): candidate_sdp = line[2:] # Remove 'a=' prefix candidate_model = ICECandidateDictModel( candidate=candidate_sdp, sdpMid=current_mid, sdpMLineIndex=current_media_index, ) payload_candidate = IceCandidateModel( peer_id=peer_id, peer_name=peer_name, candidate=candidate_model, ) logger.debug( f"_handle_session_description: Sending extracted ICE candidate for {peer_name} (answer): {candidate_sdp[:60]}..." ) await self._send_message( "relayICECandidate", payload_candidate.model_dump() ) # Send end-of-candidates signal (empty candidate) end_candidate_model = ICECandidateDictModel( candidate="", sdpMid=None, sdpMLineIndex=None, ) payload_end = IceCandidateModel( peer_id=peer_id, peer_name=peer_name, candidate=end_candidate_model ) logger.debug( f"_handle_session_description: Sending end-of-candidates signal for {peer_name} (answer)" ) await self._send_message("relayICECandidate", payload_end.model_dump()) logger.debug( f"_handle_session_description: Sent {len(candidate_lines)} ICE candidates to {peer_name} (answer)" ) session_desc_typed = SessionDescriptionTypedModel( type=answer.type, sdp=answer.sdp ) session_desc_model = SessionDescriptionModel( peer_id=peer_id, peer_name=peer_name, session_description=session_desc_typed, ) await self._send_message( "relaySessionDescription", session_desc_model.model_dump(), ) logger.info(f"Answer sent to {peer_name}") except Exception as e: logger.error(f"Failed to create/send answer to {peer_name}: {e}") async def _handle_ice_candidate(self, data: IceCandidateModel): """Handle iceCandidate message""" peer_id = data.peer_id peer_name = data.peer_name candidate_data = data.candidate logger.info(f"Received ICE candidate from {peer_name}") pc = self.peer_connections.get(peer_id) if not pc: logger.error(f"No peer connection for {peer_name}") return # Queue candidate if remote description not set if not pc.remoteDescription: logger.info( f"Remote description not set, queuing ICE candidate for {peer_name}" ) if peer_id not in self.pending_ice_candidates: self.pending_ice_candidates[peer_id] = [] # candidate_data is an ICECandidateDictModel Pydantic model self.pending_ice_candidates[peer_id].append(candidate_data) return try: from aiortc.sdp import candidate_from_sdp cand = candidate_data.candidate if not cand: # end-of-candidates await pc.addIceCandidate(None) logger.info(f"End-of-candidates added for {peer_name}") return if cand and cand.startswith("candidate:"): sdp_part = cand.split(":", 1)[1] else: sdp_part = cand # Detect type for logging try: import re m = re.search(r"\btyp\s+(host|srflx|relay|prflx)\b", sdp_part) cand_type = m.group(1) if m else "unknown" except Exception: cand_type = "unknown" try: rtc_candidate = candidate_from_sdp(sdp_part) rtc_candidate.sdpMid = candidate_data.sdpMid rtc_candidate.sdpMLineIndex = candidate_data.sdpMLineIndex # aiortc expects an object with attributes (RTCIceCandidate) await pc.addIceCandidate(rtc_candidate) logger.info(f"ICE candidate added for {peer_name}: type={cand_type}") except Exception as e: logger.error( f"Failed to add ICE candidate for {peer_name}: type={cand_type} error={e} sdp='{sdp_part}'", exc_info=True, ) except Exception as e: logger.error( f"Unexpected error handling ICE candidate for {peer_name}: {e}", exc_info=True, ) async def _handle_ice_connection_failure(self, peer_id: str, peer_name: str): """Handle ICE connection failure by logging details""" logger.info(f"ICE connection failure detected for {peer_name}") pc = self.peer_connections.get(peer_id) if not pc: logger.error( f"No peer connection found for {peer_name} during ICE failure recovery" ) return logger.error( f"ICE connection failed for {peer_name}. Connection state: {pc.connectionState}, ICE state: {pc.iceConnectionState}" ) # In a real implementation, you might want to notify the user or attempt reconnection async def _schedule_ice_timeout(self, peer_id: str, peer_name: str): """Schedule a timeout for ICE connection checking""" await asyncio.sleep(30) # Wait 30 seconds pc = self.peer_connections.get(peer_id) if not pc: return if pc.iceConnectionState == "checking": logger.warning( f"ICE connection timeout for {peer_name} - still in checking state after 30 seconds" ) logger.warning( f"Final connection state: {pc.connectionState}, ICE state: {pc.iceConnectionState}" ) logger.warning( "This might be due to network connectivity issues between the browser and Docker container" ) logger.warning( "Consider checking: 1) Port forwarding 2) TURN server config 3) Docker network mode" ) elif pc.iceConnectionState in ["failed", "closed"]: logger.info( f"ICE connection for {peer_name} resolved to: {pc.iceConnectionState}" ) else: logger.info( f"ICE connection for {peer_name} established: {pc.iceConnectionState}" ) # Example usage def _http_base_url(server_url: str) -> str: # Convert ws:// or wss:// to http(s) and ensure no trailing slash if server_url.startswith("ws://"): return "http://" + server_url[len("ws://") :].rstrip("/") if server_url.startswith("wss://"): return "https://" + server_url[len("wss://") :].rstrip("/") return server_url.rstrip("/") def _ws_url(server_url: str) -> str: # Convert http(s) to ws(s) if needed if server_url.startswith("http://"): return "ws://" + server_url[len("http://") :].rstrip("/") if server_url.startswith("https://"): return "wss://" + server_url[len("https://") :].rstrip("/") return server_url.rstrip("/") def create_or_get_session( server_url: str, session_id: str | None = None, insecure: bool = False ) -> str: """Call GET /api/session to obtain a session_id (unless one was provided). Uses urllib so no extra runtime deps are required. """ if session_id: return session_id http_base = _http_base_url(server_url) url = f"{http_base}/api/session" req = urllib.request.Request(url, method="GET") # Prepare SSL context if requested (accept self-signed certs) ssl_ctx = None if insecure: ssl_ctx = ssl.create_default_context() ssl_ctx.check_hostname = False ssl_ctx.verify_mode = ssl.CERT_NONE try: with urllib.request.urlopen(req, timeout=10, context=ssl_ctx) as resp: body = resp.read() data = json.loads(body) # Validate response shape using Pydantic try: session = SessionModel.model_validate(data) except ValidationError as e: raise RuntimeError(f"Invalid session response from {url}: {e}") sid = session.id if not sid: raise RuntimeError(f"No session id returned from {url}: {data}") return sid except urllib.error.HTTPError as e: raise RuntimeError(f"HTTP error getting session: {e}") except Exception as e: raise RuntimeError(f"Error getting session: {e}") def create_or_get_lobby( server_url: str, session_id: str, lobby_name: str, private: bool = False, insecure: bool = False, ) -> str: """Call POST /api/lobby/{session_id} to create or lookup a lobby by name. Returns the lobby id. """ http_base = _http_base_url(server_url) url = f"{http_base}/api/lobby/{urllib.parse.quote(session_id)}" payload = json.dumps( { "type": "lobby_create", "data": {"name": lobby_name, "private": private}, } ).encode("utf-8") req = urllib.request.Request( url, data=payload, headers={"Content-Type": "application/json"}, method="POST" ) # Prepare SSL context if requested (accept self-signed certs) ssl_ctx = None if insecure: ssl_ctx = ssl.create_default_context() ssl_ctx.check_hostname = False ssl_ctx.verify_mode = ssl.CERT_NONE try: with urllib.request.urlopen(req, timeout=10, context=ssl_ctx) as resp: body = resp.read() data = json.loads(body) # Expect shape: { "type": "lobby_created", "data": {"id":..., ...}} try: lobby_resp = LobbyCreateResponse.model_validate(data) except ValidationError as e: raise RuntimeError(f"Invalid lobby response from {url}: {e}") lobby_id = lobby_resp.data.id if not lobby_id: raise RuntimeError(f"No lobby id returned from {url}: {data}") return lobby_id except urllib.error.HTTPError as e: # Try to include response body for.infoging try: body = e.read() msg = body.decode("utf-8", errors="ignore") except Exception: msg = str(e) raise RuntimeError(f"HTTP error creating lobby: {msg}") except Exception as e: raise RuntimeError(f"Error creating lobby: {e}") async def main(): """Example usage of the WebRTC signaling client with CLI options to create session and lobby.""" parser = argparse.ArgumentParser(description="Python WebRTC voicebot client") parser.add_argument( "--server-url", default="http://localhost:8000/ai-voicebot", help="AI-Voicebot lobby and signaling server base URL (http:// or https://)", ) parser.add_argument( "--lobby", default="default", help="Lobby name to create or join" ) parser.add_argument( "--session-name", default="Python Bot", help="Session (user) display name" ) parser.add_argument( "--session-id", default=None, help="Optional existing session id to reuse" ) parser.add_argument( "--password", default=None, help="Optional password to register or takeover a name", ) parser.add_argument( "--private", action="store_true", help="Create the lobby as private" ) parser.add_argument( "--insecure", action="store_true", help="Allow insecure server connections when using SSL (accept self-signed certs)", ) args = parser.parse_args() # Resolve session id (create if needed) try: session_id = create_or_get_session( args.server_url, args.session_id, insecure=args.insecure ) print(f"Using session id: {session_id}") except Exception as e: print(f"Failed to get/create session: {e}") return # Create or get lobby id try: lobby_id = create_or_get_lobby( args.server_url, session_id, args.lobby, args.private, insecure=args.insecure, ) print(f"Using lobby id: {lobby_id} (name={args.lobby})") except Exception as e: print(f"Failed to create/get lobby: {e}") return # Build websocket base URL (ws:// or wss://) from server_url and pass to client so # it constructs the final websocket path (/ws/lobby/{lobby}/{session}) itself. ws_base = _ws_url(args.server_url) client = WebRTCSignalingClient( ws_base, lobby_id, session_id, args.session_name, insecure=args.insecure ) # Set up event handlers async def on_peer_added(peer: Peer): print(f"Peer added: {peer.peer_name}") async def on_peer_removed(peer: Peer): print(f"Peer removed: {peer.peer_name}") # Remove any video tracks from this peer from our synthetic video track if "video" in client.local_tracks: synthetic_video_track = client.local_tracks["video"] if isinstance(synthetic_video_track, AnimatedVideoTrack): # We need to identify and remove tracks from this specific peer # Since we don't have a direct mapping, we'll need to track this differently # For now, this is a placeholder - we might need to enhance the peer tracking logger.info( f"Peer {peer.peer_name} removed - may need to clean up video tracks" ) async def on_track_received(peer: Peer, track: MediaStreamTrack): print(f"Received {track.kind} track from {peer.peer_name}") # If it's a video track, attach it to our synthetic video track for edge detection if track.kind == "video" and "video" in client.local_tracks: synthetic_video_track = client.local_tracks["video"] if isinstance(synthetic_video_track, AnimatedVideoTrack): synthetic_video_track.add_remote_video_track(track) logger.info( f"Attached remote video track from {peer.peer_name} to synthetic video track" ) client.on_peer_added = on_peer_added client.on_peer_removed = on_peer_removed client.on_track_received = on_track_received try: # Connect and run # If a password was provided on the CLI, store it on the client for use when setting name if args.password: client.name_password = args.password await client.connect() except KeyboardInterrupt: print("Shutting down...") finally: await client.disconnect() if __name__ == "__main__": # Install required packages: # pip install aiortc websockets opencv-python numpy asyncio.run(main()) # test modification # Test comment Mon Sep 1 03:48:19 PM PDT 2025 # Test change at Mon Sep 1 03:52:13 PM PDT 2025