Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ jobs:

- uses: actions/setup-node@v4
with:
node-version: 22
node-version: 24

- name: Install
run: npm ci
Expand Down
10 changes: 1 addition & 9 deletions .github/workflows/release.yml
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ jobs:

- uses: actions/setup-node@v4
with:
node-version: 22
node-version: 24
registry-url: https://registry.npmjs.org

- name: Install
Expand All @@ -42,14 +42,6 @@ jobs:
- name: Publish @sixfathoms/lplex
working-directory: packages/lplex
run: npm publish --provenance --access public
env:
NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}

- name: Publish @sixfathoms/lplex-cli
working-directory: packages/lplex-cli
run: npm publish --provenance --access public
env:
NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}

- name: Create GitHub Release
run: gh release create "$GITHUB_REF_NAME" --generate-notes
Expand Down
6 changes: 3 additions & 3 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

5 changes: 3 additions & 2 deletions packages/lplex-cli/package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
{
"name": "@sixfathoms/lplex-cli",
"version": "0.1.0",
"private": true,
"version": "0.0.0",
"description": "CLI client for lplex CAN bus HTTP bridge (like lplexdump, but in TypeScript)",
"type": "module",
"bin": {
Expand All @@ -11,7 +12,7 @@
"typecheck": "tsc --noEmit"
},
"dependencies": {
"@sixfathoms/lplex": "^0.1.0",
"@sixfathoms/lplex": "*",
"bonjour-service": "^1.3.0"
},
"devDependencies": {
Expand Down
303 changes: 303 additions & 0 deletions packages/lplex/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,303 @@
# @sixfathoms/lplex

TypeScript client for [lplex](https://github.com/sixfathoms/lplex), a CAN bus HTTP bridge for NMEA 2000.

Zero runtime dependencies. Works in browsers and Node 18+. Ships ESM, CJS, and TypeScript declarations.

## Install

```bash
npm install @sixfathoms/lplex
```

## Quick Start

```typescript
import { Client } from "@sixfathoms/lplex";

const client = new Client("http://localhost:8089");

// list devices on the bus
const devices = await client.devices();

// stream frames
const stream = await client.subscribe({ pgn: [129025] });
for await (const event of stream) {
if (event.type === "frame") {
console.log(event.frame.pgn, event.frame.data);
}
}
```

## API

### `new Client(baseURL, options?)`

Creates a client connected to an lplex server.

```typescript
const client = new Client("http://inuc1.local:8089");
```

Inject a custom `fetch` for testing or environments without a global one:

```typescript
const client = new Client("http://localhost:8089", {
fetch: myCustomFetch,
});
```

### `client.devices(signal?): Promise<Device[]>`

Returns a snapshot of all NMEA 2000 devices discovered on the bus.

```typescript
const devices = await client.devices();
for (const d of devices) {
console.log(`${d.manufacturer} (src=${d.src}): ${d.packet_count} packets`);
}
```

### `client.subscribe(filter?, signal?): Promise<AsyncIterable<Event>>`

Opens an ephemeral SSE stream. No session state, no replay. Frames flow until you stop reading or abort.

```typescript
const stream = await client.subscribe();

for await (const event of stream) {
switch (event.type) {
case "frame":
console.log(event.frame.pgn, event.frame.src, event.frame.data);
break;
case "device":
console.log("device:", event.device.manufacturer, event.device.src);
break;
}
}
```

#### Filtering

Pass a `Filter` to narrow the stream. Categories are AND'd, values within a category are OR'd.

```typescript
const stream = await client.subscribe({
pgn: [129025, 129026], // Position Rapid OR COG/SOG Rapid
manufacturer: ["Garmin"], // AND from Garmin
});
```

#### Cancellation

Use an `AbortSignal` to stop the stream:

```typescript
const ac = new AbortController();
setTimeout(() => ac.abort(), 10_000);

const stream = await client.subscribe(undefined, ac.signal);
for await (const event of stream) {
console.log(event);
}
// loop exits when aborted
```

### `client.send(params, signal?): Promise<void>`

Transmit a CAN frame through the server to the bus.

```typescript
await client.send({
pgn: 129025,
src: 0,
dst: 255,
prio: 6,
data: "00aabbccddee",
});
```

### `client.createSession(config, signal?): Promise<Session>`

Creates or reconnects a buffered session. The server buffers frames while you're disconnected. On reconnect, you pick up where you left off.

```typescript
const session = await client.createSession({
clientId: "my-dashboard",
bufferTimeout: "PT5M",
filter: { pgn: [129025] },
});

console.log(`cursor at ${session.info.cursor}, head at ${session.info.seq}`);

const stream = await session.subscribe();
let lastSeq = 0;

for await (const event of stream) {
if (event.type === "frame") {
lastSeq = event.frame.seq;
console.log(JSON.stringify(event.frame));
}
}

// advance the cursor so the server can free buffer space
await session.ack(lastSeq);
```

### `session.subscribe(signal?): Promise<AsyncIterable<Event>>`

Opens the SSE stream for a buffered session. Replays from the cursor, then streams live.

### `session.ack(seq, signal?): Promise<void>`

Advances the cursor to the given sequence number.

### `session.info: SessionInfo`

The session metadata returned by the server on create/reconnect.

### `session.lastAckedSeq: number`

The last sequence number successfully ACK'd (0 if never ACK'd).

## Error Handling

All methods throw `HttpError` on non-success HTTP responses:

```typescript
import { HttpError } from "@sixfathoms/lplex";

try {
await client.devices();
} catch (err) {
if (err instanceof HttpError) {
console.error(`HTTP ${err.status}: ${err.body}`);
}
}
```

## Browser Usage

The library uses only web platform APIs (`fetch`, `ReadableStream`, `TextDecoder`, `AbortSignal`), so it works in any modern browser without polyfills.

```html
<script type="module">
import { Client } from "https://esm.sh/@sixfathoms/lplex";

const client = new Client("http://your-lplex-server:8089");
const devices = await client.devices();
console.log(devices);
</script>
```

### React

```tsx
import { useState, useEffect } from "react";
import { Client, type Frame, type Filter } from "@sixfathoms/lplex";

function useFrames(serverUrl: string, filter?: Filter) {
const [frames, setFrames] = useState<Frame[]>([]);

useEffect(() => {
const ac = new AbortController();
const client = new Client(serverUrl);

(async () => {
const stream = await client.subscribe(filter, ac.signal);
for await (const event of stream) {
if (event.type === "frame") {
setFrames((prev) => [...prev.slice(-99), event.frame]);
}
}
})().catch(() => {});

return () => ac.abort();
}, [serverUrl]);

return frames;
}
```

## Types

All interfaces use `snake_case` field names matching the server's JSON wire format. No mapping layer.

```typescript
interface Frame {
seq: number; // monotonic, starts at 1
ts: string; // RFC 3339 timestamp
prio: number; // 0-7
pgn: number; // Parameter Group Number
src: number; // source address (0-253)
dst: number; // destination (255 = broadcast)
data: string; // hex-encoded payload
}

interface Device {
src: number;
name: string;
manufacturer: string;
manufacturer_code: number;
device_class: number;
device_function: number;
device_instance: number;
unique_number: number;
model_id: string;
software_version: string;
model_version: string;
model_serial: string;
product_code: number;
first_seen: string;
last_seen: string;
packet_count: number;
byte_count: number;
}

type Event =
| { type: "frame"; frame: Frame }
| { type: "device"; device: Device };

interface Filter {
pgn?: number[];
manufacturer?: string[];
instance?: number[];
name?: string[];
}

interface SessionConfig {
clientId: string;
bufferTimeout: string; // ISO 8601 duration ("PT5M", "PT1H")
filter?: Filter;
}

interface SessionInfo {
client_id: string;
seq: number; // current head
cursor: number; // last ACK'd (0 = never)
devices: Device[];
}

interface SendParams {
pgn: number;
src: number;
dst: number;
prio: number;
data: string; // hex-encoded
}
```

## Server Endpoints

| Endpoint | Method | Purpose |
|---|---|---|
| `/events` | GET | Ephemeral SSE stream. Query params: `pgn`, `manufacturer`, `instance`, `name` (repeatable). |
| `/clients/{id}` | PUT | Create/reconnect buffered session. JSON body: `buffer_timeout`, `filter`. |
| `/clients/{id}/events` | GET | Buffered SSE stream with replay from cursor. |
| `/clients/{id}/ack` | PUT | ACK sequence number. JSON body: `{ "seq": N }`. Returns 204. |
| `/send` | POST | Transmit CAN frame. JSON body: `pgn`, `src`, `dst`, `prio`, `data`. Returns 202. |
| `/devices` | GET | Device snapshot. Returns JSON array. |

## License

MIT
4 changes: 2 additions & 2 deletions packages/lplex/package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "@sixfathoms/lplex",
"version": "0.1.0",
"version": "0.0.0",
"description": "TypeScript client for lplex CAN bus HTTP bridge",
"type": "module",
"main": "./dist/index.cjs",
Expand All @@ -18,7 +18,7 @@
}
}
},
"files": ["dist"],
"files": ["dist", "README.md"],
"scripts": {
"build": "tsup",
"test": "vitest run",
Expand Down