Feat/gst consumer capture time

PR_DESCRIPTION_GST_CONSUMER.md

@@ -0,0 +1,63 @@
+# Consume the synchronized A+V stream + stamp ASR audio at capture
+
+Pairs OM1 with the OM1-video-processor GStreamer pipeline, which serves a single
+muxed A+V RTSP session (`rtsp://localhost:8555/live`) on one clock. Two parts:
+config (repoint consumers) and code (stamp audio capture time at ingest).
+
+## Config — repoint consumers (non-breaking)
+
+`config/unitree_g1_conversation.json5` is env-parametrized; **defaults preserve
+current behavior exactly**:
+
+- ASR: `type: "${ASR_INPUT_PLUGIN:-GoogleASRInput}"` + `rtsp_url:
+  "${ASR_RTSP_URL:-rtsp://localhost:8555/live}"` (the local plugin ignores the
+  extra key).
+- VLM: `rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8554/top_camera_raw}"`
+  (unchanged default).
+
+`docker-compose.yml` exposes `ASR_INPUT_PLUGIN`, `ASR_RTSP_URL`, `VLM_RTSP_URL`.
+To consume the GStreamer stream:
+
+```bash
+ASR_INPUT_PLUGIN=GoogleASRRTSPInput \
+ASR_RTSP_URL=rtsp://localhost:8555/live \
+VLM_RTSP_URL=rtsp://localhost:8555/live \
+docker compose up -d om1
+```
+
+Both consumers read the same muxed URL; each ffmpeg selects its track
+(`google_asr_rtsp` uses `-vn`, `video_rtsp_stream` uses `-an`), so audio and
+video come from one synchronized source.
+
+## Code — stamp ASR audio at capture
+
+Previously the audio timestamp was set at *package/send* time
+(`packageAudio` → `time.Now().UnixMilli()`), discarding capture timing. Now:
+
+- `packageAudio(pcm, captureMs)` stamps the provided capture time.
+- `sendChunkAt(pcm, capture)` added; `sendChunk` kept as a `time.Now()` wrapper
+  for back-compat (riva/elevenlabs/parallel unchanged — trivially extendable).
+- `google_asr.go` (local mic) and `google_asr_rtsp.go` (RTSP) stamp
+  `time.Now()` at the moment the buffer/chunk is read and pass it through.
+- Tests updated to pass and assert the capture timestamp.
+
+This makes ASR chunks carry capture time, so downstream alignment with
+video-derived features works on a common timeline.
+
+## Deliberately out of scope (follow-ups)
+
+- **Transcript→capture-time mapping across the cloud round-trip.** The final
+  transcript still reaches the IO layer without a capture timestamp
+  (`asr_common.go` `AddInput(..., time.Time{})`); mapping it back to the source
+  audio window needs the ASR WS protocol to echo timestamps.
+- **Video RTSP PTS.** JPEG-over-`image2pipe` carries no per-frame PTS; true
+  capture-time for video needs a PTS-preserving transport (documented at the
+  stamp site in `video_rtsp_stream.go`).
+- Extend `sendChunkAt` to the riva/elevenlabs paths (one-line each).
+
+## Verification
+
+Config validated as JSON5; compose validated as YAML. The Go changes are
+mechanical (caller consistency checked: `sendChunk` still present; only the two
+`packageAudio` test call sites updated) but were **not** compiled in this
+environment — run `go build ./... && go test ./plugins/inputs/asr/...` in CI.

README.md

@@ -53,7 +53,7 @@ brew install portaudio ffmpeg
 For Linux:
 ```bash
 sudo apt-get update
-sudo apt-get install -y portaudio19-dev ffmpeg
+sudo apt-get install -y portaudio19-dev ffmpeg pkg-config
 ```
 
 > [!TIP]

config/unitree_g1_conversation.json5

