logs for streaming

Author: githubdoramon

backend/robot-gateway/capture_relay.py

@@ -159,6 +159,9 @@ class RelaySession:
     last_audio_chunk_at: datetime | None = None
     last_upstream_activity_at: datetime | None = None
     latest_video_frame: bytes | None = None
+    video_frame_count: int = 0
+    audio_chunk_count: int = 0
+    audio_bytes_received: int = 0
     audio_buffer: deque[bytes] = field(
         default_factory=lambda: deque(maxlen=CAPTURE_RELAY_AUDIO_BUFFER_CHUNKS)
     )
@@ -310,6 +313,7 @@ async def handle_upstream_json(self, session_id: str, message: dict[str, Any]) -
                 session.video_meta = message
             elif message_type == "audio_meta":
                 session.audio_meta = message
+                logger.info("[capture_relay] Audio metadata session=%s meta=%s", session_id, message)
             elif message_type == "goodbye":
                 session.last_error = str(message.get("reason") or "upstream_goodbye")
 
@@ -329,11 +333,23 @@ async def handle_upstream_binary(self, session_id: str, payload: bytes) -> None:
             if track_type == VIDEO_TRACK:
                 session.latest_video_frame = chunk
                 session.last_video_frame_at = utc_now()
+                session.video_frame_count += 1
                 viewers = list(session.video_viewers.values())
             elif track_type == AUDIO_TRACK:
                 session.audio_buffer.append(chunk)
                 session.last_audio_chunk_at = utc_now()
+                session.audio_chunk_count += 1
+                session.audio_bytes_received += len(chunk)
                 viewers = list(session.audio_viewers.values())
+                if session.audio_chunk_count == 1 or session.audio_chunk_count % 100 == 0:
+                    logger.info(
+                        "[capture_relay] Audio chunk received session=%s chunks=%d bytes=%d last_bytes=%d viewers=%d",
+                        session_id,
+                        session.audio_chunk_count,
+                        session.audio_bytes_received,
+                        len(chunk),
+                        len(viewers),
+                    )
             else:
                 logger.warning("[capture_relay] Unknown track=%s session=%s", track_type, session_id)
                 return
