data/edge5cdab/modbus & logging

app/src/handlers/demo-modbus.ts

@@ -1,4 +1,5 @@
 import { postCloudIngest } from '../cloud-ingest.js'
+import { log } from '../helpers/log.js'
 import type { TopicHandler } from '../types.js'
 
 const EDGE = 'demo'
@@ -42,11 +43,11 @@ async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
     }
   }
 
-  console.log(`${EDGE}.modbus.posted ${metrics.length} metrics`)
+  log(`${EDGE}.modbus.posted ${metrics.length} metrics`)
 }
 
 export const handleDemoModbus: TopicHandler = async (topic, payload) => {
-  console.log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
+  log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
 
   const values = parseValues(payload)
   const metrics = toMetrics(values, new Date().toISOString())

app/src/handlers/demo-plc.ts

@@ -1,4 +1,5 @@
 import { postCloudIngest } from '../cloud-ingest.js'
+import { log } from '../helpers/log.js'
 import { parseJson } from '../helpers/parse.js'
 import type { TopicHandler } from '../types.js'
 
@@ -33,11 +34,11 @@ async function postMetric(metric: DemoPlcMetric): Promise<void> {
     throw new Error(text)
   }
 
-  console.log('demo.plc.posted', { tag: metric.tag })
+  log('demo.plc.posted', { tag: metric.tag })
 }
 
 export const handleDemoPlc: TopicHandler = async (topic, payload) => {
-  console.log('demo.plc.received', { topic, bytes: payload.length })
+  log('demo.plc.received', { topic, bytes: payload.length })
 
   const value = parseJson<DemoPlcPayload>(payload)
   const metric = toMetric(value, new Date().toISOString())

app/src/handlers/edge5-modbus-v2.ts

@@ -1,4 +1,5 @@
 import { postCloudIngest } from '../cloud-ingest.js'
+import { log } from '../helpers/log.js'
 import { createMetricMapper, type MappedMetric } from '../mapping.js'
 import type { TopicHandler } from '../types.js'
 
@@ -38,7 +39,7 @@ async function postMetrics(metrics: MappedMetric[]): Promise<void> {
     }
   }
 
