diff --git a/docs/dev/active/JSON_TELEMETRY_API.md b/docs/dev/active/JSON_TELEMETRY_API.md new file mode 100644 index 00000000..91933672 --- /dev/null +++ b/docs/dev/active/JSON_TELEMETRY_API.md @@ -0,0 +1,214 @@ +# JSON Telemetry API (read-only export) + +## Why + +External services need Kite's live telemetry, but every existing egress path is a binary FC protocol +(LTM / MAVLink / CRSF / SmartPort over serial / BLE / TCP / UDP). A consumer that just wants "where is the +aircraft and is it armed" had to implement a protocol parser first. + +This adds a read-only JSON export: a snapshot endpoint and a live stream, over plain HTTP. Each stream is +tagged with a **mission ID** so the consumer can attribute samples to a mission record. + +## Shape + +It is **not** a new subsystem. It is one `Encoder` (`json`) plus one `OutputSink` (`http`) inside the +existing Telemetry Relay (`src-tauri/src/telemetry_forward/`), so it inherits the relay's persisted config, +settings UI, connect/disconnect lifecycle, throughput stats and reconfigure reconciler. No producer +(MAVLink / MSP / passive) is touched — the relay already taps the app's own `telemetry-*` events. + +``` +telemetry-* events ─► RelayHub tap ─► TelemetryCache ─► JsonEncoder ─► HttpSink ─► GET /api/v1/… + (6 unified fields) (rate-limited) (snapshot + SSE) +``` + +Because it rides the shared cache, the export works for **any** inbound protocol, not just MAVLink. + +## Endpoints + +Served on `127.0.0.1:` (default 8080), or `0.0.0.0:` if LAN is opted into. + +| Route | Response | +|---|---| +| `GET /api/v1/telemetry` | `200 application/json` — the most recent frame. `503` if no telemetry has arrived yet. | +| `GET /api/v1/stream` | `200 text/event-stream` — SSE, one `data:` record per frame. | +| `GET /api/v1/health` | `200 application/json` — `{ ok, schema, missionId, hasData, streamClients }`. | + +All responses carry `Access-Control-Allow-Origin: *` — safe, because the API is read-only and carries no +credentials — so a browser consumer can `fetch()` / `EventSource` it directly. + +### Payload + +Compact single-line JSON. `schema` is the contract version (`SCHEMA_VERSION` in `encoders/json.rs`); `seq` +is monotonic so a consumer can detect dropped frames. Telemetry blocks are **omitted entirely** when the +source hasn't reported them, rather than being zero-filled. + +```json +{ + "schema": 1, + "missionId": "ollebo-test-1", + "ts": 1784018754527, + "seq": 42, + "attitude": { "roll": -2.1, "pitch": 4.8, "yaw": 271.3 }, + "gps": { "fixType": 3, "numSat": 14, "lat": 59.3293, "lon": 18.0686, + "altMsl": 132.4, "groundSpeed": 17.2, "course": 271.0 }, + "altitude": { "altitude": 118.0, "vario": -0.4 }, + "battery": { "voltage": 22.1, "current": 8.4, "power": 185.6, + "mahDrawn": 1420, "percentage": 63, "cellCount": 6, "rssi": 1023 }, + "status": { "armed": true, "armingFlags": 4, "flightModeFlags": 1, + "cpuLoad": 21, "sensorStatus": 3 }, + "airspeed": { "airspeed": 18.9 } +} +``` + +`armed` is derived from `armingFlags` bit 2, which is normalized across MSP and MAVLink (mirrors +`ARMED_BIT` in `src/lib/helpers/arming.ts`) — consumers shouldn't have to decode the bitfield. + +The DTO is defined explicitly rather than serializing the internal cache structs. Those are snake_case +internals that change with the frontend; a public contract must not be coupled to them. + +## Decisions + +**The mission ID is a hard gate, enforced backend-side.** `Relay::build` resolves it *before* constructing +the output sink, so with no mission ID the relay is refused and **no port is ever bound** — the feature is +genuinely off, not merely serving untagged data. The frontend flags the empty field, but the backend check +is the authoritative one (hand-editing localStorage does not get you an untagged API). + +**HTTP output requires the JSON protocol.** The sink wraps each frame as an SSE record, so pairing it with +a binary encoder would emit garbage. The reverse is allowed: JSON out a serial/TCP/UDP sink is legitimate +newline-delimited JSON. + +**SSE, not WebSocket.** The stream is one-way and read-only. A WebSocket handshake needs SHA-1 + base64 and +client-frame unmasking, and `tokio` is compiled without `net`/`rt-multi-thread` in the release build, so +axum/hyper/tungstenite aren't available without a significant dependency change. SSE is plain HTTP/1.1 text +and hand-rolls in ~40 lines on `std::net::TcpListener` — the same approach as `video/mjpeg_server.rs`. + +**Loopback by default.** A tracker relay binds `0.0.0.0` because reaching the LAN is its whole purpose. A +telemetry API is different: it should not be silently readable by everyone on a field or public network. +LAN exposure is an explicit per-relay opt-in. + +**Rate-limited in the encoder.** The relay paces `frame_set` on the attitude update, which on MAVLink can +run at 10–50 Hz. `JsonEncoder` returns an empty `Vec` when called too soon; `Relay::emit_set` already +early-returns on that, so no frame is written and the byte/frame counters correctly stay put. Default 5 Hz, +configurable 0.1–50. + +**Stalled consumers can't stall the relay.** `HttpSink::write` runs on the Tauri event-listener thread that +drives *every* relay's dispatch. So: per-client write timeout (2 s) with dead clients dropped; the snapshot +mutex is never held across a socket write; and each connection is served on its own thread so a client that +connects and goes silent can't block the accept loop. + +## Lifecycle (known limitation) + +Relays are configured on primary connect and cleared on disconnect, so **the HTTP server binds on connect +and disappears on disconnect** — a consumer gets connection-refused, not a "no vehicle" response, whenever +Kite isn't connected to an aircraft. If a persistent endpoint is needed, the server has to be lifted out of +the relay lifecycle; that's a deliberate follow-up, not an oversight. + +## Worked example: streaming a flight into an Ollebo mission + +[Ollebo](https://www.ollebo.com) is one concrete consumer — a free service for hosting drone maps and +showing live missions. It's used here purely to show the export doing real work; **no Ollebo-specific code +exists in Kite**, and the same pattern fits any consumer. + +Ollebo ingests telemetry as `PUT https://api.ollebo.com/event/` and replays it on its own live +map. So the bridge is a plain client: read Kite's SSE stream, map each frame to an Ollebo event, PUT it. + +> **The Ollebo mission key is a credential**, not just a label — anyone holding it can write to that +> mission. Keep it in the bridge (below), and be careful about putting it in Kite's `missionId` field: that +> value is stamped into every frame, so with the LAN checkbox on it would be readable by anyone on the +> network. The `missionId` field is intended as a plain identifier; loopback is the default for a reason. + +```python +#!/usr/bin/env python3 +"""Feed Kite-GC's JSON telemetry export into an Ollebo mission. Stdlib only. + + python3 ollebo-bridge.py --mission-key +""" +import argparse, json, urllib.request + + +def to_ollebo(frame, device): + """Kite frame -> Ollebo event. Skips frames with no usable GPS fix.""" + gps = frame.get("gps") + if not gps or gps.get("fixType", 0) < 2: + return None + lon, lat = gps["lon"], gps["lat"] + alt = (frame.get("altitude") or {}).get("altitude", gps.get("altMsl", 0.0)) + return { + "type": "telemetry", + "device": device, + "geopoint": [lon, lat], + "x": lon, "y": lat, "z": alt, + "data": alt, + "jsonData": { + "seq": frame.get("seq"), + "ts": frame.get("ts"), + "armed": (frame.get("status") or {}).get("armed"), + "heading": (frame.get("attitude") or {}).get("yaw"), + "speed": gps.get("groundSpeed"), + "battery": (frame.get("battery") or {}).get("percentage"), + "voltage": (frame.get("battery") or {}).get("voltage"), + }, + } + + +def put_event(api, key, event): + req = urllib.request.Request( + f"{api}/event/{key}", + data=json.dumps(event).encode("utf-8"), + method="PUT", + headers={"Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=10) as r: + return r.status + + +def main(): + p = argparse.ArgumentParser() + p.add_argument("--kite", default="http://127.0.0.1:8080", help="Kite JSON export") + p.add_argument("--api", default="https://api.ollebo.com") + p.add_argument("--mission-key", required=True, help="per-mission credential") + p.add_argument("--device", default="kite-gc") + a = p.parse_args() + + with urllib.request.urlopen(f"{a.kite}/api/v1/stream") as stream: + for raw in stream: # SSE records: "data: {...}\n\n" + line = raw.decode("utf-8", "replace").strip() + if not line.startswith("data: "): + continue + event = to_ollebo(json.loads(line[6:]), a.device) + if event: + put_event(a.api, a.mission_key, event) + + +if __name__ == "__main__": + main() +``` + +Set the relay to JSON/HTTP with a mission ID, connect, then run the bridge — the flight shows up live on +the Ollebo mission map. + +Note the shape of this: Kite stays vendor-neutral and just *serves* telemetry; the consumer does the +pushing. An in-app push sink (Kite PUTs directly, no bridge process) is a plausible follow-up, but it would +put a third-party endpoint and a credential inside the app, which is a much bigger ask of the project. + +## Out of scope / follow-ups + +- **Richer payload.** The relay cache carries six unified fields; the MAVLink handler emits ~20 event types + (wind, EKF status, RC channels, per-battery data, nav status, statustext, GPS stats…). Adding them is + purely additive — a field on `TelemetryCache`, a `tap!` line in `telemetry_forward/mod.rs`, a field on the + DTO — and touches no producer. +- **Full-fidelity raw MAVLink.** For a consumer that speaks MAVLink natively, the raw frames are already + captured at `mavlink_proto/handler.rs` (`frame.raw_bytes`, for the `.tlog`). A passthrough sink there + would be a different, larger feature. +- **Auth.** Not needed while read-only and loopback. Would be required before any write path, or before + LAN exposure is made the default. + +## Files + +| File | Role | +|---|---| +| `src-tauri/src/telemetry_forward/encoders/json.rs` | DTO, mission-ID stamping, rate limiter | +| `src-tauri/src/telemetry_forward/output/http.rs` | HTTP server, snapshot + SSE broadcast | +| `src-tauri/src/telemetry_forward/relay.rs` | Config fields + factory, the mission-ID gate | +| `src/lib/stores/relay.ts` | `json` / `http` types, `missionIdMissing()`, defaults | +| `src/lib/components/RelayPanel.svelte` | Protocol/output options, mission-ID field, port-collision guard | diff --git a/docs/dev/reference/data-pipeline.md b/docs/dev/reference/data-pipeline.md index fcaf5eb7..cabdfada 100644 --- a/docs/dev/reference/data-pipeline.md +++ b/docs/dev/reference/data-pipeline.md @@ -190,6 +190,11 @@ These subsystems are independent of the inbound `TelemetryData` pipeline: - **Telemetry Relay — outbound transcode.** `telemetry_forward/` taps the live decoded telemetry, re-encodes it into LTM / MAVLink / CRSF / SmartPort and sends it out a chosen transport (Serial / BLE / TCP / UDP). Persisted relay configs auto-connect on primary connect. +- **JSON export API — outbound, read-only.** A relay whose protocol is `json` and whose output is `http` + serves the same tapped telemetry as JSON over a small embedded HTTP server: a snapshot + (`GET /api/v1/telemetry`) and a live SSE stream (`GET /api/v1/stream`). Loopback-only unless LAN is + explicitly opted into, rate-limited independently of the source, and **gated on a mission ID** — with + none configured the relay is refused and no port is bound. See `docs/dev/active/JSON_TELEMETRY_API.md`. ## File index (telemetry pipeline) diff --git a/src-tauri/src/telemetry_forward/encoders/json.rs b/src-tauri/src/telemetry_forward/encoders/json.rs new file mode 100644 index 00000000..9d80265c --- /dev/null +++ b/src-tauri/src/telemetry_forward/encoders/json.rs @@ -0,0 +1,218 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// Copyright (C) 2026 Marc Hoffmann (b14ckyy) + +//! JSON encoder — the read-only telemetry export, served over HTTP by `output/http.rs`. +//! +//! Unlike the FC protocols, this one has a **public** wire contract, so it does not serialize the +//! internal cache structs directly (those are snake_case and free to change with the frontend). It maps +//! them into an explicit, versioned camelCase DTO instead. Bump `SCHEMA_VERSION` on any incompatible +//! change to it. +//! +//! Every frame carries the mission ID, so a consumer can attribute samples without out-of-band context. +//! A relay with no mission ID never gets built at all (see `relay.rs`). + +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use serde::Serialize; + +use super::super::cache::TelemetryCache; +use super::Encoder; + +/// Payload contract version. Bump on any incompatible change to `Frame` and friends. +const SCHEMA_VERSION: u32 = 1; + +/// ARMED is bit 2 of the normalized `arming_flags` bitfield, for MSP and MAVLink alike (INAV's +/// `armingFlag_e`; the MAVLink path maps HEARTBEAT's armed bit onto it). Mirrors `ARMED_BIT` in +/// `src/lib/helpers/arming.ts` — keep the two in step. +const ARMED_BIT: u32 = 1 << 2; + +/// Fall back to this when the configured rate is absent or nonsensical. +const DEFAULT_RATE_HZ: f32 = 5.0; + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Frame<'a> { + /// Payload contract version — see `SCHEMA_VERSION`. + schema: u32, + mission_id: &'a str, + /// Unix epoch milliseconds, stamped at encode time. + ts: u64, + /// Monotonic counter — a gap tells the consumer it dropped frames. + seq: u64, + #[serde(skip_serializing_if = "Option::is_none")] + attitude: Option, + #[serde(skip_serializing_if = "Option::is_none")] + gps: Option, + #[serde(skip_serializing_if = "Option::is_none")] + altitude: Option, + #[serde(skip_serializing_if = "Option::is_none")] + battery: Option, + #[serde(skip_serializing_if = "Option::is_none")] + status: Option, + #[serde(skip_serializing_if = "Option::is_none")] + airspeed: Option, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Attitude { + /// Degrees, ±180. + roll: f64, + /// Degrees, ±90. + pitch: f64, + /// Heading, 0–360. + yaw: f64, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Gps { + fix_type: u8, + num_sat: u8, + /// Decimal degrees. + lat: f64, + lon: f64, + /// Metres. + alt_msl: f64, + /// m/s. + ground_speed: f64, + /// Degrees. + course: f64, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Altitude { + /// Metres. Whether this is true MSL or relative-to-home depends on the source protocol. + altitude: f64, + /// m/s, positive up. + vario: f64, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Battery { + /// Volts. + voltage: f64, + /// Amps. + current: f64, + /// Watts. + power: f64, + mah_drawn: u32, + /// 0–100. + percentage: u8, + cell_count: u8, + /// Raw RSSI as the source protocol reports it (scale is protocol-dependent). + rssi: u16, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Status { + /// Derived from `armingFlags` bit 2 — the one field a consumer almost always wants. + armed: bool, + arming_flags: u32, + flight_mode_flags: u32, + cpu_load: u16, + sensor_status: u16, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct Airspeed { + /// m/s. + airspeed: f64, +} + +pub struct JsonEncoder { + mission_id: String, + /// Minimum wall-clock gap between emitted frames — see `frame_set`. + min_interval: Duration, + last_emit: Option, + seq: u64, +} + +impl JsonEncoder { + /// `rate_hz` is clamped to a sane range: below ~0.1 Hz the export looks dead, and above 50 Hz we'd + /// just be reserializing the same cache faster than any source updates it. An absent or nonsensical + /// value falls back to the default. + pub fn new(mission_id: String, rate_hz: Option) -> Self { + let hz = match rate_hz { + Some(hz) if hz.is_finite() && hz > 0.0 => hz, + _ => DEFAULT_RATE_HZ, + }; + let hz = hz.clamp(0.1, 50.0); + Self { + mission_id, + min_interval: Duration::from_secs_f32(1.0 / hz), + last_emit: None, + seq: 0, + } + } +} + +impl Encoder for JsonEncoder { + /// The relay paces `frame_set` on the *attitude* update, which on MAVLink can run at 10–50 Hz — far + /// more than a JSON/HTTP consumer wants. So we rate-limit here and return an **empty** Vec when + /// called too soon; `Relay::emit_set` early-returns on that, so no frame is written and the byte/frame + /// counters correctly stay put. + fn frame_set(&mut self, cache: &TelemetryCache) -> Vec { + let now = Instant::now(); + if let Some(last) = self.last_emit { + if now.duration_since(last) < self.min_interval { + return Vec::new(); + } + } + self.last_emit = Some(now); + self.seq += 1; + + let frame = Frame { + schema: SCHEMA_VERSION, + mission_id: &self.mission_id, + ts: SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0), + seq: self.seq, + attitude: cache.attitude.as_ref().map(|a| Attitude { roll: a.roll, pitch: a.pitch, yaw: a.yaw }), + gps: cache.gps.as_ref().map(|g| Gps { + fix_type: g.fix_type, + num_sat: g.num_sat, + lat: g.lat, + lon: g.lon, + alt_msl: g.alt_msl, + ground_speed: g.ground_speed, + course: g.course, + }), + altitude: cache.altitude.as_ref().map(|a| Altitude { altitude: a.altitude, vario: a.vario }), + battery: cache.analog.as_ref().map(|a| Battery { + voltage: a.voltage, + current: a.current, + power: a.power, + mah_drawn: a.mah_drawn, + percentage: a.battery_percentage, + cell_count: a.cell_count, + rssi: a.rssi, + }), + status: cache.status.as_ref().map(|s| Status { + armed: (s.arming_flags & ARMED_BIT) != 0, + arming_flags: s.arming_flags, + flight_mode_flags: s.flight_mode_flags, + cpu_load: s.cpu_load, + sensor_status: s.sensor_status, + }), + airspeed: cache.airspeed.as_ref().map(|a| Airspeed { airspeed: a.airspeed }), + }; + + match serde_json::to_vec(&frame) { + Ok(mut bytes) => { + bytes.push(b'\n'); + bytes + } + Err(e) => { + log::warn!("[RELAY json] encode failed: {e}"); + Vec::new() + } + } + } +} diff --git a/src-tauri/src/telemetry_forward/encoders/mod.rs b/src-tauri/src/telemetry_forward/encoders/mod.rs index e6848e19..a4935027 100644 --- a/src-tauri/src/telemetry_forward/encoders/mod.rs +++ b/src-tauri/src/telemetry_forward/encoders/mod.rs @@ -5,6 +5,7 @@ //! unified telemetry cache into wire frames of its protocol. pub mod crsf; +pub mod json; pub mod ltm; pub mod mavlink; pub mod smartport; diff --git a/src-tauri/src/telemetry_forward/output/http.rs b/src-tauri/src/telemetry_forward/output/http.rs new file mode 100644 index 00000000..040e6d87 --- /dev/null +++ b/src-tauri/src/telemetry_forward/output/http.rs @@ -0,0 +1,247 @@ +// SPDX-License-Identifier: GPL-3.0-or-later +// Copyright (C) 2026 Marc Hoffmann (b14ckyy) + +//! HTTP output sink — Kite hosts a small read-only HTTP server that serves the encoded telemetry as a +//! snapshot (pull) and as a live SSE stream (push). Pairs with the `json` encoder; external services +//! consume it without having to parse a binary FC protocol. +//! +//! Hand-rolled on `std::net::TcpListener`, like `video/mjpeg_server.rs` — `tokio` is compiled without +//! `net`/`rt-multi-thread` in the release build, so axum/hyper aren't available to us. +//! +//! Server-Sent Events rather than WebSocket: the stream is one-way and read-only, SSE is plain HTTP/1.1 +//! text (no SHA-1 handshake, no frame masking), and it's consumable by `EventSource` in a browser and by +//! any HTTP client. +//! +//! Routes: +//! GET /api/v1/telemetry → 200 application/json, the most recent frame, then close +//! GET /api/v1/stream → 200 text/event-stream, frames pushed as they're encoded +//! GET /api/v1/health → 200 application/json, liveness + mission id +//! +//! Binds loopback by default. Unlike a tracker relay (which exists to reach the LAN), a telemetry API +//! shouldn't be silently readable by everyone on a field network — LAN exposure is an explicit opt-in. + +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::Duration; + +use super::OutputSink; + +/// Drop a client that can't accept a frame within this window, so one stalled consumer can't block the +/// broadcast — `write()` runs on the Tauri event-listener thread that drives every relay's dispatch. +const CLIENT_WRITE_TIMEOUT: Duration = Duration::from_secs(2); + +/// Give up on a client that connects but never sends a request line. +const CLIENT_READ_TIMEOUT: Duration = Duration::from_secs(5); + +/// Cap the request we're willing to read. We only need the request line; anything larger is not a client +/// we want to serve. +const MAX_REQUEST_BYTES: usize = 2048; + +/// Read-only API with no credentials, so a wildcard origin is safe and lets a browser-based consumer +/// fetch it directly. +const CORS: &str = "Access-Control-Allow-Origin: *\r\n"; + +pub struct HttpSink { + addr: String, + /// Most recently encoded frame, served by `GET /api/v1/telemetry`. `None` until the first frame. + snapshot: Arc>>>, + /// Clients currently subscribed to `GET /api/v1/stream`. + clients: Arc>>, + running: Arc, +} + +impl HttpSink { + /// Bind the API server. `lan` exposes it on `0.0.0.0` instead of loopback. + pub fn open(port: u16, lan: bool, mission_id: String) -> Result { + let host = if lan { "0.0.0.0" } else { "127.0.0.1" }; + let addr = format!("{host}:{port}"); + let listener = + TcpListener::bind(&addr).map_err(|e| format!("HTTP relay bind {addr} failed: {e}"))?; + listener + .set_nonblocking(true) + .map_err(|e| format!("HTTP relay set_nonblocking failed: {e}"))?; + + let snapshot: Arc>>> = Arc::new(Mutex::new(None)); + let clients: Arc>> = Arc::new(Mutex::new(Vec::new())); + let running = Arc::new(AtomicBool::new(true)); + + // Accept loop on a background thread (non-blocking poll so Drop can stop it promptly). Each + // connection is served on its own thread so a client that connects and then dawdles can't hold up + // the accept loop for its whole read timeout. Request volume here is a handful of clients, so a + // thread per connection is cheap. + let snap = snapshot.clone(); + let cl = clients.clone(); + let r = running.clone(); + thread::spawn(move || { + while r.load(Ordering::Relaxed) { + match listener.accept() { + Ok((stream, peer)) => { + let _ = stream.set_nodelay(true); + let snap = snap.clone(); + let cl = cl.clone(); + let mission_id = mission_id.clone(); + thread::spawn(move || serve(stream, peer.to_string(), &snap, &cl, &mission_id)); + } + Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { + thread::sleep(Duration::from_millis(50)); + } + Err(e) => { + log::warn!("[RELAY http] accept error: {e}"); + thread::sleep(Duration::from_millis(200)); + } + } + } + }); + + Ok(Self { addr, snapshot, clients, running }) + } +} + +impl OutputSink for HttpSink { + /// Cache the frame for the snapshot route, then push it to every SSE subscriber. Clients that error + /// or time out are dropped. Never fails: a relay with no consumers is idle, not broken. + fn write(&mut self, data: &[u8]) -> Result<(), String> { + *self.snapshot.lock().unwrap() = Some(data.to_vec()); + + let mut clients = self.clients.lock().unwrap(); + if clients.is_empty() { + return Ok(()); + } + // SSE framing. The JSON encoder emits compact, newline-terminated JSON — one line, so it maps to + // a single `data:` field. Strip its trailing newline; the blank line is the record terminator. + let payload = data.strip_suffix(b"\n").unwrap_or(data); + let mut frame = Vec::with_capacity(payload.len() + 8); + frame.extend_from_slice(b"data: "); + frame.extend_from_slice(payload); + frame.extend_from_slice(b"\n\n"); + + clients.retain_mut(|c| c.write_all(&frame).is_ok()); + Ok(()) + } + + fn description(&self) -> String { + format!("HTTP({})", self.addr) + } + + /// "Pending" while nobody is streaming — the server is up but nothing is being pushed. + fn pending(&self) -> bool { + self.clients.lock().unwrap().is_empty() + } +} + +impl Drop for HttpSink { + fn drop(&mut self) { + self.running.store(false, Ordering::Relaxed); + } +} + +/// Handle one accepted connection: parse the request line and either answer and close, or (for the +/// stream route) hand the socket to the broadcast list. +fn serve( + mut stream: TcpStream, + peer: String, + snapshot: &Arc>>>, + clients: &Arc>>, + mission_id: &str, +) { + // The listener is non-blocking, and accepted sockets inherit that on some platforms — force blocking + // with explicit timeouts so a silent client can't wedge the accept loop. + let _ = stream.set_nonblocking(false); + let _ = stream.set_read_timeout(Some(CLIENT_READ_TIMEOUT)); + let _ = stream.set_write_timeout(Some(CLIENT_WRITE_TIMEOUT)); + + let Some((method, path)) = read_request_line(&mut stream) else { + return; + }; + + if method == "OPTIONS" { + let _ = stream.write_all( + format!("HTTP/1.1 204 No Content\r\n{CORS}Access-Control-Allow-Methods: GET, OPTIONS\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").as_bytes(), + ); + return; + } + if method != "GET" { + respond(&mut stream, "405 Method Not Allowed", "application/json", br#"{"error":"method not allowed"}"#); + return; + } + + // Ignore any query string — none of the routes take parameters. + let route = path.split('?').next().unwrap_or(""); + match route { + "/api/v1/telemetry" => { + // Copy the frame out and release the lock *before* writing: a response can block for up to + // CLIENT_WRITE_TIMEOUT, and `write()` needs this same lock on the relay dispatch thread. + // Holding it across the write would stall every relay behind one slow HTTP client. + let frame = snapshot.lock().unwrap().clone(); + match frame { + Some(f) => respond(&mut stream, "200 OK", "application/json", &f), + // Server is up but no telemetry has arrived yet — a real state, not an error. + None => respond(&mut stream, "503 Service Unavailable", "application/json", br#"{"error":"no telemetry yet"}"#), + } + } + "/api/v1/health" => { + // Same reasoning: snapshot the values, drop the locks, then write. + let has_data = snapshot.lock().unwrap().is_some(); + let stream_clients = clients.lock().unwrap().len(); + let body = serde_json::json!({ + "ok": true, + "schema": 1, + "missionId": mission_id, + "hasData": has_data, + "streamClients": stream_clients, + }) + .to_string(); + respond(&mut stream, "200 OK", "application/json", body.as_bytes()); + } + "/api/v1/stream" => { + let headers = format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n{CORS}Cache-Control: no-cache\r\nConnection: keep-alive\r\n\r\n" + ); + if stream.write_all(headers.as_bytes()).is_ok() && stream.flush().is_ok() { + log::info!("[RELAY http] stream client connected: {peer}"); + clients.lock().unwrap().push(stream); + } + } + _ => respond(&mut stream, "404 Not Found", "application/json", br#"{"error":"not found"}"#), + } +} + +/// Write a complete response and let the socket close on drop. +fn respond(stream: &mut TcpStream, status: &str, content_type: &str, body: &[u8]) { + let headers = format!( + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\n{CORS}Cache-Control: no-cache\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ); + let _ = stream.write_all(headers.as_bytes()); + let _ = stream.write_all(body); + let _ = stream.flush(); +} + +/// Read until the end of the request headers (or the cap) and return the request line's method + path. +fn read_request_line(stream: &mut TcpStream) -> Option<(String, String)> { + let mut buf = vec![0u8; MAX_REQUEST_BYTES]; + let mut len = 0; + while len < buf.len() { + match stream.read(&mut buf[len..]) { + Ok(0) => break, + Ok(n) => { + len += n; + if buf[..len].windows(4).any(|w| w == b"\r\n\r\n") { + break; + } + } + Err(_) => break, + } + } + if len == 0 { + return None; + } + let head = String::from_utf8_lossy(&buf[..len]); + let mut parts = head.lines().next()?.split_whitespace(); + let method = parts.next()?.to_string(); + let path = parts.next()?.to_string(); + Some((method, path)) +} diff --git a/src-tauri/src/telemetry_forward/output/mod.rs b/src-tauri/src/telemetry_forward/output/mod.rs index 49994c7f..5ab96b7f 100644 --- a/src-tauri/src/telemetry_forward/output/mod.rs +++ b/src-tauri/src/telemetry_forward/output/mod.rs @@ -6,6 +6,7 @@ //! BLE / TCP-server / UDP follow in Phase 2. pub mod ble; +pub mod http; pub mod serial; pub mod tcp; pub mod udp; diff --git a/src-tauri/src/telemetry_forward/relay.rs b/src-tauri/src/telemetry_forward/relay.rs index 2b2f31ba..8f4a52b3 100644 --- a/src-tauri/src/telemetry_forward/relay.rs +++ b/src-tauri/src/telemetry_forward/relay.rs @@ -8,11 +8,13 @@ use serde::{Deserialize, Serialize}; use super::cache::TelemetryCache; use super::encoders::crsf::CrsfEncoder; +use super::encoders::json::JsonEncoder; use super::encoders::ltm::LtmEncoder; use super::encoders::mavlink::MavlinkEncoder; use super::encoders::smartport::SmartportEncoder; use super::encoders::Encoder; use super::output::ble::BleSink; +use super::output::http::HttpSink; use super::output::serial::SerialSink; use super::output::tcp::TcpSink; use super::output::udp::UdpSink; @@ -24,12 +26,20 @@ use super::output::OutputSink; pub struct RelayConfig { pub id: String, pub enabled: bool, - /// Output protocol: "ltm" (more in later phases: "mavlink" / "crsf" / "smartport"). + /// Output protocol: "ltm" / "mavlink" / "crsf" / "smartport" / "json". pub protocol: String, + /// Identifies the mission these samples belong to, stamped into every frame of the `json` export. + /// **Required** for `json` — without it that relay is refused (see `build`). Unused by the FC + /// protocols, which have no field to carry it. + pub mission_id: Option, + /// Output rate for `json` (Hz). Absent → the encoder's default. Ignored by the FC protocols, which + /// pace themselves off the source. + pub rate_hz: Option, pub output: RelayOutput, } -/// Output transport configuration. Supported `kind`: `serial` / `ble` / `tcp` (server) / `udp`. +/// Output transport configuration. Supported `kind`: `serial` / `ble` / `tcp` (server) / `udp` / +/// `http` (server). #[derive(Debug, Clone, PartialEq, Deserialize)] #[serde(rename_all = "camelCase")] pub struct RelayOutput { @@ -39,11 +49,14 @@ pub struct RelayOutput { pub baud: Option, /// ble (device id) pub ble_device_id: Option, - /// tcp (server listen port) + /// tcp + http (server listen port) pub listen_port: Option, /// udp (send target) pub host: Option, pub udp_port: Option, + /// http: bind `0.0.0.0` (LAN-reachable) instead of loopback. Off by default — a telemetry API + /// shouldn't be silently readable by everyone on a field network. + pub lan: Option, } /// Per-relay status snapshot pushed to the frontend (`relay-stats` event). @@ -80,11 +93,20 @@ impl Relay { /// Build a live relay from config (opens the output transport — may fail if the device is missing). /// Async because the BLE output has to connect (scan + GATT); serial/tcp/udp are immediate. pub async fn build(cfg: &RelayConfig) -> Result { + // The JSON export is gated on a mission ID: without one there's nothing to attribute the samples + // to, so the relay must not come up at all. Resolved *before* the sink is built, so a missing id + // means no port is ever bound — the feature is genuinely off, not merely serving untagged data. + let mission_id = cfg.mission_id.as_deref().map(str::trim).filter(|s| !s.is_empty()); + let encoder: Box = match cfg.protocol.as_str() { "ltm" => Box::new(LtmEncoder::new()), "mavlink" => Box::new(MavlinkEncoder::new()), "crsf" => Box::new(CrsfEncoder::new()), "smartport" => Box::new(SmartportEncoder::new()), + "json" => { + let id = mission_id.ok_or("JSON relay disabled: no mission ID configured")?; + Box::new(JsonEncoder::new(id.to_string(), cfg.rate_hz)) + } other => return Err(format!("Unsupported relay protocol: {}", other)), }; let sink: Box = match cfg.output.kind.as_str() { @@ -106,6 +128,17 @@ impl Relay { let port = cfg.output.udp_port.ok_or("udp relay needs a port")?; Box::new(UdpSink::open(host, port)?) } + "http" => { + // The sink wraps each frame as an SSE record, so it only makes sense for a text protocol. + // (JSON out a serial/tcp/udp sink is fine — newline-delimited — so the guard is one-way.) + if cfg.protocol != "json" { + return Err("HTTP output requires the JSON protocol".to_string()); + } + let port = cfg.output.listen_port.ok_or("http relay needs a listen port")?; + // Unreachable fallback: protocol == "json" already required a mission id above. + let id = mission_id.unwrap_or_default().to_string(); + Box::new(HttpSink::open(port, cfg.output.lan.unwrap_or(false), id)?) + } other => return Err(format!("Unsupported relay output kind: {}", other)), }; let target = sink.description(); diff --git a/src/lib/components/RelayPanel.svelte b/src/lib/components/RelayPanel.svelte index f0ffd1cd..38a9ac08 100644 --- a/src/lib/components/RelayPanel.svelte +++ b/src/lib/components/RelayPanel.svelte @@ -13,7 +13,15 @@ import { listen, type UnlistenFn } from '@tauri-apps/api/event'; import { connection, connectionProtocol, availablePorts, bleDevices } from '$lib/stores/connection'; import { settings } from '$lib/stores/settings'; - import { relayStats, relayResults, newRelay, type RelayConfig } from '$lib/stores/relay'; + import { + relayStats, + relayResults, + newRelay, + missionIdMissing, + DEFAULT_HTTP_PORT, + DEFAULT_JSON_RATE_HZ, + type RelayConfig, + } from '$lib/stores/relay'; import { reconfigureRelays } from '$lib/controllers/relayController'; import { startBleScan, stopBleScan, startBleDeviceListener, stopBleDeviceListener, refreshSerialPorts } from '$lib/controllers/connectionController'; @@ -56,11 +64,24 @@ let changed = false; const fixed = cur.map((r) => { const o = r.output; - if (o.baud === undefined || o.bleDeviceId === undefined || o.listenPort === undefined || o.host === undefined || o.udpPort === undefined) { + const outStale = + o.baud === undefined || o.bleDeviceId === undefined || o.listenPort === undefined || o.host === undefined || o.udpPort === undefined || o.lan === undefined; + const relayStale = r.missionId === undefined || r.rateHz === undefined; + if (outStale || relayStale) { changed = true; return { ...r, - output: { ...o, baud: o.baud ?? 115200, bleDeviceId: o.bleDeviceId ?? '', listenPort: o.listenPort ?? 5760, host: o.host ?? '', udpPort: o.udpPort ?? 14550 }, + missionId: r.missionId ?? '', + rateHz: r.rateHz ?? DEFAULT_JSON_RATE_HZ, + output: { + ...o, + baud: o.baud ?? 115200, + bleDeviceId: o.bleDeviceId ?? '', + listenPort: o.listenPort ?? 5760, + host: o.host ?? '', + udpPort: o.udpPort ?? 14550, + lan: o.lan ?? false, + }, }; } return r; @@ -109,13 +130,17 @@ void reconfigureRelays(); } // ── Port guards ────────────────────────────────────────────────────────────── - // TCP listen ports are local binds → must be unique (a duplicate makes the 2nd relay fail to bind). + // TCP and HTTP listen ports are local binds → must be unique (a duplicate makes the 2nd relay fail to + // bind). They share ONE port space, so the check spans both kinds — a TCP relay and an HTTP relay on + // the same port would otherwise both be accepted here and only fail at runtime. // UDP targets must be a unique host:port pair (same port to different hosts is fine). We auto-bump to // the next free port so a duplicate can't be configured. - function nextFreeTcpPort(start: number, excludeId: string): number { - let p = Math.max(1, Math.min(65535, start || 5760)); + const SERVER_KINDS: RelayConfig['output']['kind'][] = ['tcp', 'http']; + + function nextFreeTcpPort(start: number, excludeId: string, fallback = 5760): number { + let p = Math.max(1, Math.min(65535, start || fallback)); const used = (port: number) => - relays.some((r) => r.id !== excludeId && r.output.kind === 'tcp' && r.output.listenPort === port); + relays.some((r) => r.id !== excludeId && SERVER_KINDS.includes(r.output.kind) && r.output.listenPort === port); while (used(p) && p < 65535) p++; return p; } @@ -133,16 +158,44 @@ // fallback) so the backend never sees a missing field, and that ports don't collide with other relays. function setKind(r: RelayConfig, kind: RelayConfig['output']['kind']) { const host = r.output.host ?? ''; + // tcp and http both bind a listen port, so both must be de-duplicated against the shared port space. + // http defaults to 8080 rather than 5760 (that's a MAVLink-ish port, misleading for a JSON API). + const serverKind = SERVER_KINDS.includes(kind); + const portFallback = kind === 'http' ? DEFAULT_HTTP_PORT : 5760; + const listenPort = r.output.listenPort ?? portFallback; patchOutput(r.id, { kind, baud: r.output.baud ?? 115200, bleDeviceId: r.output.bleDeviceId ?? '', host, - listenPort: kind === 'tcp' ? nextFreeTcpPort(r.output.listenPort ?? 5760, r.id) : (r.output.listenPort ?? 5760), + lan: r.output.lan ?? false, + listenPort: serverKind ? nextFreeTcpPort(listenPort, r.id, portFallback) : listenPort, udpPort: kind === 'udp' ? nextFreeUdpPort(host, r.output.udpPort ?? 14550, r.id) : (r.output.udpPort ?? 14550), }); } + // Switching to the JSON protocol implies the HTTP output (that's the pairing the backend supports, and + // the only one an external service can consume). Switching away from JSON leaves the output alone — an + // HTTP output with a binary protocol is refused by the backend, and the row shows why. + // Done as ONE patch: two chained patches would each re-read `relays`, and the second would overwrite + // the first from a stale snapshot. + function setProtocol(r: RelayConfig, protocol: RelayConfig['protocol']) { + const patch: Partial = { + protocol, + missionId: r.missionId ?? '', + rateHz: r.rateHz ?? DEFAULT_JSON_RATE_HZ, + }; + if (protocol === 'json' && r.output.kind !== 'http') { + patch.output = { + ...r.output, + kind: 'http', + lan: r.output.lan ?? false, + listenPort: nextFreeTcpPort(r.output.listenPort ?? DEFAULT_HTTP_PORT, r.id, DEFAULT_HTTP_PORT), + }; + } + patchRelay(r.id, patch); + } + function addRelay() { settings.patch({ relays: [...relays, newRelay()] }); } @@ -190,6 +243,7 @@ {/if} {#each relays as r (r.id)} {@const st = rowState(r.id)} + {@const needsMissionId = missionIdMissing(r)}
patchRelay(r.id, { enabled: e.currentTarget.checked })} /> - setProtocol(r, e.currentTarget.value as RelayConfig['protocol'])}> + {#if r.output.kind === 'serial'} patchOutput(r.id, { listenPort: nextFreeTcpPort(Number(e.currentTarget.value), r.id, DEFAULT_HTTP_PORT) })} + /> + {/if} {st.label}{#if st.detail}{st.detail}{/if}
+ + + {#if r.protocol === 'json'} +
+ patchRelay(r.id, { missionId: e.currentTarget.value })} + /> + patchRelay(r.id, { rateHz: Number(e.currentTarget.value) || DEFAULT_JSON_RATE_HZ })} + /> + Hz +
+ {#if needsMissionId} +
{$t('relay.missionIdRequired')}
+ {/if} + {/if} {/each} @@ -352,6 +455,37 @@ .r-input.port-num::-webkit-inner-spin-button, .r-input.port-num::-webkit-outer-spin-button { -webkit-appearance: none; margin: 0; } + /* JSON export sub-row: mission id (required) + emit rate, indented under its relay row. */ + .relay-sub { + display: flex; + align-items: center; + gap: 6px; + margin: -2px 0 6px 21px; /* 21px ≈ the enable checkbox + its gap, so it lines up with the protocol select */ + } + .r-input.mission { flex: 1 1 auto; min-width: 0; } + .r-input.mission.invalid { border-color: #d40000; background: rgba(212, 0, 0, 0.08); } + .r-input.rate { width: 68px; appearance: textfield; -moz-appearance: textfield; } + .r-input.rate::-webkit-inner-spin-button, + .r-input.rate::-webkit-outer-spin-button { -webkit-appearance: none; margin: 0; } + .unit { font-size: 11px; color: #949494; } + + .sub-hint { + font-size: 11px; + color: #d40000; + margin: -3px 0 6px 21px; + } + + .chk { + display: inline-flex; + align-items: center; + gap: 4px; + font-size: 11px; + color: #cfcfcf; + white-space: nowrap; + flex: none; + } + .chk input { width: 13px; height: 13px; accent-color: #37a8db; } + .status { display: inline-flex; align-items: center; diff --git a/src/lib/i18n/locales/de.json b/src/lib/i18n/locales/de.json index d37445d2..fc77029d 100644 --- a/src/lib/i18n/locales/de.json +++ b/src/lib/i18n/locales/de.json @@ -37,7 +37,12 @@ "error": "Fehler", "waiting": "wartet", "deviceMissing": "Gerät fehlt", - "idle": "inaktiv" + "idle": "inaktiv", + "missionId": "Missions-ID", + "missionIdRequired": "Eine Missions-ID ist erforderlich — ohne sie bleibt der JSON-Export aus.", + "rateHint": "JSON-Exportrate (Hz)", + "exposeLan": "LAN", + "exposeLanHint": "Die API im lokalen Netzwerk bereitstellen statt nur auf diesem Rechner. Jeder im Netzwerk kann dann deine Telemetrie mitlesen." }, "connection": { "connect": "Verbinden", diff --git a/src/lib/i18n/locales/en.json b/src/lib/i18n/locales/en.json index 0b8743a4..3cc7cbe6 100644 --- a/src/lib/i18n/locales/en.json +++ b/src/lib/i18n/locales/en.json @@ -37,7 +37,12 @@ "error": "error", "waiting": "waiting", "deviceMissing": "device missing", - "idle": "idle" + "idle": "idle", + "missionId": "Mission ID", + "missionIdRequired": "A mission ID is required — the JSON export stays off without one.", + "rateHint": "JSON export rate (Hz)", + "exposeLan": "LAN", + "exposeLanHint": "Serve the API on the local network instead of this machine only. Anyone on the network can then read your telemetry." }, "connection": { "connect": "Connect", diff --git a/src/lib/i18n/locales/fr.json b/src/lib/i18n/locales/fr.json index da729588..aaebb37a 100644 --- a/src/lib/i18n/locales/fr.json +++ b/src/lib/i18n/locales/fr.json @@ -31,7 +31,12 @@ "error": "erreur", "waiting": "en attente", "deviceMissing": "appareil absent", - "idle": "inactif" + "idle": "inactif", + "missionId": "ID de mission", + "missionIdRequired": "Un ID de mission est requis — sans lui, l'export JSON reste désactivé.", + "rateHint": "Fréquence de l'export JSON (Hz)", + "exposeLan": "LAN", + "exposeLanHint": "Exposer l'API sur le réseau local au lieu de cette machine uniquement. Toute personne sur le réseau pourra alors lire votre télémétrie." }, "connection": { "connect": "Connecter", diff --git a/src/lib/stores/relay.ts b/src/lib/stores/relay.ts index 68346fb1..02631859 100644 --- a/src/lib/stores/relay.ts +++ b/src/lib/stores/relay.ts @@ -8,11 +8,13 @@ import { writable } from 'svelte/store'; -/** Output protocol to encode into. */ -export type RelayProtocol = 'ltm' | 'mavlink' | 'crsf' | 'smartport'; +/** Output protocol to encode into. `json` is the read-only export for external services (served by the + * `http` output); the rest are binary FC protocols for trackers/GCS. */ +export type RelayProtocol = 'ltm' | 'mavlink' | 'crsf' | 'smartport' | 'json'; -/** Output transport kind: serial (covers HC-05/BT-SPP virtual COM) / ble / tcp (server) / udp. */ -export type RelayOutputKind = 'serial' | 'ble' | 'tcp' | 'udp'; +/** Output transport kind: serial (covers HC-05/BT-SPP virtual COM) / ble / tcp (server) / udp / + * http (server: JSON snapshot + SSE stream). */ +export type RelayOutputKind = 'serial' | 'ble' | 'tcp' | 'udp' | 'http'; export interface RelayOutput { kind: RelayOutputKind; @@ -21,11 +23,13 @@ export interface RelayOutput { baud?: number; /** ble — device id */ bleDeviceId?: string; - /** tcp — server listen port */ + /** tcp + http — server listen port */ listenPort?: number; /** udp — send target */ host?: string; udpPort?: number; + /** http — bind 0.0.0.0 (LAN-reachable) instead of loopback. Off by default. */ + lan?: boolean; } /** One configured relay (persisted). */ @@ -33,6 +37,12 @@ export interface RelayConfig { id: string; enabled: boolean; protocol: RelayProtocol; + /** Stamped into every frame of the `json` export so a consumer can attribute the samples. REQUIRED for + * `json` — the backend refuses to start the relay without one, so no port is bound. Unused by the FC + * protocols, which have no field to carry it. */ + missionId?: string; + /** Output rate for `json`, in Hz. Absent → backend default (5 Hz). Ignored by the FC protocols. */ + rateHz?: number; output: RelayOutput; } @@ -62,12 +72,25 @@ export const relayStats = writable([]); /** Last configure result per relay id (so the UI can show "device missing" / errors). */ export const relayResults = writable>({}); +/** Default emit rate for the JSON export (Hz) — mirrors the backend's fallback in encoders/json.rs. */ +export const DEFAULT_JSON_RATE_HZ = 5; + +/** Default listen port for the HTTP export server. */ +export const DEFAULT_HTTP_PORT = 8080; + +/** A `json` relay with no mission ID is refused by the backend (no port is bound), so the UI flags it. */ +export function missionIdMissing(r: RelayConfig): boolean { + return r.protocol === 'json' && !r.missionId?.trim(); +} + /** Create a fresh default relay config row. */ export function newRelay(): RelayConfig { return { id: crypto.randomUUID(), enabled: true, protocol: 'ltm', - output: { kind: 'serial', port: '', baud: 115200, bleDeviceId: '', listenPort: 5760, host: '', udpPort: 14550 }, + missionId: '', + rateHz: DEFAULT_JSON_RATE_HZ, + output: { kind: 'serial', port: '', baud: 115200, bleDeviceId: '', listenPort: 5760, host: '', udpPort: 14550, lan: false }, }; }