diff --git a/src/actions/gpu-graph.ts b/src/actions/gpu-graph.ts index e56ea0a..44eb9de 100644 --- a/src/actions/gpu-graph.ts +++ b/src/actions/gpu-graph.ts @@ -5,6 +5,7 @@ import streamDeck, { type KeyDownEvent, SingletonAction, type WillAppearEvent, + type WillDisappearEvent, } from "@elgato/streamdeck"; import { type GpuMetricKind } from "../lib/metrics-parser"; import { renderGpuGraph, svgDataUrl } from "../lib/render"; @@ -16,52 +17,69 @@ type GpuSettings = CfgSettings & { metric?: GpuMetricKind; }; +type GpuState = { + settings: GpuSettings; + action?: KeyAction; + unsubscribe?: () => void; + lastSvg?: string; +}; + @action({ UUID: "com.bryce.llamawatch.gpu" }) export class GpuGraph extends SingletonAction { - private settings: GpuSettings = {}; - private action?: KeyAction; - private unsubscribe?: () => void; - private lastSvg?: string; + private states = new Map(); + + private stateFor(action: KeyAction): GpuState { + let state = this.states.get(action.id); + if (!state) { + state = { settings: {} }; + this.states.set(action.id, state); + } + return state; + } override onWillAppear(ev: WillAppearEvent): void { if (!ev.action.isKey()) return; - this.settings = ev.payload.settings; - this.action = ev.action; - runtime.ensureConnections(cfgFromSettings(this.settings)); - this.unsubscribe = runtime.subscribe(() => this.render()); - this.render(); + const state = this.stateFor(ev.action); + state.settings = ev.payload.settings; + state.action = ev.action; + runtime.ensureConnections(cfgFromSettings(state.settings)); + state.unsubscribe = runtime.subscribe(() => this.render(state)); + this.render(state); } - override onWillDisappear(): void { - this.unsubscribe?.(); - this.unsubscribe = undefined; - this.action = undefined; + override onWillDisappear(ev: WillDisappearEvent): void { + const state = this.states.get(ev.action.id); + if (!state) return; + state.unsubscribe?.(); + this.states.delete(ev.action.id); } override onDidReceiveSettings(ev: DidReceiveSettingsEvent): void { if (!ev.action.isKey()) return; - this.settings = ev.payload.settings; - this.action = ev.action; - this.render(); + const state = this.stateFor(ev.action); + state.settings = ev.payload.settings; + state.action = ev.action; + runtime.ensureConnections(cfgFromSettings(state.settings)); + this.render(state); } override onKeyDown(ev: KeyDownEvent): void { void streamDeck.system.openUrl(`${cfgFromSettings(ev.payload.settings).baseUrl}/ui`); } - private render(): void { - const action = this.action; + private render(state: GpuState): void { + const action = state.action; if (!action) return; const poller = runtime.pollerInstance; - const gpuId = this.settings.gpuId ?? "all"; - const metric = this.settings.metric ?? "util_percent"; + const gpuId = state.settings.gpuId ?? "all"; + const metric = state.settings.metric ?? "util_percent"; const value = poller?.getValue(gpuId, metric); const history = poller?.getHistory(gpuId, metric) ?? []; const offline = !poller || poller.isOffline(); const gpuName = gpuId === "all" ? "ALL GPUS" : this.gpuLabel(gpuId); const svg = renderGpuGraph({ gpuName, metric, value, history, offline }); - if (svg === this.lastSvg) return; - this.lastSvg = svg; + if (svg === state.lastSvg) return; + state.lastSvg = svg; void action.setImage(svgDataUrl(svg)); } diff --git a/src/actions/inflight-monitor.ts b/src/actions/inflight-monitor.ts index aaea232..ea4bd2e 100644 --- a/src/actions/inflight-monitor.ts +++ b/src/actions/inflight-monitor.ts @@ -5,6 +5,7 @@ import streamDeck, { type KeyDownEvent, SingletonAction, type WillAppearEvent, + type WillDisappearEvent, } from "@elgato/streamdeck"; import { renderInflight, svgDataUrl } from "../lib/render"; import { runtime } from "../lib/runtime"; @@ -14,74 +15,90 @@ type InflightSettings = CfgSettings & { modelId?: string; }; +type InflightState = { + settings: InflightSettings; + action?: KeyAction; + unsubscribe?: () => void; + pulseTimer?: ReturnType; + pulse: boolean; +}; + @action({ UUID: "com.bryce.llamawatch.inflight" }) export class InflightMonitor extends SingletonAction { - private settings: InflightSettings = {}; - private action?: KeyAction; - private unsubscribe?: () => void; - private pulseTimer?: ReturnType; - private pulse = false; + private states = new Map(); + + private stateFor(action: KeyAction): InflightState { + let state = this.states.get(action.id); + if (!state) { + state = { settings: {}, pulse: false }; + this.states.set(action.id, state); + } + return state; + } override onWillAppear(ev: WillAppearEvent): void { if (!ev.action.isKey()) return; - this.settings = ev.payload.settings; - this.action = ev.action; - runtime.ensureConnections(cfgFromSettings(this.settings)); - this.unsubscribe = runtime.subscribe(() => this.render()); - this.render(); + const state = this.stateFor(ev.action); + state.settings = ev.payload.settings; + state.action = ev.action; + runtime.ensureConnections(cfgFromSettings(state.settings)); + state.unsubscribe = runtime.subscribe(() => this.render(state)); + this.render(state); } - override onWillDisappear(): void { - this.unsubscribe?.(); - this.unsubscribe = undefined; - this.action = undefined; - this.clearPulse(); + override onWillDisappear(ev: WillDisappearEvent): void { + const state = this.states.get(ev.action.id); + if (!state) return; + state.unsubscribe?.(); + this.clearPulse(state); + this.states.delete(ev.action.id); } override onDidReceiveSettings(ev: DidReceiveSettingsEvent): void { if (!ev.action.isKey()) return; - this.settings = ev.payload.settings; - this.action = ev.action; - runtime.ensureConnections(cfgFromSettings(this.settings)); - this.render(); + const state = this.stateFor(ev.action); + state.settings = ev.payload.settings; + state.action = ev.action; + runtime.ensureConnections(cfgFromSettings(state.settings)); + this.render(state); } override onKeyDown(ev: KeyDownEvent): void { void streamDeck.system.openUrl(`${cfgFromSettings(ev.payload.settings).baseUrl}/ui`); } - private render(): void { - const action = this.action; + private render(state: InflightState): void { + const action = state.action; if (!action) return; - const modelId = this.settings.modelId ?? ""; - const state = runtime.tracker.state(modelId); + const modelId = state.settings.modelId ?? ""; + const trackerState = runtime.tracker.state(modelId); const count = runtime.tracker.count(modelId); const offline = runtime.offline && modelId.length > 0; void action.setImage( svgDataUrl( renderInflight({ modelName: modelId.length > 0 ? modelId : "unset", - state, + state: trackerState, count, offline, - pulse: this.pulse, + pulse: state.pulse, }), ), ); - if (state === "ready" && count > 0 && !this.pulseTimer) { - this.pulseTimer = setInterval(() => { - this.pulse = !this.pulse; - this.render(); + if (trackerState === "ready" && count > 0 && !state.pulseTimer) { + state.pulseTimer = setInterval(() => { + state.pulse = !state.pulse; + this.render(state); }, 500); - } else if (!(state === "ready" && count > 0) && this.pulseTimer) { - this.clearPulse(); + } else if (!(trackerState === "ready" && count > 0) && state.pulseTimer) { + this.clearPulse(state); } } - private clearPulse(): void { - if (this.pulseTimer) clearInterval(this.pulseTimer); - this.pulseTimer = undefined; - this.pulse = false; + private clearPulse(state: InflightState): void { + if (state.pulseTimer) clearInterval(state.pulseTimer); + state.pulseTimer = undefined; + state.pulse = false; } } diff --git a/src/lib/datasources.ts b/src/lib/datasources.ts index 0d98b12..11dc102 100644 --- a/src/lib/datasources.ts +++ b/src/lib/datasources.ts @@ -7,36 +7,39 @@ type DataSourceItem = { label: string; value: string }; export function registerDataSources(): void { streamDeck.ui.onSendToPlugin(async (ev) => { - const request = ev.payload as { event?: string } | undefined; - const event = request?.event; - if (!event) return; - const settings = await ev.action.getSettings(); + try { + const request = ev.payload as { event?: string } | undefined; + const event = request?.event; + if (!event) return; + const settings = await ev.action.getSettings(); - if (event === "models") { - let items: DataSourceItem[] = []; - try { - const models = await fetchModels(cfgFromSettings(settings)); - items = models.map((m) => ({ label: `${m.id} (${m.status})`, value: m.id })); - } catch { - items = []; + if (event === "models") { + let items: DataSourceItem[] = []; + try { + const models = await fetchModels(cfgFromSettings(settings)); + items = models.map((m) => ({ label: `${m.id} (${m.status})`, value: m.id })); + } catch { + items = []; + } + await streamDeck.ui.sendToPropertyInspector({ event, items }); + return; } - await streamDeck.ui.sendToPropertyInspector({ event, items }); - return; - } - if (event === "gpus") { - let items: DataSourceItem[] = []; - try { - const text = await fetchMetrics(cfgFromSettings(settings)); - const gpus = gpuInfos(parseGpuMetrics(text)); - items = [ - { label: "All GPUs", value: "all" }, - ...gpus.map((g) => ({ label: `GPU ${g.id} · ${g.name}`, value: g.id })), - ]; - } catch { - items = [{ label: "All GPUs", value: "all" }]; + if (event === "gpus") { + let items: DataSourceItem[] = []; + try { + const text = await fetchMetrics(cfgFromSettings(settings)); + const gpus = gpuInfos(parseGpuMetrics(text)); + items = [ + { label: "All GPUs", value: "all" }, + ...gpus.map((g) => ({ label: `GPU ${g.id} · ${g.name}`, value: g.id })), + ]; + } catch { + items = [{ label: "All GPUs", value: "all" }]; + } + await streamDeck.ui.sendToPropertyInspector({ event, items }); } - await streamDeck.ui.sendToPropertyInspector({ event, items }); + } catch { } }); } diff --git a/src/lib/event-feed.ts b/src/lib/event-feed.ts index e955902..411ce08 100644 --- a/src/lib/event-feed.ts +++ b/src/lib/event-feed.ts @@ -9,8 +9,13 @@ export function decodeEvent(msg: SseMessage): FeedEvent | null { const inner = JSON.parse(outer.data) as Record; if (outer.type === "inflight") { - const requests: InflightRequest[] = - (inner.requests as InflightRequest[]) ?? (inner.request ? [inner.request as InflightRequest] : []); + const requests: InflightRequest[] = Array.isArray(inner.requests) + ? (inner.requests as InflightRequest[]) + : Array.isArray(inner.request) + ? (inner.request as InflightRequest[]) + : inner.request && typeof inner.request === "object" + ? [inner.request as InflightRequest] + : []; return { type: "inflight", operation: inner.operation as "snapshot" | "add" | "remove", @@ -19,6 +24,7 @@ export function decodeEvent(msg: SseMessage): FeedEvent | null { } if (outer.type === "modelStatus") { + if (!Array.isArray(inner)) return null; return { type: "modelStatus", models: inner as unknown as ModelState[] }; } diff --git a/src/lib/metrics-poller.ts b/src/lib/metrics-poller.ts index 920c4b4..f0a5034 100644 --- a/src/lib/metrics-poller.ts +++ b/src/lib/metrics-poller.ts @@ -1,6 +1,6 @@ import { fetchMetrics } from "./llamaswap"; import { type LlamaSwapConfig } from "./util"; -import { gpuInfos, parseGpuMetrics, type GpuMetricKind, type GpuSample } from "./metrics-parser"; +import { parseGpuMetrics, type GpuMetricKind, type GpuSample } from "./metrics-parser"; const RING_SIZE = 60; @@ -34,6 +34,7 @@ export class MetricsPoller { private gpuIds: string[] = []; private listeners = new Set<() => void>(); private timer?: ReturnType; + private ticking = false; private lastError?: string; constructor( @@ -43,6 +44,7 @@ export class MetricsPoller { ) {} start(): void { + if (this.timer) return; void this.tick(); this.timer = setInterval(() => void this.tick(), this.intervalMs); } @@ -60,6 +62,8 @@ export class MetricsPoller { } async tick(): Promise { + if (this.ticking) return; + this.ticking = true; try { const text = await this.fetchFn(this.cfg); this.apply(parseGpuMetrics(text)); @@ -67,6 +71,7 @@ export class MetricsPoller { } catch (err) { this.lastError = err instanceof Error ? err.message : String(err); } finally { + this.ticking = false; for (const listener of this.listeners) listener(); } } diff --git a/src/lib/runtime.ts b/src/lib/runtime.ts index f5e89e0..2f68b0f 100644 --- a/src/lib/runtime.ts +++ b/src/lib/runtime.ts @@ -6,11 +6,20 @@ import { type LlamaSwapConfig } from "./util"; class Runtime { readonly tracker = new InflightTracker(); offline = true; + private cfg?: LlamaSwapConfig; private feed?: EventFeed; private poller?: MetricsPoller; private listeners = new Set<() => void>(); ensureConnections(cfg: LlamaSwapConfig): void { + const changed = !this.cfg || this.cfg.baseUrl !== cfg.baseUrl || this.cfg.apiKey !== cfg.apiKey; + if (changed) { + this.feed?.stop(); + this.poller?.stop(); + this.feed = undefined; + this.poller = undefined; + this.cfg = cfg; + } if (!this.feed) { this.feed = new EventFeed(cfg, (ev) => { this.tracker.apply(ev); diff --git a/src/lib/sse.ts b/src/lib/sse.ts index 01fa21b..e8ff2a4 100644 --- a/src/lib/sse.ts +++ b/src/lib/sse.ts @@ -4,6 +4,7 @@ export interface SseMessage { } export function parseSse(chunk: string): { messages: SseMessage[]; rest: string } { + chunk = chunk.replace(/\r\n/g, "\n"); const messages: SseMessage[] = []; let rest = chunk; while (true) { diff --git a/tests/event-feed.test.ts b/tests/event-feed.test.ts index 3ec6d0f..52167bd 100644 --- a/tests/event-feed.test.ts +++ b/tests/event-feed.test.ts @@ -64,3 +64,30 @@ test("decodeEvent returns null for unrelated or malformed events", () => { assert.equal(decodeEvent({ event: "message", data: JSON.stringify({ type: "logData", data: "{}" }) }), null); assert.equal(decodeEvent({ event: "message", data: "not json" }), null); }); + +test("decodeEvent returns null when modelStatus data is not an array", () => { + const msg: SseMessage = { + event: "message", + data: JSON.stringify({ + type: "modelStatus", + data: JSON.stringify({ id: "DeepSeek-V4-Flash-0731", state: "ready" }), + }), + }; + assert.equal(decodeEvent(msg), null); +}); + +test("decodeEvent handles inflight with a non-array requests field and no request field", () => { + const msg: SseMessage = { + event: "message", + data: JSON.stringify({ + type: "inflight", + data: JSON.stringify({ operation: "snapshot", requests: "oops" }), + }), + }; + const ev = decodeEvent(msg); + assert.ok(ev && ev.type === "inflight"); + if (ev && ev.type === "inflight") { + assert.equal(ev.operation, "snapshot"); + assert.deepEqual(ev.requests, []); + } +}); diff --git a/tests/metrics-poller.test.ts b/tests/metrics-poller.test.ts index 56c4d07..79ff4ac 100644 --- a/tests/metrics-poller.test.ts +++ b/tests/metrics-poller.test.ts @@ -3,7 +3,7 @@ import { readFileSync } from "node:fs"; import { test } from "node:test"; import { type LlamaSwapConfig } from "../src/lib/util"; import { MetricsPoller } from "../src/lib/metrics-poller"; -import { parseGpuMetrics } from "../src/lib/metrics-parser"; + const FIXTURE = readFileSync(new URL("./fixtures/metrics.txt", import.meta.url), "utf8"); const FIXTURE2 = FIXTURE diff --git a/tests/sse.test.ts b/tests/sse.test.ts index 4c71b43..da2d41c 100644 --- a/tests/sse.test.ts +++ b/tests/sse.test.ts @@ -22,3 +22,10 @@ test("parseSse joins multi-line data fields with newline", () => { assert.equal(messages.length, 1); assert.equal(messages[0].data, "line1\nline2"); }); + +test("parseSse normalizes CRLF-delimited events", () => { + const { messages, rest } = parseSse("data:hello\r\n\r\n"); + assert.equal(rest, ""); + assert.equal(messages.length, 1); + assert.equal(messages[0].data, "hello"); +});