From d7bfaa678e0bd2e594ca4d0ded4ccd91c839e4e5 Mon Sep 17 00:00:00 2001 From: Nikita Biziaev Date: Thu, 2 Jul 2026 18:39:43 +0500 Subject: [PATCH] data/edge5cdab/modbus & logging --- app/src/handlers/demo-modbus.ts | 5 +- app/src/handlers/demo-plc.ts | 5 +- app/src/handlers/edge5-modbus-v2.ts | 5 +- app/src/handlers/edge5-video.ts | 12 ++--- app/src/handlers/edge5cdab-modbus.ts | 68 ++++++++++++++++++++++++++++ app/src/handlers/edge5i-modbus.ts | 5 +- app/src/handlers/index.ts | 30 ++++++------ app/src/helpers/log.ts | 3 ++ app/src/http/server.ts | 4 +- app/src/index.ts | 4 +- app/src/mqtt.ts | 7 +-- app/src/stream/edge-chunk-relay.ts | 8 ++-- 12 files changed, 117 insertions(+), 39 deletions(-) create mode 100644 app/src/handlers/edge5cdab-modbus.ts create mode 100644 app/src/helpers/log.ts diff --git a/app/src/handlers/demo-modbus.ts b/app/src/handlers/demo-modbus.ts index 5f1ac11..3ba719b 100644 --- a/app/src/handlers/demo-modbus.ts +++ b/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 { } } - 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()) diff --git a/app/src/handlers/demo-plc.ts b/app/src/handlers/demo-plc.ts index cc6e7c8..efd49f0 100644 --- a/app/src/handlers/demo-plc.ts +++ b/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 { 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(payload) const metric = toMetric(value, new Date().toISOString()) diff --git a/app/src/handlers/edge5-modbus-v2.ts b/app/src/handlers/edge5-modbus-v2.ts index ac4cb94..33172d1 100644 --- a/app/src/handlers/edge5-modbus-v2.ts +++ b/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 { } } - 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 { } 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()) diff --git a/app/src/handlers/edge5-video.ts b/app/src/handlers/edge5-video.ts index d68042d..24a4a83 100644 --- a/app/src/handlers/edge5-video.ts +++ b/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); } diff --git a/app/src/handlers/edge5cdab-modbus.ts b/app/src/handlers/edge5cdab-modbus.ts new file mode 100644 index 0000000..cc4bece --- /dev/null +++ b/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 { + 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) +} diff --git a/app/src/handlers/edge5i-modbus.ts b/app/src/handlers/edge5i-modbus.ts index 061b369..e85e993 100644 --- a/app/src/handlers/edge5i-modbus.ts +++ b/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 { } } - 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()) diff --git a/app/src/handlers/index.ts b/app/src/handlers/index.ts index f76f3e3..6bf8849 100644 --- a/app/src/handlers/index.ts +++ b/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); diff --git a/app/src/helpers/log.ts b/app/src/helpers/log.ts new file mode 100644 index 0000000..f81b4fc --- /dev/null +++ b/app/src/helpers/log.ts @@ -0,0 +1,3 @@ +export function log(...args: unknown[]): void { + console.log(`[${new Date().toISOString()}]`, ...args) +} diff --git a/app/src/http/server.ts b/app/src/http/server.ts index edf9d15..2c556ac 100644 --- a/app/src/http/server.ts +++ b/app/src/http/server.ts @@ -3,6 +3,8 @@ import { createServer } from 'node:http' import { dirname, join } from 'node:path' import { fileURLToPath } from 'node:url' +import { log } from '../helpers/log.js' + const httpPort = Number(process.env.HTTP_PORT) const publicDir = join(dirname(fileURLToPath(import.meta.url)), '../../public') const indexPath = join(publicDir, 'index.html') @@ -24,5 +26,5 @@ createServer(async (req, res) => { res.writeHead(404).end('Not Found') }).listen(httpPort, () => { - console.log('http.server_started', { port: httpPort }) + log('http.server_started', { port: httpPort }) }) diff --git a/app/src/index.ts b/app/src/index.ts index f6ead35..9fbd394 100644 --- a/app/src/index.ts +++ b/app/src/index.ts @@ -3,4 +3,6 @@ import './http/server.js' import './mqtt.js' import './stream/edge-chunk-relay.js' -console.log('mqtt-worker.started') +import { log } from './helpers/log.js' + +log('mqtt-worker.started') diff --git a/app/src/mqtt.ts b/app/src/mqtt.ts index 994ef29..9637eb8 100644 --- a/app/src/mqtt.ts +++ b/app/src/mqtt.ts @@ -1,5 +1,6 @@ import mqtt from 'mqtt' +import { log } from './helpers/log.js' import { getRegisteredTopics } from './registry.js' import { routeMessage } from './router.js' @@ -8,7 +9,7 @@ const mqttUrl = process.env.MQTT_URL as string export const mqttClient = mqtt.connect(mqttUrl) mqttClient.on('connect', () => { - console.log('mqtt.connected', { url: mqttUrl }) + log('mqtt.connected', { url: mqttUrl }) const topics = getRegisteredTopics() if (topics.length === 0) { @@ -21,7 +22,7 @@ mqttClient.on('connect', () => { console.error('mqtt.subscribe_error', err) return } - console.log('mqtt.subscribed', { topics }) + log('mqtt.subscribed', { topics }) }) }) @@ -34,5 +35,5 @@ mqttClient.on('error', (err) => { }) mqttClient.on('reconnect', () => { - console.log('mqtt.reconnect') + log('mqtt.reconnect') }) diff --git a/app/src/stream/edge-chunk-relay.ts b/app/src/stream/edge-chunk-relay.ts index c61ad32..4cecd24 100644 --- a/app/src/stream/edge-chunk-relay.ts +++ b/app/src/stream/edge-chunk-relay.ts @@ -1,6 +1,8 @@ import type { IncomingMessage } from 'http' import { WebSocket, WebSocketServer } from 'ws' +import { log } from '../helpers/log.js' + const wsPort = Number(process.env.WS_PORT) const wss = new WebSocketServer({ port: wsPort, perMessageDeflate: false }) @@ -8,19 +10,19 @@ const clients = new Set() wss.on('connection', (ws: WebSocket, req: IncomingMessage) => { clients.add(ws) - console.log('ws.client_connected', { + log('ws.client_connected', { clients: clients.size, ip: req.socket.remoteAddress, }) ws.on('close', () => { clients.delete(ws) - console.log('ws.client_closed', { clients: clients.size }) + log('ws.client_closed', { clients: clients.size }) }) }) wss.on('listening', () => { - console.log('ws.server_started', { port: wsPort }) + log('ws.server_started', { port: wsPort }) }) export function onEdgeChunk(chunk: Buffer): void {