diff --git a/app/src/handlers/edge5-modbus.ts b/app/src/handlers/edge5-modbus.ts new file mode 100644 index 0000000..368a1b0 --- /dev/null +++ b/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 { + // 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) +} diff --git a/app/src/handlers/index.ts b/app/src/handlers/index.ts index 22d1f2d..573acc8 100644 --- a/app/src/handlers/index.ts +++ b/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) diff --git a/docker-compose.yml b/docker-compose.yml index 6fa96f4..f5634c5 100644 --- a/docker-compose.yml +++ b/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}