Skip to content

Commit 1015010

Browse files
haoruizhouhaoruizhou
committed
Updated WSS.ts for consistency between pecan and FDR; UI change for FDR
1 parent 536327c commit 1015010

6 files changed

Lines changed: 457 additions & 322 deletions

File tree

‎flight-recorder/src/App.tsx‎

Lines changed: 169 additions & 109 deletions
Large diffs are not rendered by default.

‎flight-recorder/src/services/LoggingHandler.ts‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
import { webSocketService } from './WebSocketService';
2+
import { loggingService } from './LoggingService';
3+
4+
class LoggingHandler {
5+
private isInitialized = false;
6+
7+
initialize() {
8+
if (this.isInitialized) return;
9+
10+
webSocketService.on('decoded', (decoded: any) => {
11+
const messages = Array.isArray(decoded) ? decoded : [decoded];
12+
13+
messages.forEach(msg => {
14+
if (msg?.signals) {
15+
loggingService.logFrame(
16+
msg.time || Date.now(),
17+
msg.canId,
18+
msg.data // Using raw buffer data from processor
19+
);
20+
}
21+
});
22+
});
23+
24+
this.isInitialized = true;
25+
console.log('[LoggingHandler] Initialized');
26+
}
27+
}
28+
29+
export const loggingHandler = new LoggingHandler();

‎flight-recorder/src/services/WebSocketService.ts‎

Lines changed: 65 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,33 @@
1-
import { createCanProcessor, formatCanId } from '../utils/canProcessor';
2-
import { loggingService } from './LoggingService';
1+
import { createCanProcessor } from '../utils/canProcessor';
32

43
export type MessageHandler = (data: any) => void;
54

