Skip to content

Architecture

Overview

Dive Detector is a four-layer pipeline:

  • A collection layer pulls a curated set of sensor keys from each ship's NATS JetStream KV bucket and mirrors them into Redis.
  • A detection layer reads that cache on a configurable interval, combines the sensor signals with operator toggles, and decides whether OBS should be streaming.
  • A control layer issues OBS commands via WebSocket and writes its observed health/state back to Redis.
  • A presentation layer (SvelteKit dashboard) renders both the sensor data and the detector's observed state, and writes the operator toggles back to Redis.
flowchart TD
    NATS["NATS JetStream\ncoredata-{ship}.{domain}.mbari.org:4222"]
    NC["navproc_collector\nallow-list from collector_keys.toml"]
    REDIS[("Redis\nshared cache")]
    DD["dive_detector\npolls + drives OBS"]
    OBS["OBS Studio\n{obs-host}:4455"]
    STREAM["Live Stream Platform"]
    FE["SvelteKit dashboard\n:3000"]
    USER(["Operator browser"])

    NATS -->|"KV get (configured keys only)"| NC
    NC -->|"SET {ship}:{key}"| REDIS
    REDIS -->|"GET sensors + config"| DD
    DD -->|"GetStreamStatus / Start / Stop / Scene"| OBS
    DD -->|"SET state:streaming / obs_ok / obs_last_error"| REDIS
    REDIS <-->|"GET sensors + state / SET config"| FE
    FE -->|"HTTP"| USER
    OBS -->|"RTMP/SRT"| STREAM

Components

Runs continuously alongside each ship instance. It loads an allow-list of NATS KV keys from backend/collector_keys.toml once at startup, then on every poll fetches only those keys from the ship's bucket and writes them to Redis. Earlier versions called get_keys() every poll and mirrored everything in the bucket — that was wasteful and noisy. Today the collector reads O(allow-list) keys per poll, and per-key fetch failures are isolated so a missing or malformed key doesn't kill the batch.

Key naming: dots in NATS key names become colons; the bucket name is prefixed; the whole thing is lowercased.

NATS key:   ROV.CTD.INWATER
Redis key:  rcsn:rov:ctd:inwater

All Redis keys written by the collector carry a 600-second TTL. If NATS goes away, sensor keys expire and the detector treats missing values as "not diving."

dive_detector

Polls Redis at a configurable interval (default: 10 s). Each cycle it:

  1. Reads {ship}:rov:ctd:inwater and {ship}:rov:mbari:depth to compute dive_status (in water AND below threshold).
  2. Reads the operator toggles {ship}:config:auto_stream and {ship}:config:force_stream.
  3. Probes OBS with GetStreamStatus — a side-effect-free request that serves as the comms heartbeat AND returns OBS's actual streaming state. Writes the probe result to {ship}:state:obs_ok and any error to {ship}:state:obs_last_error.
  4. Computes the desired state via the control table below.
  5. Sends SetCurrentProgramScene to the appropriate scene, and only issues StartStream/StopStream when OBS's actual state differs from the desired state (avoids the OBS v5 error responses for redundant start/stop calls).
  6. Writes {ship}:state:streaming to reflect the desired state.

Control table

The detector always drives OBS to a definite state when --obs-host is configured:

force_stream auto_stream dive_status OBS scene Stream
ON any any MainOnly start
OFF ON yes MainOnly start
OFF ON no FeedOffline stop
OFF OFF any FeedOffline stop

obs_control

A thin wrapper around obs-websocket-py that exposes three action methods (change_scene, start_stream, stop_stream) and one probe (get_stream_status). Action methods return bool and stash the most recent failure on last_error so the detector can surface it to the dashboard. The wrapper transparently reconnects when its socket is dead on entry, and treats "already in target state" responses as success.

Redis

Standard Redis instance shared by all services. No persistence is required — sensor values are regenerated from NATS, operator toggles are the only durable user-controlled state.

Frontend dashboard

SvelteKit + Tailwind dashboard served from a Node container on port 3000. The browser never talks to Redis directly — the SvelteKit server-side routes (+server.ts) use ioredis to bridge HTTP/JSON ↔ Redis. The dashboard polls /api/state every 3 s, renders one card per ship with status pills, telemetry, a map, and the two toggles, and posts toggle changes to /api/ships/{ship}/auto-stream and /api/ships/{ship}/force-stream. See Frontend Dashboard for details.

Data flow detail

sequenceDiagram
    participant NATS
    participant Collector as navproc_collector
    participant Redis
    participant Detector as dive_detector
    participant OBS
    participant FE as SvelteKit
    participant User as Operator

    loop Every poll_interval (collector)
        Collector->>NATS: kv.get(K) for K in collector_keys.toml
        NATS-->>Collector: {ROV.CTD.INWATER: "1", ROV.MBARI.DEPTH: "47.3", ...}
        Collector->>Redis: SET {ship}:rov:ctd:inwater "1" EX 600
        Collector->>Redis: SET {ship}:rov:mbari:depth "47.3" EX 600
    end

    loop Every poll_interval (detector)
        Detector->>Redis: GET sensor + config keys
        Redis-->>Detector: values
        Detector->>OBS: GetStreamStatus  (heartbeat + real state)
        OBS-->>Detector: outputActive=true|false
        alt should_stream != currently_streaming
            Detector->>OBS: SetCurrentProgramScene + Start/StopStream
        end
        Detector->>Redis: SET state:streaming / obs_ok / obs_last_error
    end

    loop Every 3s (dashboard)
        User->>FE: GET /
        FE->>Redis: pipeline GET of all sensor/state/config keys
        Redis-->>FE: values
        FE-->>User: rendered card with pills, map, toggles
    end

    User->>FE: POST /api/ships/rcsn/force-stream {enabled: true}
    FE->>Redis: SET rcsn:config:force_stream "1"

Deployment topology

All services run in a single Docker Compose stack, sharing a redisnet bridge network.

┌────────────────────────────────────────────────────┐
│  Docker Compose                                    │
│                                                    │
│  ┌─────────┐    ┌────────────────────────┐         │
│  │  Redis  │    │  frontend (port 3000)  │         │
│  └────┬────┘    └─────────┬──────────────┘         │
│       │                   │                        │
│       └───── redisnet ────┴────┬────────────────┐  │
│  ┌──────────────┐  ┌────────────────┐           │  │
│  │  navproc     │  │  navproc       │           │  │
│  │  collector   │  │  collector     │           │  │
│  │  (rcsn)      │  │  (dpkd)        │           │  │
│  └──────────────┘  └────────────────┘           │  │
│  ┌──────────────┐  ┌────────────────┐           │  │
│  │  dive-       │  │  dive-         │           │  │
│  │  detector    │  │  detector      │           │  │
│  │  (rcsn)      │  │  (dpkd)        │           │  │
│  └──────┬───────┘  └────────┬───────┘           │  │
│         │                   │                   │  │
└─────────┼───────────────────┼───────────────────┘
          ▼                   ▼
       OBS (rc-teleops)    OBS (dp-teleops)