@@ -370,9 +386,12 @@ async def register_audio_viewer(self, session_id: str) -> tuple[RelaySession, Re
             session.audio_viewers[viewer.viewer_id] = viewer
             backlog = list(session.audio_buffer)
             logger.info(
-                "[capture_relay] Audio viewer joined session=%s viewers=%d",
+                "[capture_relay] Audio viewer joined session=%s viewers=%d backlog=%d chunks_received=%d last_audio=%s",
                 session_id,
                 session.viewer_count,
+                len(backlog),
+                session.audio_chunk_count,
+                session.last_audio_chunk_at.isoformat() if session.last_audio_chunk_at else None,
             )
             return session, viewer, backlog
 

backend/robot-gateway/routes/capture.py

@@ -103,19 +103,39 @@ async def frame_stream():
     @router.websocket("/api/capture/streams/{session_id}/audio.pcm")
     async def stream_audio_pcm(websocket: WebSocket, session_id: str):
         viewer_token = websocket.query_params.get("token", "")
+        logger.info(
+            "[capture_relay] Audio viewer upgrade session=%s client=%s has_token=%s header_keys=%s",
+            session_id,
+            websocket.client,
+            "yes" if viewer_token else "no",
+            sorted(websocket.headers.keys()),
+        )
         if viewer_token:
             try:
                 claims = validate_relay_token(viewer_token)
-            except HTTPException:
+            except HTTPException as exc:
+                logger.warning(
+                    "[capture_relay] Audio viewer token rejected session=%s status=%s detail=%s",
+                    session_id,
+                    exc.status_code,
+                    exc.detail,
+                )
                 await websocket.close(code=4401)
                 return
             if (
                 claims.get("session_id") != session_id
                 or claims.get("tracks") != ["audio"]
             ):
+                logger.warning(
+                    "[capture_relay] Audio viewer token scope rejected session=%s claims_session=%s tracks=%s",
+                    session_id,
+                    claims.get("session_id"),
+                    claims.get("tracks"),
+                )
                 await websocket.close(code=4403)
                 return
         else:
+            logger.warning("[capture_relay] Audio viewer rejected session=%s reason=missing_token", session_id)
             await websocket.close(code=4401)
             return
 
@@ -124,18 +144,55 @@ async def stream_audio_pcm(websocket: WebSocket, session_id: str):
         session, viewer, backlog = await relay.register_audio_viewer(session_id)
         if not session.audio_enabled:
             await relay.unregister_viewer(session_id, viewer, mqtt)
+            logger.warning("[capture_relay] Audio viewer rejected session=%s reason=audio_disabled", session_id)
             await websocket.close(code=4400, reason="Audio disabled")
             return
 
         await websocket.accept()
+        logger.info(
+            "[capture_relay] Audio viewer accepted session=%s viewer=%s backlog=%d",
+            session_id,
+            viewer.viewer_id,
+            len(backlog),
+        )
+        chunks_sent = 0
+        bytes_sent = 0
         try:
             for chunk in backlog:
                 await websocket.send_bytes(chunk)
+                chunks_sent += 1
+                bytes_sent += len(chunk)
             while True:
                 chunk = await viewer.queue.get()
                 await websocket.send_bytes(chunk)
+                chunks_sent += 1
+                bytes_sent += len(chunk)
+                if chunks_sent == 1 or chunks_sent % 100 == 0:
+                    logger.info(
+                        "[capture_relay] Audio viewer send session=%s viewer=%s chunks=%d bytes=%d last_bytes=%d",
+                        session_id,
+                        viewer.viewer_id,
+                        chunks_sent,
+                        bytes_sent,
+                        len(chunk),
+                    )
         except WebSocketDisconnect:
-            pass
+            logger.info(
+                "[capture_relay] Audio viewer disconnected session=%s viewer=%s chunks=%d bytes=%d",
+                session_id,
+                viewer.viewer_id,
+                chunks_sent,
+                bytes_sent,
+            )
+        except Exception as exc:
+            logger.warning(
+                "[capture_relay] Audio viewer send failed session=%s viewer=%s chunks=%d bytes=%d error=%s",
+                session_id,
+                viewer.viewer_id,
+                chunks_sent,
+                bytes_sent,
+                exc,
+            )
         finally:
             await relay.unregister_viewer(session_id, viewer, mqtt)
 

frontend/web/src/app/robots/stream-test/page.tsx

@@ -116,7 +116,11 @@ export default function RobotStreamTestPage() {
     const interval = window.setInterval(async () => {
       try {
         const next = await gw<CaptureStream>(statusPath);
-        setSession(next);
+        setSession((current) =>
+          current?.session_id === next.session_id
+            ? { ...next, viewer_paths: current.viewer_paths }
+            : next
+        );
       } catch {
         setSession(null);
       }
@@ -125,9 +129,7 @@ export default function RobotStreamTestPage() {
     return () => window.clearInterval(interval);
   }, [sessionId, statusPath]);
 
-  const mjpgUrl = session
-    ? `${GW_BASE}${session.viewer_paths.camera_mjpg}?t=${encodeURIComponent(session.last_video_frame_at ?? session.created_at)}`
-    : null;
+  const mjpgUrl = session ? `${GW_BASE}${session.viewer_paths.camera_mjpg}` : null;
 
   useEffect(() => {
     if (!session?.audio_enabled || !session.viewer_paths.audio_pcm_ws_public) {
@@ -152,23 +154,55 @@ export default function RobotStreamTestPage() {
     socket.binaryType = "arraybuffer";
     audioSocketRef.current = socket;
     setAudioStatus("connecting");
+    console.info("[stream-test] audio websocket connecting", {
+      session_id: session.session_id,
+      url: session.viewer_paths.audio_pcm_ws_public,
+    });
 
     const sampleRate = Number(session.audio_meta?.sample_rate_hz ?? 16000) || 16000;
     const channelCount = Number(session.audio_meta?.channels ?? 1) || 1;
+    let receivedChunks = 0;
+    let receivedBytes = 0;
 
-    socket.onopen = () => setAudioStatus("live");
-    socket.onclose = () => {
+    socket.onopen = () => {
+      console.info("[stream-test] audio websocket open", { session_id: session.session_id });
+      setAudioStatus("live");
+    };
+    socket.onclose = (event) => {
+      console.info("[stream-test] audio websocket closed", {
+        session_id: session.session_id,
+        code: event.code,
+        reason: event.reason,
+        wasClean: event.wasClean,
+        receivedChunks,
+        receivedBytes,
+      });
       if (audioSocketRef.current === socket) {
         audioSocketRef.current = null;
         setAudioStatus("disconnected");
       }
     };
-    socket.onerror = () => setAudioStatus("error");
+    socket.onerror = (event) => {
+      console.warn("[stream-test] audio websocket error", { session_id: session.session_id, event });
+      setAudioStatus("error");
+    };
     socket.onmessage = (event) => {
       if (!(event.data instanceof ArrayBuffer)) return;
 
       const pcm = new Int16Array(event.data);
       if (!pcm.length) return;
+      receivedChunks += 1;
+      receivedBytes += event.data.byteLength;
+      if (receivedChunks === 1 || receivedChunks % 100 === 0) {
+        console.info("[stream-test] audio chunk received", {
+          session_id: session.session_id,
+          chunks: receivedChunks,
+          bytes: receivedBytes,
+          lastBytes: event.data.byteLength,
+          sampleRate,
+          channelCount,
+        });
+      }
 
       const frameCount = Math.floor(pcm.length / channelCount);
       const audioBuffer = context.createBuffer(channelCount, frameCount, sampleRate);
@@ -196,6 +230,7 @@ export default function RobotStreamTestPage() {
     };
   }, [
     session?.audio_enabled,
+    session?.session_id,
     session?.viewer_paths.audio_pcm_ws_public,
     session?.audio_meta?.sample_rate_hz,
     session?.audio_meta?.channels,