Telemetry stream: GET /api/telemetry + MCP get_telemetry (#16)

src/mesh/mesh.service.ts

@@ -11,12 +11,14 @@ import {
   BROADCAST_ADDR,
   PORT_POSITION_APP,
   PORT_ROUTING_APP,
+  PORT_TELEMETRY_APP,
   PORT_TEXT_MESSAGE_APP,
   buildTextToRadio,
   buildWantConfigToRadio,
   decodeFromRadio,
   decodePosition,
   decodeRouting,
+  decodeTelemetry,
   describeRoutingError,
   extractFrames,
   frame,
@@ -41,6 +43,18 @@ export interface MessageRecord {
   ackError?: string;
 }
 
+/** One entry in the telemetry history: a TELEMETRY_APP packet from some node. */
+export interface TelemetryRecord {
+  /** sender node num */
+  from: number;
+  /** epoch ms */
+  at: number;
+  /** meshtastic.Telemetry oneof case name (e.g. "deviceMetrics", "environmentMetrics") */
+  kind: string;
+  /** the variant payload, JSON-safe (proto uint64 fields as decimal strings); unset optional fields omitted */
+  data: object;
+}
+
 // bounded history: keep the newest N records, in-memory only
 const HISTORY_LIMIT = 100;
 
@@ -77,13 +91,15 @@ export interface DeviceInfo {
 }
 
 // protobuf-es decodes proto uint64 fields as native bigint, which JSON.stringify rejects;
-// recursively convert so deviceInfo() output is always JSON-safe (bigint -> decimal string)
+// recursively convert so deviceInfo()/telemetry() output is always JSON-safe (bigint -> decimal string).
+// $-prefixed keys are protobuf-es internals ($typeName/$unknown), not device data — dropped.
 function jsonSafe<T>(value: T): T {
   if (typeof value === 'bigint') return String(value) as T;
   if (Array.isArray(value)) return value.map(jsonSafe) as T;
   if (value !== null && typeof value === 'object') {
     const out: Record<string, unknown> = {};
-    for (const [k, v] of Object.entries(value)) out[k] = jsonSafe(v);
+    for (const [k, v] of Object.entries(value))
+      if (!k.startsWith('$')) out[k] = jsonSafe(v);
     return out as T;
   }
   return value; // primitives + undefined pass through
@@ -137,6 +153,8 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
   // message history (newest kept, capped at HISTORY_LIMIT) + pending tx records awaiting their routing ACK/NACK
   private history: MessageRecord[] = [];
   private pendingTx = new Map<number, MessageRecord>();
+  // telemetry history (newest kept, capped at HISTORY_LIMIT)
+  private telemetryHistory: TelemetryRecord[] = [];
   private reconnectTimer: NodeJS.Timeout | null = null;
   private shuttingDown = false;
 
@@ -285,6 +303,16 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
     return [...this.history].reverse();
   }
 
+  /**
+   * Telemetry records, newest first: TELEMETRY_APP packets from any node — device metrics
+   * (battery/voltage/uptime/channel+air utilization) and environment sensor readings where nodes have them.
+   * `kind` is the meshtastic.Telemetry oneof case name; `data` is that variant's payload with proto uint64
+   * fields as decimal strings, unset optional fields omitted. In-memory only, last 100, reset on reconnect.
+   */
+  telemetry(): TelemetryRecord[] {
+    return [...this.telemetryHistory].reverse();
+  }
+
   /** Send `text` to node `to` on the pinned test channel; returns the packet id. Throws ServiceUnavailable when the bridge is down. */
   async sendText(to: number, text: string): Promise<number> {
     if (!this.connected || !this.socket)
@@ -311,6 +339,13 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
     if (this.history.length > HISTORY_LIMIT) this.history.shift();
   }
 
+  private pushTelemetry(rec: TelemetryRecord): void {
+    this.telemetryHistory.push(rec);
+    // ponytail: shift on a ≤100 array is fine; true ring buffer only if this ever gets hot
+    if (this.telemetryHistory.length > HISTORY_LIMIT)
+      this.telemetryHistory.shift();
+  }
+
   private summarize(n: NodeEntry, names: Map<number, string>): NodeSummary {
     const s: NodeSummary = { num: n.num, lastHeard: n.lastHeard || null };
     if (n.user?.longName) s.longName = n.user.longName;
@@ -347,6 +382,7 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
       this.configSections.clear();
       this.moduleConfigSections.clear();
       this.history = [];
+      this.telemetryHistory = [];
       this.pendingTx.clear();
       this.logger.log(`bridge connected ${this.host}:${this.port}`);
       const nonce = Math.floor(Math.random() * 0xffffffff) || 1;
@@ -481,6 +517,23 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
               this.pendingTx.delete(decoded.requestId);
             }
           }
+        } else if (decoded && decoded.portnum === PORT_TELEMETRY_APP) {
+          // live telemetry: the Telemetry is serialized in Data.payload
+          let tel: Proto.Telemetry.Telemetry | undefined;
+          try {
+            tel = await decodeTelemetry(decoded.payload);
+          } catch {
+            tel = undefined; // malformed payload — skip
+          }
+          const variant = tel?.variant;
+          if (variant?.case) {
+            this.pushTelemetry({
+              from: pkt.from,
+              at: Date.now(),
+              kind: variant.case,
+              data: jsonSafe(variant.value),
+            });
+          }
         }
         break;
       }