5+
export interface ConnectionStatus {
6+
connected: boolean;
7+
url: string;
8+
error?: string;
9+
}
10+
611
export class WebSocketService {
712
private ws: WebSocket | null = null;
813
private processor: any = null;
914
private reconnectAttempts = 0;
1015
private maxReconnectAttempts = 5;
1116
private reconnectDelay = 2000;
12-
private messageCount = 0;
17+
private messageCount = 0;
1318
private messageListeners = new Map<string, Set<MessageHandler>>();
1419
private suppressIngestion = false;
1520

1621
async initialize() {
22+
this.disconnect();
23+
1724
try {
18-
this.processor = await createCanProcessor();
19-
console.log('CAN processor initialized');
25+
if (!this.processor) {
26+
this.processor = await createCanProcessor();
27+
console.log('[WebSocket] CAN processor initialized');
28+
}
2029
} catch (error) {
21-
console.error('Failed to initialize CAN processor:', error);
30+
console.error('[WebSocket] Failed to initialize CAN processor:', error);
2231
return;
2332
}
2433
this.connect();
@@ -46,58 +55,66 @@ export class WebSocketService {
4655
}
4756
}
4857

49-
console.log(`Connecting to WebSocket: ${wsUrl}`);
58+
console.log(`[WebSocket] Connecting to: ${wsUrl}`);
5059

5160
try {
5261
this.ws = new WebSocket(wsUrl);
5362

5463
this.ws.onopen = () => {
55-
console.log('WebSocket connected');
64+
console.log('[WebSocket] Connected');
5665
this.reconnectAttempts = 0;
5766
this.messageCount = 0;
67+
this.notify('status', { connected: true, url: wsUrl });
5868
};
5969

6070
this.ws.onmessage = (event) => {
6171
try {
72+
// Milestone and initial message logging (replicated from original pecan)
73+
if (this.messageCount < 3) {
74+
console.log(`[WebSocket] Message #${this.messageCount + 1}:`, event.data);
75+
}
76+
if (this.messageCount === 10 || this.messageCount === 100 || (this.messageCount > 0 && this.messageCount % 1000 === 0)) {
77+
console.log(`[WebSocket] Received ${this.messageCount} messages`);
78+
}
79+
6280
const messageData = JSON.parse(event.data);
6381

64-
if (messageData && typeof messageData === 'object' && messageData.type) {
65-
const listeners = this.messageListeners.get(messageData.type);
66-
if (listeners) {
67-
listeners.forEach(handler => handler(messageData));
68-
}
82+
if (messageData?.type) {
83+
this.notify(messageData.type, messageData);
6984
}
85+
this.notify('raw', messageData);
7086

7187
const decoded = this.processor.processWebSocketMessage(messageData);
72-
if (!decoded) return;
73-
74-
this.messageCount++;
75-
const messages = Array.isArray(decoded) ? decoded : [decoded];
76-
77-
messages.forEach(msg => {
78-
if (msg?.signals) {
79-
if (this.suppressIngestion) return;
80-
loggingService.logFrame(msg.time || Date.now(), msg.canId, msg.data);
88+
if (decoded) {
89+
if (this.messageCount < 3) {
90+
console.log(`[WebSocket] Decoded message(s) #${this.messageCount + 1}:`, decoded);
8191
}
82-
});
92+
this.notify('decoded', decoded);
93+
}
94+
95+
this.messageCount++;
8396
} catch (error) {
84-
console.error('Error processing WebSocket message:', error);
97+
console.error('[WebSocket] Error processing message:', error);
8598
}
8699
};
87100

88101
this.ws.onerror = (error) => {
89-
console.error('WebSocket error:', error);
102+
console.error('[WebSocket] Error:', error);
103+
this.notify('status', { connected: false, url: wsUrl, error: 'Connection error' });
90104
};
91105

92106
this.ws.onclose = (event) => {
93-
console.log('WebSocket disconnected');
107+
console.log('[WebSocket] Disconnected');
108+
this.notify('status', { connected: false, url: wsUrl, error: `Socket closed (code: ${event.code})` });
109+
94110
if (event.code !== 1000 && this.reconnectAttempts < this.maxReconnectAttempts) {
95111
this.reconnectAttempts++;
96112
setTimeout(() => this.connect(), this.reconnectDelay * this.reconnectAttempts);
97113
}
98114
};
99115
} catch (error) {
100-
console.error('WebSocket error:', error);
116+
console.error('[WebSocket] Connection failed:', error);
117+
this.notify('status', { connected: false, url: wsUrl, error: String(error) });
101118
}
102119
}
103120

@@ -118,16 +135,36 @@ export class WebSocketService {
118135
this.messageListeners.get(type)?.delete(handler);
119136
}
120137

138+
private notify(type: string, data: any) {
139+
const listeners = this.messageListeners.get(type);
140+
if (listeners) {
141+
listeners.forEach(handler => handler(data));
142+
}
143+
}
144+
121145
disconnect() {
122146
if (this.ws) {
123-
this.ws.close(1000, 'Client disconnect');
147+
this.ws.close(1000, 'Manual disconnect');
124148
this.ws = null;
125149
}
126150
}
127151

128152
isConnected(): boolean {
129153
return this.ws !== null && this.ws.readyState === WebSocket.OPEN;
130154
}
155+
156+
public setSuppressIngestion(suppress: boolean) {
157+
console.log(`[WebSocket] Ingestion suppression: ${suppress}`);
158+
this.suppressIngestion = suppress;
159+
}
160+
161+
public isIngestionSuppressed(): boolean {
162+
return this.suppressIngestion;
163+
}
164+
165+
public getMessageCount(): number {
166+
return this.messageCount;
167+
}
131168
}
132169

133170
export const webSocketService = new WebSocketService();

‎pecan/src/App.tsx‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import {
1010
} from "./utils/canProcessor";
1111
import { Outlet } from "react-router";
1212
import { webSocketService } from "./services/WebSocketService";
13+
import { telemetryHandler } from "./services/TelemetryHandler";
1314
import { serialService } from "./services/SerialService";
1415
import { DefaultBanner, CacheBanner } from "./components/AppBanners";
1516
import FloatingTools from "./components/FloatingTools";
@@ -78,6 +79,7 @@ function App() {
7879

7980
// Initialize WebSocket service once when app loads
8081
useEffect(() => {
82+
telemetryHandler.initialize();
8183
webSocketService.initialize();
8284

8385
// Cleanup on unmount

‎pecan/src/services/TelemetryHandler.ts‎

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,129 @@
1+
import { webSocketService } from './WebSocketService';
2+
import { dataStore } from '../lib/DataStore';
3+
import { formatCanId } from '../utils/canProcessor';
4+
5+
// Synthetic message IDs for non-CAN diagnostic data
6+
export const DIAG_MSG_IDS = {
7+
SYSTEM_STATS: '__system_stats__',
8+
LINK_PING: '__link_ping__',
9+
LINK_THROUGHPUT: '__link_throughput__',
10+
LINK_RADIO: '__link_radio__',
11+
} as const;
12+
13+
class TelemetryHandler {
14+
private isInitialized = false;
15+
16+
initialize() {
17+
if (this.isInitialized) return;
18+
19+
// Listen for CAN messages
20+
webSocketService.on('decoded', (decoded: any) => {
21+
// Respect ingestion suppression from SerialService
22+
if (webSocketService.isIngestionSuppressed()) {
23+
const count = webSocketService.getMessageCount();
24+
if (count % 100 === 0) {
25+
console.log(`[TelemetryHandler] Suppressing ingestion (Local Connection Active)`);
26+
}
27+
return;
28+
}
29+
30+
const messages = Array.isArray(decoded) ? decoded : [decoded];
31+
messages.forEach(msg => {
32+
if (msg?.signals) {
33+
const hexId = formatCanId(msg.canId);
34+
dataStore.ingestMessage({
35+
msgID: hexId,
36+
messageName: msg.messageName || `CAN_${hexId}`,
37+
data: msg.signals,
38+
rawData: msg.rawData,
39+
timestamp: msg.time || Date.now()
40+
});
41+
}
42+
});
43+
});
44+
45+
// Listen for raw messages to catch diagnostic data
46+
webSocketService.on('raw', (messageData: any) => {
47+
if (webSocketService.isIngestionSuppressed()) return;
48+
this.handleDiagnosticMessage(messageData);
49+
});
50+
51+
this.isInitialized = true;
52+
console.log('[TelemetryHandler] Initialized');
53+
}
54+
55+
private handleDiagnosticMessage(msg: any): boolean {
56+
if (!msg || typeof msg !== 'object' || Array.isArray(msg)) return false;
57+
58+
// system_stats: {"received": N, "missing": N, "recovered": N}
59+
if ('received' in msg && 'missing' in msg && 'recovered' in msg && !('canId' in msg)) {
60+
dataStore.ingestMessage({
61+
msgID: DIAG_MSG_IDS.SYSTEM_STATS,
62+
messageName: 'System Stats',
63+
data: {
64+
received: { sensorReading: Number(msg.received), unit: 'pkt/s' },
65+
missing: { sensorReading: Number(msg.missing), unit: 'pkt/s' },
66+
recovered: { sensorReading: Number(msg.recovered), unit: 'pkt/s' },
67+
},
68+
rawData: '',
69+
timestamp: Date.now(),
70+
});
71+
return true;
72+
}
73+
74+
// link_diagnostics: {"type": "ping"|"throughput"|"radio", ...}
75+
if (msg.type === 'ping' || msg.type === 'throughput' || msg.type === 'radio') {
76+
const ts = typeof msg.ts === 'number' ? msg.ts : Date.now();
77+
78+
if (msg.type === 'ping') {
79+
dataStore.ingestMessage({
80+
msgID: DIAG_MSG_IDS.LINK_PING,
81+
messageName: 'Link Ping',
82+
data: {
83+
rtt_ms: { sensorReading: msg.rtt_ms != null ? Number(msg.rtt_ms) : -1, unit: 'ms' },
84+
},
85+
rawData: '',
86+
timestamp: ts,
87+
});
88+
return true;
89+
}
90+
91+
if (msg.type === 'throughput') {
92+
dataStore.ingestMessage({
93+
msgID: DIAG_MSG_IDS.LINK_THROUGHPUT,
94+
messageName: 'Link Throughput',
95+
data: {
96+
mbps: { sensorReading: msg.mbps != null ? Number(msg.mbps) : -1, unit: 'Mbps' },
97+
loss_pct: { sensorReading: Number(msg.loss_pct ?? 0), unit: '%' },
98+
sent: { sensorReading: Number(msg.sent ?? 0), unit: 'pkt' },
99+
received: { sensorReading: Number(msg.received ?? 0), unit: 'pkt' },
100+
},
101+
rawData: '',
102+
timestamp: ts,
103+
});
104+
return true;
105+
}
106+
107+
if (msg.type === 'radio') {
108+
dataStore.ingestMessage({
109+
msgID: DIAG_MSG_IDS.LINK_RADIO,
110+
messageName: 'Radio Stats',
111+
data: {
112+
rssi_dbm: { sensorReading: Number(msg.rssi_dbm ?? 0), unit: 'dBm' },
113+
tx_mbps: { sensorReading: Number(msg.tx_mbps ?? 0), unit: 'Mbps' },
114+
rx_mbps: { sensorReading: Number(msg.rx_mbps ?? 0), unit: 'Mbps' },
115+
ccq_pct: { sensorReading: Number(msg.ccq_pct ?? 0), unit: '%' },
116+
error: { sensorReading: msg.error ? 1 : 0, unit: '' },
117+
},
118+
rawData: typeof msg.error === 'string' ? msg.error : '',
119+
timestamp: ts,
120+
});
121+
return true;
122+
}
123+
}
124+
125+
return false;
126+
}
127+
}
128+
129+
export const telemetryHandler = new TelemetryHandler();

0 commit comments

Comments
 (0)