app/src/handlers/edge5-modbus.ts

@@ -0,0 +1,88 @@
+import type { TopicHandler } from '../types.js'
+
+export const TOPIC = 'data/edge5/modbus/v1'
+
+const EDGE = 'edge5'
+const REGISTER_COUNT = 125
+const postUrl =
+  process.env.CLOUD_INGEST_URL ?? 'https://greact.drll.cloud/api/ingest'
+
+type Edge5ModbusMetric = {
+  edge: string
+  timestamp: number
+  tag: string
+  value: number
+}
+
+function parseValues(payload: Buffer): number[] {
+  let value: unknown
+
+  try {
+    value = JSON.parse(payload.toString('utf8')) as unknown
+  } catch {
+    throw new Error(`Expected JSON number[${REGISTER_COUNT}]`)
+  }
+
+  if (
+    Array.isArray(value) &&
+    value.length === REGISTER_COUNT &&
+    value.every((item) => typeof item === 'number' && Number.isFinite(item))
+  ) {
+    return value
+  }
+
+  throw new Error(`Expected JSON number[${REGISTER_COUNT}]`)
+}
+
+function toMetrics(values: number[], timestamp: number): Edge5ModbusMetric[] {
+  return values.map((value, index) => ({
+    edge: EDGE,
+    timestamp,
+    tag: `${EDGE}.modbus.${index + 1}`,
+    value,
+  }))
+}
+
+async function postMetrics(metrics: Edge5ModbusMetric[]): Promise<void> {
+  // console.log('edge5.modbus.post_payload', JSON.stringify(metrics))
+  // console.log('edge5.modbus.post_skipped', {
+  //   url: postUrl,
+  //   count: metrics.length,
+  // })
+
+  const response = await fetch(postUrl, {
+    method: 'POST',
+    headers: {
+      'content-type': 'application/json',
+    },
+    body: JSON.stringify(metrics),
+  })
+
+  if (!response.ok) {
+    const text = await response.text()
+    console.error('edge5.modbus.post_error', {
+      status: response.status,
+      statusText: response.statusText,
+      body: text.slice(0, 500),
+    })
+    return
+  }
+
+  console.log('edge5.modbus.posted', { count: metrics.length })
+}
+
+export const handleEdge5Modbus: TopicHandler = async (topic, payload) => {
+  console.log('edge5.modbus.received', { topic, bytes: payload.length })
+
+  let values: number[]
+
+  try {
+    values = parseValues(payload)
+  } catch (err) {
+    console.warn('edge5.modbus.invalid_payload', { topic, err })
+    return
+  }
+
+  const metrics = toMetrics(values, Date.now())
+  await postMetrics(metrics)
+}

app/src/handlers/index.ts

@@ -4,6 +4,7 @@ import { handleTopic2, TOPIC as topic2 } from './topic2.js'
 import { handleTopic3, TOPIC as topic3 } from './topic3.js'
 import { handleTopic4, TOPIC as topic4 } from './topic4.js'
 import { handleTopic5, TOPIC as topic5 } from './topic5.js'
+import { handleEdge5Modbus, TOPIC as edge5Modbus } from './edge5-modbus.js'
 import { handleEdge5Video, TOPIC as edge5Video } from './edge5-video.js'
 
 register(topic1, handleTopic1)
@@ -11,4 +12,5 @@ register(topic2, handleTopic2)
 register(topic3, handleTopic3)
 register(topic4, handleTopic4)
 register(topic5, handleTopic5)
+register(edge5Modbus, handleEdge5Modbus)
 register(edge5Video, handleEdge5Video)

docker-compose.yml

@@ -21,3 +21,4 @@ services:
       MQTT_URL: mqtt://mosquitto:1883
       HTTP_PORT: "80"
       WS_PORT: "9090"
+      CLOUD_INGEST_URL: ${CLOUD_INGEST_URL:-https://greact.drll.cloud/api/ingest}