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

Merged
kamilkrzywda merged 1 commits from feat/telemetry into main 2026-09-24 23:56:11 +02:00
11 changed files with 286 additions and 6 deletions
+2 -2
View File
@@ -17,8 +17,8 @@
## API ## API
1. **Live base URL** — `http://bot.dev.kkhost.pl` (Traefik → docker-compose stack on this host; container port 3000 via external network dev_net, no published ports). 1. **Live base URL** — `http://bot.dev.kkhost.pl` (Traefik → docker-compose stack on this host; container port 3000 via external network dev_net, no published ports).
2. **Endpoints** — all under the `/api` global prefix: `GET /api/status` (`{ connected, myNodeNum, host, port }`), `POST /api/send` (body `{ "text": "...", "to"?: <node number | node name | "!b" | "broadcast"> }` — numbers pass through as-is (integer 1..0xffffffff), names resolve like `GET /api/nodes/:id`, omitted/""/"!b"/"broadcast" → broadcast; unknown or ambiguous targets are a 400. Returns `{ id }`), `GET /api/channels` (the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]`; psk/keys redacted), `GET /api/device` (device info captured during the initial sync: `{ myNodeNum, firmwareVersion?, metadata?, localConfig: {<section>: ...}, moduleConfig: {<module>: ...} }` — DeviceMetadata as-is plus per-section Config/ModuleConfig messages; security publicKey/privateKey/adminKey and network wifiPsk redacted to `"REDACTED"`; in-memory, reset on reconnect), `GET /api/messages` (message history newest first: rx text with `from`, tx sends with `to` + `status` pending→delivered/failed via routing ACKs — broadcasts stay pending because firmware strips want_ack; last 100, in-memory, reset on reconnect), `GET /api/nodes` (known nodes sorted by num as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? }, snr?, hopsAway?, viaMqtt? }[]`, from the initial sync plus live POSITION_APP updates; reset on reconnect) and `GET /api/nodes/:id` (lookup by exact node number/numeric string, then case-insensitive exact long/short name, else unique substring; 404 when ambiguous or missing). Scalar API docs are served at `/` (OpenAPI JSON at `/docs-json`). Keep this list in sync with the controller. All `POST /api/send` traffic targets channel index 3 (`test`) on purpose — never the primary channel; override via `MESH_SEND_CHANNEL` (default `3`). 2. **Endpoints** — all under the `/api` global prefix: `GET /api/status` (`{ connected, myNodeNum, host, port }`), `POST /api/send` (body `{ "text": "...", "to"?: <node number | node name | "!b" | "broadcast"> }` — numbers pass through as-is (integer 1..0xffffffff), names resolve like `GET /api/nodes/:id`, omitted/""/"!b"/"broadcast" → broadcast; unknown or ambiguous targets are a 400. Returns `{ id }`), `GET /api/channels` (the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]`; psk/keys redacted), `GET /api/device` (device info captured during the initial sync: `{ myNodeNum, firmwareVersion?, metadata?, localConfig: {<section>: ...}, moduleConfig: {<module>: ...} }` — DeviceMetadata as-is plus per-section Config/ModuleConfig messages; security publicKey/privateKey/adminKey and network wifiPsk redacted to `"REDACTED"`; in-memory, reset on reconnect), `GET /api/messages` (message history newest first: rx text with `from`, tx sends with `to` + `status` pending→delivered/failed via routing ACKs — broadcasts stay pending because firmware strips want_ack; last 100, in-memory, reset on reconnect), `GET /api/telemetry` (telemetry history newest first as `{ from, at /* epoch ms */, kind, data }[]`: live TELEMETRY_APP packets — device metrics and environment sensor readings where nodes have them; kind is the meshtastic.Telemetry oneof case name (deviceMetrics/environmentMetrics/airQualityMetrics/powerMetrics/localStats/healthMetrics/hostMetrics/trafficManagementStats), data that variant's payload with proto uint64 fields as decimal strings; last 100, in-memory, reset on reconnect), `GET /api/nodes` (known nodes sorted by num as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? }, snr?, hopsAway?, viaMqtt? }[]`, from the initial sync plus live POSITION_APP updates; reset on reconnect) and `GET /api/nodes/:id` (lookup by exact node number/numeric string, then case-insensitive exact long/short name, else unique substring; 404 when ambiguous or missing). Scalar API docs are served at `/` (OpenAPI JSON at `/docs-json`). Keep this list in sync with the controller. All `POST /api/send` traffic targets channel index 3 (`test`) on purpose — never the primary channel; override via `MESH_SEND_CHANNEL` (default `3`).
3. **MCP endpoint** — `POST /mcp` (excluded from the `/api` prefix in main.ts, like Scalar at `/`): an in-process MCP server over the official SDK's Streamable HTTP transport in stateless mode (`sessionIdGenerator: undefined`, fresh `McpServer` + transport per request). Tools map 1:1 to MeshService methods: `mesh_status`, `send_message(text, to?)` (same resolution as `POST /api/send`; errors surface as tool errors with the same messages), `list_channels`, `list_nodes`, `get_node(id)` (tool error when missing/ambiguous), `get_messages`, `device_info` (same redaction as `GET /api/device`). Results are JSON in a text content block; sends ride channel index 3 like everything else. Note: the SDK's root entry is broken upstream (#2273) — import via subpaths (`@modelcontextprotocol/sdk/server/mcp.js` etc.). 3. **MCP endpoint** — `POST /mcp` (excluded from the `/api` prefix in main.ts, like Scalar at `/`): an in-process MCP server over the official SDK's Streamable HTTP transport in stateless mode (`sessionIdGenerator: undefined`, fresh `McpServer` + transport per request). Tools map 1:1 to MeshService methods: `mesh_status`, `send_message(text, to?)` (same resolution as `POST /api/send`; errors surface as tool errors with the same messages), `list_channels`, `list_nodes`, `get_node(id)` (tool error when missing/ambiguous), `get_messages`, `get_telemetry` (same bounded recent window as `GET /api/telemetry`), `device_info` (same redaction as `GET /api/device`). Results are JSON in a text content block; sends ride channel index 3 like everything else. Note: the SDK's root entry is broken upstream (#2273) — import via subpaths (`@modelcontextprotocol/sdk/server/mcp.js` etc.).
## Other ## Other
+8
View File
@@ -1,5 +1,13 @@
# Changelog # Changelog
## v0.8.0 — 2026-09-24 21:40
- New `GET /api/telemetry` endpoint (and matching MCP tool `get_telemetry`, now eight tools): the telemetry stream captured from live `TELEMETRY_APP` packets (portnum 67) — device metrics (uptime/battery/voltage/channel+air utilization) and environment sensor readings where nodes have them. Records are `{ from, at /* epoch ms */, kind, data }[]` newest first: `kind` is the `meshtastic.Telemetry` oneof case name (`deviceMetrics`, `environmentMetrics`, `airQualityMetrics`, `powerMetrics`, `localStats`, `healthMetrics`, `hostMetrics` or `trafficManagementStats`) and `data` is that variant's payload. In-memory only, last 100 (same bounded pattern as message history), reset on every bridge reconnect.
- `MeshService.telemetry()` captures the new portnum in the existing packet handler: the `Data.payload` is decoded as a `meshtastic.Telemetry`, and packets without a variant payload are ignored. The variant payload goes through the same module-scope `jsonSafe` converter added in v0.7.1 — telemetry variants carry proto `uint64` fields (e.g. host-metrics byte counters) that decode to native bigint and would 500 Express' `res.json` otherwise, so they come back as decimal strings; unset optional fields are omitted from `data`.
- Verified against the installed `@meshtastic/protobufs` 2.8.0: `PortNum.TELEMETRY_APP = 67`; `Telemetry` is a `fixed32 time` plus oneof variant with exactly those eight cases; `DeviceMetrics` fields are batteryLevel/voltage/channelUtilization/airUtilTx/uptimeSeconds and `EnvironmentMetrics` the full sensor set (temperature, relativeHumidity, barometricPressure, gasResistance, voltage, current, iaq, distance, lux variants, wind/rainfall/soil/lightning, adc channels).
- Swagger: `GET /api/telemetry` documented with the kind values and an example carrying one record per kind. README + AGENTS.md endpoint/tool lists updated.
- Tests: fake-bridge gains a shared `telemetryPacketFrame` builder; new telemetry specs (device-metrics + environment variants → correct from/kind/data, cap at 100 dropping the oldest, reset on reconnect); MCP spec updated to eight tools incl. a `get_telemetry` call check.
## v0.7.1 — 2026-09-24 21:10 ## v0.7.1 — 2026-09-24 21:10
- Fix `GET /api/device` and MCP `device_info` crashing with `TypeError: Do not know how to serialize a BigInt` on live hardware (fw 2.7.15): protobuf-es decodes proto `uint64` fields as native JS `bigint`, which Express' `res.json`/`JSON.stringify` reject, and `MeshService.deviceInfo()` passed raw decoded proto objects through unconverted. Known affected fields: `Config.PowerConfig.powermonEnables` (in `localConfig.power`) and host-metrics byte counters (`freememBytes`/`diskfree*`). - Fix `GET /api/device` and MCP `device_info` crashing with `TypeError: Do not know how to serialize a BigInt` on live hardware (fw 2.7.15): protobuf-es decodes proto `uint64` fields as native JS `bigint`, which Express' `res.json`/`JSON.stringify` reject, and `MeshService.deviceInfo()` passed raw decoded proto objects through unconverted. Known affected fields: `Config.PowerConfig.powermonEnables` (in `localConfig.power`) and host-metrics byte counters (`freememBytes`/`diskfree*`).
+3
View File
@@ -30,6 +30,7 @@ All API endpoints live under the `/api` prefix:
- `GET /api/channels` — the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]` (psk/keys are redacted; empty until the initial sync completes) - `GET /api/channels` — the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]` (psk/keys are redacted; empty until the initial sync completes)
- `GET /api/device` — device info captured during the initial sync: `{ myNodeNum, firmwareVersion?, metadata?, localConfig: {<section>: ...}, moduleConfig: {<module>: ...} }`. Key material is never exposed (security publicKey/privateKey/adminKey and network wifiPsk are replaced with `"REDACTED"`); in-memory only, reset on every bridge reconnect - `GET /api/device` — device info captured during the initial sync: `{ myNodeNum, firmwareVersion?, metadata?, localConfig: {<section>: ...}, moduleConfig: {<module>: ...} }`. Key material is never exposed (security publicKey/privateKey/adminKey and network wifiPsk are replaced with `"REDACTED"`); in-memory only, reset on every bridge reconnect
- `GET /api/messages` — message history, newest first: inbound text (`dir:'rx'`, with `from`) and our sends (`dir:'tx'`, with `to` + `status`). tx records start `pending` and move to `delivered`/`failed` (with human-readable `ackError`) when the routing ACK/NACK arrives; broadcasts stay `pending` because firmware strips want_ack before TX. In-memory only, last 100, reset on every bridge reconnect - `GET /api/messages` — message history, newest first: inbound text (`dir:'rx'`, with `from`) and our sends (`dir:'tx'`, with `to` + `status`). tx records start `pending` and move to `delivered`/`failed` (with human-readable `ackError`) when the routing ACK/NACK arrives; broadcasts stay `pending` because firmware strips want_ack before TX. In-memory only, last 100, reset on every bridge reconnect
- `GET /api/telemetry` — telemetry history, newest first: live `TELEMETRY_APP` packets from any node as `{ from, at /* epoch ms */, kind, data }[]`. `kind` is the `meshtastic.Telemetry` oneof case name (`deviceMetrics`, `environmentMetrics`, `airQualityMetrics`, `powerMetrics`, `localStats`, `healthMetrics`, `hostMetrics` or `trafficManagementStats`) and `data` carries that variant's payload (proto uint64 fields as decimal strings; unset optional fields omitted). In-memory only, last 100, reset on every bridge reconnect
- `GET /api/nodes` — known mesh nodes as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? }, snr?, hopsAway?, viaMqtt? }[]`, sorted by node number. Collected during the initial sync plus live position updates; names/hwModel/battery appear only when reported, `lastHeard` is epoch seconds (null = never heard), `position` only when both coordinates are known, and snr/hopsAway/viaMqtt only when carried by the NodeInfo - `GET /api/nodes` — known mesh nodes as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? }, snr?, hopsAway?, viaMqtt? }[]`, sorted by node number. Collected during the initial sync plus live position updates; names/hwModel/battery appear only when reported, `lastHeard` is epoch seconds (null = never heard), `position` only when both coordinates are known, and snr/hopsAway/viaMqtt only when carried by the NodeInfo
- `GET /api/nodes/:id` — one node looked up by exact node number (numbers or numeric strings), then case-insensitive exact long/short name, else the unique case-insensitive substring match; 404 when nothing or more than one node matches - `GET /api/nodes/:id` — one node looked up by exact node number (numbers or numeric strings), then case-insensitive exact long/short name, else the unique case-insensitive substring match; 404 when nothing or more than one node matches
@@ -43,6 +44,7 @@ curl -X POST http://localhost:3000/api/send -H 'content-type: application/json'
curl http://localhost:3000/api/channels curl http://localhost:3000/api/channels
curl http://localhost:3000/api/device curl http://localhost:3000/api/device
curl http://localhost:3000/api/messages curl http://localhost:3000/api/messages
curl http://localhost:3000/api/telemetry
curl http://localhost:3000/api/nodes curl http://localhost:3000/api/nodes
curl http://localhost:3000/api/nodes/kamil curl http://localhost:3000/api/nodes/kamil
``` ```
@@ -68,6 +70,7 @@ endpoints — no new logic, same channel pinning:
- `list_nodes` — known nodes as lean summaries - `list_nodes` — known nodes as lean summaries
- `get_node(id)` — one node by number or name; tool error when missing/ambiguous - `get_node(id)` — one node by number or name; tool error when missing/ambiguous
- `get_messages` — message history, newest first (last 100, tx records carry delivery status) - `get_messages` — message history, newest first (last 100, tx records carry delivery status)
- `get_telemetry` — telemetry history, newest first (last 100): device metrics and environment sensor readings from any node; `kind` is the Telemetry oneof case name
- `device_info` — device info from the initial sync: node number, firmware version, metadata and per-section config/module-config (key material redacted as `"REDACTED"`) - `device_info` — device info from the initial sync: node number, firmware version, metadata and per-section config/module-config (key material redacted as `"REDACTED"`)
Point any MCP client at it (Streamable HTTP URL): Point any MCP client at it (Streamable HTTP URL):
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "mesh-bot", "name": "mesh-bot",
"version": "0.7.1", "version": "0.8.0",
"description": "", "description": "",
"author": "", "author": "",
"private": true, "private": true,
+10
View File
@@ -14,6 +14,8 @@ export const PORT_TEXT_MESSAGE_APP = 1;
export const PORT_POSITION_APP = 3; export const PORT_POSITION_APP = 3;
// verified against installed @meshtastic/protobufs Portnums.PortNum.ROUTING_APP // verified against installed @meshtastic/protobufs Portnums.PortNum.ROUTING_APP
export const PORT_ROUTING_APP = 5; export const PORT_ROUTING_APP = 5;
// verified against installed @meshtastic/protobufs Portnums.PortNum.TELEMETRY_APP
export const PORT_TELEMETRY_APP = 67;
const START1 = 0x94; const START1 = 0x94;
const START2 = 0xc3; const START2 = 0xc3;
@@ -81,6 +83,14 @@ export async function decodeRouting(
return fromBinary(P.Mesh.RoutingSchema, bytes); return fromBinary(P.Mesh.RoutingSchema, bytes);
} }
/** TELEMETRY_APP packets carry a serialized meshtastic.Telemetry in Data.payload. */
export async function decodeTelemetry(
bytes: Uint8Array,
): Promise<import('@meshtastic/protobufs').Telemetry.Telemetry> {
const P = await loadProto();
return fromBinary(P.Telemetry.TelemetrySchema, bytes);
}
/** Human-readable name for a Routing.Error code (from the installed @meshtastic/protobufs enum), `reason <n>` fallback. */ /** Human-readable name for a Routing.Error code (from the installed @meshtastic/protobufs enum), `reason <n>` fallback. */
export async function describeRoutingError(code: number): Promise<string> { export async function describeRoutingError(code: number): Promise<string> {
const P = await loadProto(); const P = await loadProto();
+27
View File
@@ -121,6 +121,33 @@ export function nodeInfoFrame(
); );
} }
/** TELEMETRY_APP packet frame with a serialized Telemetry in Data.payload (live telemetry from any node). */
export function telemetryPacketFrame(
from: number,
variant: P.Telemetry.Telemetry['variant'],
): Uint8Array {
const data = create(P.Mesh.DataSchema, {
portnum: P.Portnums.PortNum.TELEMETRY_APP,
payload: toBinary(
P.Telemetry.TelemetrySchema,
create(P.Telemetry.TelemetrySchema, { variant }),
),
});
const pkt = create(P.Mesh.MeshPacketSchema, {
from,
id: 1,
payloadVariant: { case: 'decoded', value: data },
});
return frame(
toBinary(
P.Mesh.FromRadioSchema,
create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'packet', value: pkt },
}),
),
);
}
export const KAMIL_NUM = 123456789; export const KAMIL_NUM = 123456789;
export function kamilNodeInfo(): NodeInfoFields { export function kamilNodeInfo(): NodeInfoFields {
+22 -1
View File
@@ -8,6 +8,7 @@ import {
kamilNodeInfo, kamilNodeInfo,
nodeInfoFrame, nodeInfoFrame,
startBridge, startBridge,
telemetryPacketFrame,
waitFor, waitFor,
} from './fake-bridge.js'; } from './fake-bridge.js';
import { createMeshMcpServer } from './mcp.controller.js'; import { createMeshMcpServer } from './mcp.controller.js';
@@ -64,7 +65,7 @@ async function connectMcp(svc: MeshService): Promise<{
} }
describe('MCP server tools', () => { describe('MCP server tools', () => {
it('lists all seven tools with descriptions and input schemas', async () => { it('lists all eight tools with descriptions and input schemas', async () => {
const env = await connectedService(); const env = await connectedService();
try { try {
const mcp = await connectMcp(env.svc); const mcp = await connectMcp(env.svc);
@@ -74,6 +75,7 @@ describe('MCP server tools', () => {
'device_info', 'device_info',
'get_messages', 'get_messages',
'get_node', 'get_node',
'get_telemetry',
'list_channels', 'list_channels',
'list_nodes', 'list_nodes',
'mesh_status', 'mesh_status',
@@ -263,6 +265,25 @@ describe('MCP server tools', () => {
), ),
) as { dir: string; to?: number }[]; ) as { dir: string; to?: number }[];
expect(msgs[0]).toMatchObject({ dir: 'tx', to: BROADCAST_ADDR }); // newest first expect(msgs[0]).toMatchObject({ dir: 'tx', to: BROADCAST_ADDR }); // newest first
env.sockets[0].write(
telemetryPacketFrame(KAMIL_NUM, {
case: 'deviceMetrics',
value: { batteryLevel: 87 },
}),
);
await waitFor(() => env.svc.telemetry().length === 1);
const tel = JSON.parse(
toolText(
await mcp.client.callTool({ name: 'get_telemetry', arguments: {} }),
),
) as { from: number; kind: string; data: object }[];
expect(tel).toHaveLength(1);
expect(tel[0]).toMatchObject({
from: KAMIL_NUM,
kind: 'deviceMetrics',
});
expect(tel[0].data).toEqual({ batteryLevel: 87 });
} finally { } finally {
await mcp.close(); await mcp.close();
} }
+8
View File
@@ -77,6 +77,14 @@ export function createMeshMcpServer(mesh: MeshService): McpServer {
async () => jsonResult(mesh.messages()), async () => jsonResult(mesh.messages()),
); );
server.registerTool(
'get_telemetry',
{
description: `Telemetry history, newest first (bounded recent window, last 100): TELEMETRY_APP packets from any node — device metrics (uptime/battery/voltage/channel+air utilization) and environment sensor readings where nodes have them. kind is the meshtastic.Telemetry oneof case name; data carries that variant's payload (proto uint64 fields as decimal strings). In-memory only, reset on every bridge reconnect.${CHANNEL_NOTE}`,
},
async () => jsonResult(mesh.telemetry()),
);
server.registerTool( server.registerTool(
'device_info', 'device_info',
{ {
+34
View File
@@ -183,6 +183,40 @@ export class MeshController {
return this.mesh.messages(); return this.mesh.messages();
} }
@Get('telemetry')
@ApiOperation({
summary: 'Telemetry history (last 100)',
description:
"TELEMETRY_APP packets from any node, newest first. Device metrics (uptime/battery/voltage/channel+air utilization) and environment sensor readings where nodes have them; kind is the meshtastic.Telemetry oneof case name — deviceMetrics, environmentMetrics, airQualityMetrics, powerMetrics, localStats, healthMetrics, hostMetrics or trafficManagementStats — and data carries that variant's payload (proto uint64 fields as decimal strings). In-memory only (last 100), reset on every bridge reconnect.",
})
@ApiResponse({
status: 200,
description: 'Telemetry records newest first',
example: [
{
from: 987654321,
at: 1758000123456,
kind: 'deviceMetrics',
data: {
batteryLevel: 87,
voltage: 4.1,
channelUtilization: 12.5,
airUtilTx: 3.2,
uptimeSeconds: 86400,
},
},
{
from: 123456789,
at: 1758000000000,
kind: 'environmentMetrics',
data: { temperature: 21.5, relativeHumidity: 48 },
},
],
})
telemetry() {
return this.mesh.telemetry();
}
@Get('nodes') @Get('nodes')
@ApiOperation({ @ApiOperation({
summary: 'Known mesh nodes (lean summaries)', summary: 'Known mesh nodes (lean summaries)',
+116
View File
@@ -10,6 +10,7 @@ import {
moduleConfigSectionFrame, moduleConfigSectionFrame,
nodeInfoFrame, nodeInfoFrame,
startBridge, startBridge,
telemetryPacketFrame,
waitFor, waitFor,
} from './fake-bridge.js'; } from './fake-bridge.js';
import { MeshService } from './mesh.service.js'; import { MeshService } from './mesh.service.js';
@@ -560,6 +561,121 @@ describe('MeshService message history', () => {
}); });
}); });
describe('MeshService telemetry', () => {
it('captures device metrics and environment variants with from/kind/data', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
expect(svc.telemetry()).toEqual([]); // nothing before the first packet
bridge.sockets[0].write(
telemetryPacketFrame(KAMIL_NUM, {
case: 'deviceMetrics',
value: { batteryLevel: 87, voltage: 4.1, uptimeSeconds: 3600 },
}),
);
bridge.sockets[0].write(
telemetryPacketFrame(987654321, {
case: 'environmentMetrics',
value: { temperature: 21.5, relativeHumidity: 48 },
}),
);
await waitFor(() => svc.telemetry().length === 2);
expect(svc.telemetry()).toEqual([
{
from: 987654321, // newest first
at: expect.any(Number),
kind: 'environmentMetrics',
// the deprecated repeated one_wire_temperature field decodes as an empty list even when unset
data: {
temperature: 21.5,
relativeHumidity: 48,
oneWireTemperature: [],
},
},
{
from: KAMIL_NUM,
at: expect.any(Number),
kind: 'deviceMetrics',
// proto float is IEEE-754 single precision — 4.1 widens to this exact double
data: {
batteryLevel: 87,
voltage: 4.099999904632568,
uptimeSeconds: 3600,
},
},
]);
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('caps telemetry at 100 records, dropping the oldest', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
for (let i = 0; i < 105; i++)
bridge.sockets[0].write(
telemetryPacketFrame(42, {
case: 'deviceMetrics',
value: { batteryLevel: i },
}),
);
await waitFor(() => svc.telemetry().length === 100, 10000);
expect(svc.telemetry()[0]?.data).toEqual({ batteryLevel: 104 }); // newest first
expect(svc.telemetry()[99]?.data).toEqual({ batteryLevel: 5 }); // i=0..4 dropped
} finally {
svc.onModuleDestroy();
bridge.close();
}
});
it('resets telemetry on reconnect (no stale data)', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
process.env.MESH_RECONNECT_MS = '50';
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
bridge.sockets[0].write(
telemetryPacketFrame(KAMIL_NUM, { case: 'deviceMetrics', value: {} }),
);
await waitFor(() => svc.telemetry().length === 1);
// drop the connection; bot reconnects and must start with an empty history
bridge.sockets[0].destroy();
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
await waitFor(() => svc.telemetry().length === 0);
} finally {
delete process.env.MESH_RECONNECT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService resolveDest', () => { describe('MeshService resolveDest', () => {
it('resolves broadcast sentinels, numbers and names to node numbers', async () => { it('resolves broadcast sentinels, numbers and names to node numbers', async () => {
const bridge = await startBridge(); const bridge = await startBridge();
+55 -2
View File
@@ -11,12 +11,14 @@ import {
BROADCAST_ADDR, BROADCAST_ADDR,
PORT_POSITION_APP, PORT_POSITION_APP,
PORT_ROUTING_APP, PORT_ROUTING_APP,
PORT_TELEMETRY_APP,
PORT_TEXT_MESSAGE_APP, PORT_TEXT_MESSAGE_APP,
buildTextToRadio, buildTextToRadio,
buildWantConfigToRadio, buildWantConfigToRadio,
decodeFromRadio, decodeFromRadio,
decodePosition, decodePosition,
decodeRouting, decodeRouting,
decodeTelemetry,
describeRoutingError, describeRoutingError,
extractFrames, extractFrames,
frame, frame,
@@ -41,6 +43,18 @@ export interface MessageRecord {
ackError?: string; 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 // bounded history: keep the newest N records, in-memory only
const HISTORY_LIMIT = 100; const HISTORY_LIMIT = 100;
@@ -77,13 +91,15 @@ export interface DeviceInfo {
} }
// protobuf-es decodes proto uint64 fields as native bigint, which JSON.stringify rejects; // 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 { function jsonSafe<T>(value: T): T {
if (typeof value === 'bigint') return String(value) as T; if (typeof value === 'bigint') return String(value) as T;
if (Array.isArray(value)) return value.map(jsonSafe) as T; if (Array.isArray(value)) return value.map(jsonSafe) as T;
if (value !== null && typeof value === 'object') { if (value !== null && typeof value === 'object') {
const out: Record<string, unknown> = {}; 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 out as T;
} }
return value; // primitives + undefined pass through 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 // message history (newest kept, capped at HISTORY_LIMIT) + pending tx records awaiting their routing ACK/NACK
private history: MessageRecord[] = []; private history: MessageRecord[] = [];
private pendingTx = new Map<number, 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 reconnectTimer: NodeJS.Timeout | null = null;
private shuttingDown = false; private shuttingDown = false;
@@ -285,6 +303,16 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
return [...this.history].reverse(); 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. */ /** 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> { async sendText(to: number, text: string): Promise<number> {
if (!this.connected || !this.socket) if (!this.connected || !this.socket)
@@ -311,6 +339,13 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
if (this.history.length > HISTORY_LIMIT) this.history.shift(); 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 { private summarize(n: NodeEntry, names: Map<number, string>): NodeSummary {
const s: NodeSummary = { num: n.num, lastHeard: n.lastHeard || null }; const s: NodeSummary = { num: n.num, lastHeard: n.lastHeard || null };
if (n.user?.longName) s.longName = n.user.longName; if (n.user?.longName) s.longName = n.user.longName;
@@ -347,6 +382,7 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
this.configSections.clear(); this.configSections.clear();
this.moduleConfigSections.clear(); this.moduleConfigSections.clear();
this.history = []; this.history = [];
this.telemetryHistory = [];
this.pendingTx.clear(); this.pendingTx.clear();
this.logger.log(`bridge connected ${this.host}:${this.port}`); this.logger.log(`bridge connected ${this.host}:${this.port}`);
const nonce = Math.floor(Math.random() * 0xffffffff) || 1; const nonce = Math.floor(Math.random() * 0xffffffff) || 1;
@@ -481,6 +517,23 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
this.pendingTx.delete(decoded.requestId); 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; break;
} }