-  console.log('edge5.modbus.v2.posted', {
+  log('edge5.modbus.v2.posted', {
     total: metrics.length,
     succeeded: metrics.length - errors.length,
     failed: errors.length,
@@ -50,7 +51,7 @@ async function postMetrics(metrics: MappedMetric[]): Promise<void> {
 }
 
 export const handleEdge5ModbusV2: TopicHandler = async (topic, payload) => {
-  console.log('edge5.modbus.v2.received', { topic, bytes: payload.length })
+  log('edge5.modbus.v2.received', { topic, bytes: payload.length })
 
   const values = parseValues(payload)
   const metrics = mapper.mapValues(values, new Date().toISOString())

app/src/handlers/edge5-video.ts

@@ -1,14 +1,8 @@
+import { log } from '../helpers/log.js'
 import { onEdgeChunk } from '../stream/edge-chunk-relay.js'
 import type { TopicHandler } from '../types.js'
 
-const BEACON_MS = 3000
-let lastBeacon = 0
-
 export const handleEdge5Video: TopicHandler = async (_topic, payload) => {
-  const now = Date.now()
-  if (now - lastBeacon >= BEACON_MS) {
-    console.log(`edge5.video ↑ ${payload.length}B`)
-    lastBeacon = now
-  }
-  onEdgeChunk(payload)
+  log(`edge5.video ↑ ${payload.length}B`);
+  onEdgeChunk(payload);
 }

app/src/handlers/edge5cdab-modbus.ts

@@ -0,0 +1,68 @@
+import { postCloudIngest } from '../cloud-ingest.js'
+import { log } from '../helpers/log.js'
+import type { TopicHandler } from '../types.js'
+
+const EDGE = 'edge5cdab'
+const REGISTER_COUNT = 100
+
+type Edge5iiModbusMetric = {
+  edge: string
+  timestamp: string
+  tag: string
+  value: number
+}
+
+function toFloatCDAB(reg0: number, reg1: number): number {
+  const buf = Buffer.allocUnsafe(4)
+  buf.writeUInt16BE(reg1, 0)
+  buf.writeUInt16BE(reg0, 2)
+  return buf.readFloatBE(0)
+}
+
+function parseValues(payload: Buffer): number[] {
+  const values = JSON.parse(payload.toString('utf8')) as number[]
+  assertRegisterCount(values)
+  return values
+}
+
+function assertRegisterCount(values: number[]): void {
+  if (values.length !== REGISTER_COUNT) {
+    throw new Error(`Expected ${REGISTER_COUNT} modbus values, got ${values.length}`)
+  }
+}
+
+function toMetrics(values: number[], timestamp: string): Edge5iiModbusMetric[] {
+  const metrics: Edge5iiModbusMetric[] = []
+
+  for (let i = 0; i < values.length; i += 2) {
+    metrics.push({
+      edge: EDGE,
+      timestamp,
+      tag: `register-${String(i).padStart(3, '0')}`,
+      value: toFloatCDAB(values[i], values[i + 1]),
+    })
+  }
+
+  return metrics
+}
+
+async function postMetrics(metrics: Edge5iiModbusMetric[]): Promise<void> {
+  for (const metric of metrics) {
+    const response = await postCloudIngest(metric)
+
+    if (!response.ok) {
+      const text = await response.text()
+      throw new Error(text)
+    }
+  }
+
+  log(`${EDGE}.modbus.posted ${metrics.length} metrics`)
+}
+
+export const handle: TopicHandler = async (topic, payload) => {
+  log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
+
+  const values = parseValues(payload)
+  const metrics = toMetrics(values, new Date().toISOString())
+  await postMetrics(metrics)
+}

app/src/handlers/edge5i-modbus.ts

@@ -1,4 +1,5 @@
 import { postCloudIngest } from '../cloud-ingest.js'
+import { log } from '../helpers/log.js'
 import type { TopicHandler } from '../types.js'
 
 const EDGE = 'edge5i'
@@ -42,11 +43,11 @@ async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
     }
   }
 
-  console.log(`${EDGE}.modbus.posted ${metrics.length} metrics`)
+  log(`${EDGE}.modbus.posted ${metrics.length} metrics`)
 }
 
 export const handle: TopicHandler = async (topic, payload) => {
-  console.log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
+  log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
 
   const values = parseValues(payload)
   const metrics = toMetrics(values, new Date().toISOString())

app/src/handlers/index.ts

@@ -1,20 +1,22 @@
-import { register } from '../registry.js'
+import { register } from '../registry.js';
 
-import { handleDemoPlc } from './demo-plc.js'
-import { handleDemoModbus } from './demo-modbus.js'
-import { handleEdge5ModbusV2 } from './edge5-modbus-v2.js'
-import { handleEdge5Video } from './edge5-video.js'
+import { handleDemoPlc } from './demo-plc.js';
+import { handleDemoModbus } from './demo-modbus.js';
+import { handleEdge5ModbusV2 } from './edge5-modbus-v2.js';
+import { handleEdge5Video } from './edge5-video.js';
 
-import { handle as interpretation } from './edge5i-modbus.js'
+import { handle as interpretation } from './edge5i-modbus.js';
+import { handle as cdab} from './edge5cdab-modbus.js';
 
-const DEMO_PLC_TOPIC = 'data/demo/plc/v1'
-const DEMO_MODBUS_TOPIC = 'data/demo/modbus/v1'
-const EDGE5_MODBUS_TOPIC = 'data/edge5/modbus/v2'
-const EDGE5_VIDEO_TOPIC = 'data/edge5/video/v1'
+const DEMO_PLC_TOPIC = 'data/demo/plc/v1';
+const DEMO_MODBUS_TOPIC = 'data/demo/modbus/v1';
+const EDGE5_MODBUS_TOPIC = 'data/edge5/modbus/v2';
+const EDGE5_VIDEO_TOPIC = 'data/edge5/video/v1';
 
-register(DEMO_PLC_TOPIC, handleDemoPlc)
-register(DEMO_MODBUS_TOPIC, handleDemoModbus)
-register(EDGE5_MODBUS_TOPIC, handleEdge5ModbusV2)
-register(EDGE5_VIDEO_TOPIC, handleEdge5Video)
+register(DEMO_PLC_TOPIC, handleDemoPlc);
+register(DEMO_MODBUS_TOPIC, handleDemoModbus);
+register(EDGE5_MODBUS_TOPIC, handleEdge5ModbusV2);
+register(EDGE5_VIDEO_TOPIC, handleEdge5Video);
 
 register('data/edge5i/modbus', interpretation);
+register('data/edge5cdab/modbus', cdab);