@@ -80,16 +80,29 @@ You should prioritize safe, comfortable, and positive human interaction.",
       hertz: 0.001,
       agent_inputs: [
         {
-          type: "GoogleASRInput",
+          // Defaults preserve legacy behavior (local PortAudio mic). To consume
+          // the synchronized A+V stream from the GStreamer video-processor, set
+          // ASR_INPUT_PLUGIN=GoogleASRRTSPInput and
+          // ASR_RTSP_URL=rtsp://localhost:8555/live. GoogleASRInput ignores the
+          // extra rtsp_url key, so the default is harmless.
+          type: "${ASR_INPUT_PLUGIN:-GoogleASRInput}",
           config: {
             api_version: "v2",
             enable_tts_interrupt: true,
             rate: 16000,
             chunk: 1600,
+            rtsp_url: "${ASR_RTSP_URL:-rtsp://localhost:8555/live}",
           },
         },
         {
+          // VLM consumes the video-processor's RAW (pre-CV, no overlays/blur)
+          // camera view directly — the clean scene gives the best descriptions.
+          // Override with VLM_RTSP_URL (e.g. rtsp://localhost:8555/live for the
+          // processed/blurred view).
           type: "VLMGeminiRTSP",
+          config: {
+            rtsp_url: "${VLM_RTSP_URL:-rtsp://localhost:8556/raw}",
+          },
         }
       ],
       cortex_llm: {

docker-compose.yml

@@ -13,6 +13,16 @@ services:
       - OM_API_KEY=${OM_API_KEY}
       - OM1_COMMAND=${OM1_COMMAND:-unitree_g1_conversation}
       - OM1_SKIP_INTERNET_CHECK=${OM1_SKIP_INTERNET_CHECK:-false}
+      # ASR/VLM source selection. ASR default is the local mic (standalone-safe);
+      # VLM defaults to the video-processor's raw camera view. When paired with
+      # the GStreamer video-processor, consume its streams directly (low latency;
+      # mediamtx is for people, not the agent):
+      #   ASR_INPUT_PLUGIN=GoogleASRRTSPInput   (audio from :8555/live, -vn)
+      #   ASR_RTSP_URL=rtsp://localhost:8555/live
+      #   VLM_RTSP_URL=rtsp://localhost:8556/raw   (clean scene for the VLM)
+      - ASR_INPUT_PLUGIN=${ASR_INPUT_PLUGIN:-GoogleASRInput}
+      - ASR_RTSP_URL=${ASR_RTSP_URL:-rtsp://localhost:8555/live}
+      - VLM_RTSP_URL=${VLM_RTSP_URL:-rtsp://localhost:8556/raw}
       - XDG_RUNTIME_DIR=${XDG_RUNTIME_DIR}
       - PULSE_SERVER=unix:${XDG_RUNTIME_DIR}/pulse/native
       - PULSE_COOKIE=/root/.config/pulse/cookie

docs/SUMMARY.md

@@ -13,6 +13,7 @@
     * [Project Structure](developing/7_project_structure.md)
     * [Configuration](developing/3_configuration.md)
     * [Inputs](developing/4_inputs.md)
+    * [Video Processor Integration](developing/video_processor_integration.md)
     * [LLMs](developing/5_llms.md)
     * [Actions](developing/6_actions.md)
     * [Backgrounds](developing/8_backgrounds.md)

docs/developing/video_processor_integration.md

@@ -0,0 +1,97 @@
+# Video Processor Integration (audio & video sources)
+
+On a robot (e.g. NVIDIA Thor), OM1 runs alongside the **OM1 Video Processor**, a
+separate container that owns the camera and microphone. This page explains how
+OM1 consumes that data, and — importantly — the trade-offs behind the defaults
+so you can change them deliberately later.
+
+## Topology
+
+The video-processor is the capture authority. Its GStreamer pipeline captures
+camera + mic on one clock and serves, on the robot host:
+
+| Endpoint | Contents | Who it's for |
+| --- | --- | --- |
+| `rtsp://localhost:8555/live` | Processed video (recognition boxes, blur) **+ audio**, synchronized | machine consumers (OM1, cloud relay) — low latency |
+| `rtsp://localhost:8556/raw` | **Raw** camera view (no overlays/blur), video only | machine consumers wanting a clean image |
+| `mediamtx` `:8554` (RTSP) / `:8888` (HLS) / `:8889` (WebRTC) | Re-serves `/live` and `/raw` | people / external clients / browsers |
+
+**OM1 consumes the gst endpoints directly (`:8555`/`:8556`), not mediamtx.**
+mediamtx is an on-demand fan-out hub for humans; routing the agent's real-time
+inputs through it would add a relay hop and couple OM1's core perception to an
+optional convenience service. Keep the agent on the direct, low-latency source;
+leave mediamtx for people.
+
+## How OM1 consumes it
+
+Set via env (see `docker-compose.yml` and `config/unitree_g1_conversation.json5`):
+
+| Variable | Default | Meaning |
+| --- | --- | --- |
+| `ASR_INPUT_PLUGIN` | `GoogleASRInput` | ASR source plugin (local mic vs RTSP) |
+| `ASR_RTSP_URL` | `rtsp://localhost:8555/live` | Audio source when using the RTSP ASR plugin (ffmpeg selects the audio track with `-vn`) |
+| `VLM_RTSP_URL` | `rtsp://localhost:8556/raw` | Video source for the VLM |
+
+### VLM video source — why `/raw`
+
+The VLM describes the scene for the LLM. `/live` has recognition boxes and
+**blurred faces** burned into the pixels; feeding that to the VLM degrades its
+descriptions. `/raw` is the clean camera view, so it gives the best scene
+understanding — hence the default.
+
+Trade-off to know: `/raw` is **not** anonymized. If your deployment must keep
+faces blurred even in what's sent to the (cloud) VLM for privacy reasons, set
+`VLM_RTSP_URL=rtsp://localhost:8555/live` instead and accept the description
+quality hit.
+
+### ASR source — why local mic is the default
+
+ASR is env-selectable between the local mic (`GoogleASRInput`, opens the mic via
+PortAudio) and RTSP (`GoogleASRRTSPInput`, pulls audio from `:8555/live`). The
+default is the **local mic**. Reasoning:
+
+- **Standalone-safe.** OM1 runs without the video-processor (dev laptops, other
+  robots, `conversation.json5`). A local-mic default doesn't break when there's
+  no `:8555`.
+- **Lower latency & decoupled.** Direct PortAudio capture avoids the RTSP +
+  ffmpeg decode hop, and OM1's hearing doesn't depend on the video-processor
+  being up.
+- **The double mic capture is harmless** (see below).
+
+Switch to RTSP (`ASR_INPUT_PLUGIN=GoogleASRRTSPInput`, `ASR_RTSP_URL=
+rtsp://localhost:8555/live`) when you specifically want a **single audio
+authority** — one capture, one AEC path, and OM1's transcripts aligned to the
+*exact* audio that the video-processor records and streams to the cloud. The
+cost is coupling OM1's ASR to the video-processor (startup ordering; goes deaf
+if `:8555` is down) and a small latency hop.
+
+Rule of thumb: timing-critical / sample-consistent audio work belongs in the
+video-processor (which already has the single capture-stamped audio for
+recording, features, and cloud). OM1's own job — hear the user, respond fast —
+is best served by the decoupled local mic. Flip to RTSP only when a concrete
+need for shared/consistent audio appears.
+
+## Can OM1 and the video-processor both use the microphone at once?
+
+**Yes — because access goes through PulseAudio.** `default_mic_aec` is a
+PulseAudio (echo-cancelled) *virtual source*, and PulseAudio sources are not
+exclusive: it duplicates the stream to every client. Both containers mount the
+host Pulse socket (`PULSE_SERVER=unix:$XDG_RUNTIME_DIR/pulse/native`), so OM1's
+PortAudio→Pulse capture and the video-processor's `pulsesrc`/ffmpeg capture read
+the same source concurrently without conflict.
+
+Caveat: this only holds via PulseAudio. If either side opened the **raw ALSA**
+device (`hw:X`) directly, ALSA hardware devices are exclusive and the second
+opener would fail with "device busy" (unless using `dmix`/`dsnoop`). Everything
+here is Pulse-based, so concurrent access is fine.
+
+## Video timestamps in OM1
+
+OM1's RTSP video consumer stamps each frame with local **receive time**
+(`time.Now()`), not the original capture time — the JPEG-over-pipe transport
+carries no per-frame PTS. This is intentional: OM1 fuses VLM output as coarse
+"recent context", so sub-second pipeline jitter is irrelevant, and any
+frame-accurate A/V work lives in the video-processor (capture-stamped at the
+source). If OM1 ever needs true capture time, switch the consumer to an RTSP
+client that exposes RTP/RTCP timing (e.g. `gortsplib`) instead of the
+ffmpeg→JPEG pipe. See `internal/providers/vlm/video_rtsp_stream.go`.

internal/actions/schema.go

@@ -49,7 +49,7 @@ func BuildSchemaForAction(actionName, llmLabel string) (map[string]any, bool) {
 // tag overrides the auto-generated description.
 func BuildSchema(llmLabel, description string, inputExample any) map[string]any {
 	inputType := reflect.TypeOf(inputExample)
-	if inputType.Kind() == reflect.Ptr {
+	if inputType.Kind() == reflect.Pointer {
 		inputType = inputType.Elem()
 	}
 

internal/providers/vlm/video_rtsp_stream.go

@@ -14,7 +14,7 @@ import (
 )
 
 const (
-	defaultRTSPURL     = "rtsp://localhost:8554/live"
+	defaultRTSPURL     = "rtsp://localhost:8556/raw"
 	defaultRTSPWidth   = 480
 	defaultRTSPHeight  = 640
 	rtspReconnectDelay = 2 * time.Second
@@ -124,11 +124,24 @@ func (v *VideoRTSPStream) stream(ctx context.Context) error {
 		if ctx.Err() != nil {
 			return false
 		}
-		v.send(Frame{Timestamp: time.Now(), JPEG: frame})
+		v.emitFrame(frame)
 		return true
 	})
 }
 
+// emitFrame forwards a decoded JPEG stamped with its receive time.
+//
+// Receive-time stamp is by design: this is when OM1 got the decoded frame, not
+// the original capture instant — JPEG-over-image2pipe carries no per-frame RTP
+// PTS. That's fine here: OM1 fuses VLM output as coarse "recent context", so
+// sub-second pipeline jitter doesn't matter. Any timing-critical, frame-accurate
+// A/V work lives in the video-processor (capture-stamped at the source). If OM1
+// ever needs true capture time, switch this consumer to an RTSP client that
+// exposes RTP/RTCP timing (e.g. gortsplib) rather than the ffmpeg->JPEG pipe.
+func (v *VideoRTSPStream) emitFrame(frame []byte) {
+	v.send(Frame{Timestamp: time.Now(), JPEG: frame})
+}
+
 // GrabFrame captures a single frame from the RTSP source using ffmpeg and returns it as a JPEG-encoded byte slice.
 func GrabFrame(ctx context.Context, cfg VideoRTSPStreamConfig) ([]byte, error) {
 	v := NewVideoRTSPStream(cfg)

internal/providers/vlm/video_rtsp_stream_test.go

@@ -0,0 +1,47 @@
+package vlm
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestVideoRTSPDefaultURLValue pins the actual default endpoint string to the
+// video-processor's clean (pre-CV) camera view. The sibling
+// TestNewVideoRTSPStreamDefaults asserts against the defaultRTSPURL constant, so
+// this guards the constant's value itself against accidental changes.
+func TestVideoRTSPDefaultURLValue(t *testing.T) {
+	require.Equal(t, "rtsp://localhost:8556/raw", defaultRTSPURL)
+
+	v := NewVideoRTSPStream(VideoRTSPStreamConfig{})
+	require.Equal(t, "rtsp://localhost:8556/raw", v.cfg.RTSPURL)
+}
+
+// TestNewVideoRTSPStreamPreservesExplicitURL covers the non-default branch: a
+// caller-supplied URL must not be overwritten.
+func TestNewVideoRTSPStreamPreservesExplicitURL(t *testing.T) {
+	v := NewVideoRTSPStream(VideoRTSPStreamConfig{RTSPURL: "rtsp://example.test/live"})
+	require.Equal(t, "rtsp://example.test/live", v.cfg.RTSPURL)
+}
+
+// TestEmitFrameStampsReceiveTime verifies each decoded frame is forwarded with a
+// receive-time stamp and its JPEG payload intact.
+func TestEmitFrameStampsReceiveTime(t *testing.T) {
+	v := NewVideoRTSPStream(VideoRTSPStreamConfig{})
+	v.out = make(chan Frame, 1) // stand in for the channel start() would create
+
+	jpeg := []byte{0xFF, 0xD8, 0xFF, 0xD9}
+	before := time.Now()
+	v.emitFrame(jpeg)
+	after := time.Now()
+
+	select {
+	case f := <-v.out:
+		require.Equal(t, jpeg, f.JPEG)
+		require.False(t, f.Timestamp.Before(before), "timestamp must not precede the call")
+		require.False(t, f.Timestamp.After(after), "timestamp must not follow the call")
+	default:
+		t.Fatal("expected emitFrame to enqueue a frame")
+	}
+}

plugins/backgrounds/vlm/vlm.go

@@ -23,8 +23,9 @@ func init() {
 }
 
 const (
-	defaultFPS     = 10
-	defaultRTSPURL = "rtsp://localhost:8554/top_camera_raw"
+	defaultFPS = 10
+	// GStreamer video-processor's raw (pre-CV) camera view; see inputs/vlm.
+	defaultRTSPURL = "rtsp://localhost:8556/raw"
 
 	vlmRestartDelay = 2 * time.Second
 )

plugins/inputs/asr/asr_common.go

@@ -284,7 +284,26 @@ func newASRCommon(cfg asrCommonConfig) *asrCommon {
 func (c *asrCommon) connect() error { return c.stream.connect() }
 
 // sendChunk forwards a PCM chunk to the single stream's websocket.
-func (c *asrCommon) sendChunk(pcm []byte) { c.stream.sendChunk(pcm) }
+//func (c *asrCommon) sendChunk(pcm []byte) { c.stream.sendChunk(pcm) }
+
+// sendChunkAt forwards a PCM chunk with its capture time to the single stream.
+func (c *asrCommon) sendChunkAt(pcm []byte, capture time.Time) {
+	c.stream.sendChunkAt(pcm, capture)
+}
+
+// forwardChunk copies a freshly-read PCM chunk and sends it stamped at its
+// capture time. The chunk is dropped (and false returned) when TTS is currently
+// speaking and interruption is disabled. The copy is required because callers
+// reuse their read buffer for the next chunk.
+func (c *asrCommon) forwardChunk(buf []byte, capture time.Time) bool {
+	if tts.Speaking.Load() && !c.enableTTSInterrupt {
+		return false
+	}
+	pcm := make([]byte, len(buf))
+	copy(pcm, buf)
+	c.sendChunkAt(pcm, capture)
+	return true
+}
 
 // statsLoop logs the single stream's send statistics until ctx is cancelled.
 func (c *asrCommon) statsLoop(ctx context.Context) { c.stream.statsLoop(ctx) }

plugins/inputs/asr/asr_stream.go

@@ -87,12 +87,16 @@ func (s *transcriberStream) closeWS() {
 }
 
 // packageAudio prepends the JSON audio header (length-prefixed) to a PCM chunk.
-func (s *transcriberStream) packageAudio(pcm []byte) ([]byte, error) {
+// captureMs is the capture time (Unix ms) of this chunk, stamped at ingest so
+// the timestamp reflects when the audio was captured rather than when it was
+// packaged/sent. This lets downstream consumers align audio with video-derived
+// features on a common timeline.
+func (s *transcriberStream) packageAudio(pcm []byte, captureMs int64) ([]byte, error) {
 	meta := AudioMetadata{
 		Rate:                     s.rate,
 		LanguageCode:             s.languageCode,
 		AlternativeLanguageCodes: s.altCodes,
-		Timestamp:                time.Now().UnixMilli(),
+		Timestamp:                captureMs,
 	}
 
 	headerBytes, err := json.Marshal(meta)
@@ -108,9 +112,16 @@ func (s *transcriberStream) packageAudio(pcm []byte) ([]byte, error) {
 	return packet, nil
 }
 
-// sendChunk packages and sends a PCM chunk over the websocket, updating statistics.
-func (s *transcriberStream) sendChunk(pcm []byte) {
-	packet, err := s.packageAudio(pcm)
+// sendChunk packages and sends a PCM chunk, stamping the capture time as now.
+// Callers that know the true acoustic capture instant should use sendChunkAt.
+// func (s *transcriberStream) sendChunk(pcm []byte) {
+// 	s.sendChunkAt(pcm, time.Now())
+// }
+
+// sendChunkAt packages and sends a PCM chunk captured at the given time,
+// updating statistics.
+func (s *transcriberStream) sendChunkAt(pcm []byte, capture time.Time) {
+	packet, err := s.packageAudio(pcm, capture.UnixMilli())
 	if err != nil {
 		s.log.Warn("package error", zap.Error(err))
 		return