Connect test CAN processing to DataStore and notify on retention changes

Author: haoruizhou

.gitignore

@@ -15,6 +15,7 @@ downloads/
 eggs/
 .eggs/
 lib/
+!pecan/Frontend/pecan-live-dashboard/src/lib/
 lib64/
 parts/
 sdist/

pecan/Frontend/pecan-live-dashboard/src/lib/DataStore.ts

@@ -57,6 +57,8 @@ class DataStore {
   public static getInstance(retentionWindowMs?: number): DataStore {
     if (!DataStore.instance) {
       DataStore.instance = new DataStore(retentionWindowMs);
+    } else if (typeof retentionWindowMs === 'number') {
+      DataStore.instance.setRetentionWindow(retentionWindowMs);
     }
     return DataStore.instance;
   }
@@ -267,12 +269,19 @@ class DataStore {
    * @param windowMs - New retention window in milliseconds
    */
   public setRetentionWindow(windowMs: number): void {
+    if (windowMs === this.retentionWindowMs) {
+      return;
+    }
+
     this.retentionWindowMs = windowMs;
-    
+
     // Prune all messages with new window
     for (const msgID of this.buffer.keys()) {
       this.pruneOldSamples(msgID);
     }
+
+    // Notify subscribers since data might have been pruned
+    this.notifyAll();
   }
 
   /**

pecan/Frontend/pecan-live-dashboard/src/utils/canProcessor.ts

@@ -1,4 +1,5 @@
 import { Dbc, Can } from 'candied';
+import { dataStore } from '../lib/DataStore';
 
 // Simple type definitions for our use, align with InfluxDB3 schema for consistency
 // InfluxDB3 Schema: id -> canId, name -> messageName, signalName, sensorReading, time
@@ -103,7 +104,7 @@ const testMessages = testMessagesRaw.map(line => {
 export async function processTestMessages() {
   try {
     console.log('--- Starting CAN Message Processing ---');
-    
+
     // Fetch the DBC file content
     const dbcResponse = await fetch('/assets/dbc.dbc');
     const dbcText = await dbcResponse.text();
@@ -122,17 +123,17 @@ export async function processTestMessages() {
     for (const testMsg of testMessages) {
       // Create a CAN frame from the message ID and data
       const frame = can.createFrame(testMsg.canId, testMsg.data);
-      
+
       // Decode the frame using the DBC definitions
       const decoded = can.decode(frame);
-      
+
       if (decoded) {
         console.log(`\nTime: ${testMsg.time}, Message ID: ${testMsg.canId} (${decoded.name})`);
-        
+
         // Candied uses boundSignals property (not signals)
         if (decoded.boundSignals && decoded.boundSignals instanceof Map) {
           const signals: { [key: string]: any } = {};
-          
+
           decoded.boundSignals.forEach((signal, signalName) => {
             const parsed = parsePhysValue(signal.physValue);
             signals[signalName] = {
@@ -141,8 +142,18 @@ export async function processTestMessages() {
               // rawValue: signal.rawValue
             };
           });
-          
+
           console.log('Signals:', signals);
+
+          dataStore.ingestMessage({
+            msgID: testMsg.canId.toString(),
+            messageName: decoded.name || `CAN_${testMsg.canId}`,
+            data: signals,
+            rawData: testMsg.data
+              .map((byte) => byte.toString(16).padStart(2, '0').toUpperCase())
+              .join(' '),
+            timestamp: testMsg.time,
+          });
         } else {
           console.log('No signals found in decoded message');
         }