добавлена поддержка дев окружения

app/src/handlers/demo-modbus.ts

@@ -1,9 +1,10 @@
-import { log } from '../helpers/log.js'
 import type { TopicHandler } from '../types.js'
 
+export const TOPIC = 'data/demo/modbus/v1'
+
 const EDGE = 'demo'
 const REGISTER_COUNT = 125
-const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL as string
+const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL
 
 type DemoModbusMetric = {
   edge: string
@@ -28,14 +29,14 @@ function toMetrics(values: number[], timestamp: string): DemoModbusMetric[] {
   return values.map((value, index) => ({
     edge: EDGE,
     timestamp,
-    tag: `register-${index}`,
+    tag: `${EDGE}.modbus.${index + 1}`,
     value,
   }))
 }
 
 async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
   for (const metric of metrics) {
-    const response = await fetch(DEMO_CLOUD_INGEST_URL, {
+    const response = await fetch(DEMO_CLOUD_INGEST_URL as string, {
       method: 'POST',
       headers: {
         'content-type': 'application/json',
@@ -50,11 +51,11 @@ async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
     }
   }
 
-  log(`${EDGE}.modbus.posted ${metrics.length} metrics`)
+  console.log('demo.modbus.posted', { count: metrics.length })
 }
 
 export const handleDemoModbus: TopicHandler = async (topic, payload) => {
-  log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
+  console.log('demo.modbus.received', { topic, bytes: payload.length })
 
   const values = parseValues(payload)
   const metrics = toMetrics(values, new Date().toISOString())

app/src/handlers/demo-plc.ts

@@ -1,9 +1,10 @@
-import { log } from '../helpers/log.js'
 import { parseJson } from '../helpers/parse.js'
 import type { TopicHandler } from '../types.js'
 
+export const TOPIC = 'data/demo/plc/v1'
+
 const EDGE = 'demo'
-const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL as string
+const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL
 
 type DemoPlcPayload = {
   tag: string
@@ -27,7 +28,7 @@ function toMetric(payload: DemoPlcPayload, timestamp: string): DemoPlcMetric {
 }
 
 async function postMetric(metric: DemoPlcMetric): Promise<void> {
-  const response = await fetch(DEMO_CLOUD_INGEST_URL, {
+  const response = await fetch(DEMO_CLOUD_INGEST_URL as string, {
     method: 'POST',
     headers: {
       'content-type': 'application/json',
@@ -41,11 +42,11 @@ async function postMetric(metric: DemoPlcMetric): Promise<void> {
     throw new Error(text)
   }
 
-  log('demo.plc.posted', { tag: metric.tag })
+  console.log('demo.plc.posted', { tag: metric.tag })
 }
 
 export const handleDemoPlc: TopicHandler = async (topic, payload) => {
-  log('demo.plc.received', { topic, bytes: payload.length })
+  console.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,10 +1,10 @@
 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'
 
-const REGISTER_COUNT = 125
+export const TOPIC = 'data/edge5/modbus/v2'
 
+const REGISTER_COUNT = 125
 const mapper = createMetricMapper({
   filePath: process.env.EDGE5_MODBUS_V2_MAPPING_FILE as string,
 })
@@ -39,7 +39,7 @@ async function postMetrics(metrics: MappedMetric[]): Promise<void> {
     }
   }
 
-  log('edge5.modbus.v2.posted', {
+  console.log('edge5.modbus.v2.posted', {
     total: metrics.length,
     succeeded: metrics.length - errors.length,
     failed: errors.length,
@@ -51,7 +51,7 @@ async function postMetrics(metrics: MappedMetric[]): Promise<void> {
 }
 
 export const handleEdge5ModbusV2: TopicHandler = async (topic, payload) => {
-  log('edge5.modbus.v2.received', { topic, bytes: payload.length })
+  console.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-modbus.ts

@@ -1,11 +1,12 @@
 import { postCloudIngest } from '../cloud-ingest.js'
-import { log } from '../helpers/log.js'
 import type { TopicHandler } from '../types.js'
 
-const EDGE = 'edge5i'
+export const TOPIC = 'data/edge5/modbus/v1'
-const REGISTER_COUNT = 100
 
-type DemoModbusMetric = {
+const EDGE = 'edge5'
+const REGISTER_COUNT = 125
+
+type Edge5ModbusMetric = {
   edge: string
   timestamp: string
   tag: string
@@ -24,16 +25,16 @@ function assertRegisterCount(values: number[]): void {
   }
 }
 
-function toMetrics(values: number[], timestamp: string): DemoModbusMetric[] {
+function toMetrics(values: number[], timestamp: string): Edge5ModbusMetric[] {
   return values.map((value, index) => ({
     edge: EDGE,
     timestamp,
-    tag: `register-${String(index).padStart(3, '0')}`,
+    tag: `${EDGE}.modbus.${index + 1}`,
     value,
   }))
 }
 
-async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
+async function postMetrics(metrics: Edge5ModbusMetric[]): Promise<void> {
   for (const metric of metrics) {
     const response = await postCloudIngest(metric)
 
@@ -43,11 +44,11 @@ async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
     }
   }
 
-  log(`${EDGE}.modbus.posted ${metrics.length} metrics`)
+  console.log('edge5.modbus.posted', { count: metrics.length })
 }
 
-export const handle: TopicHandler = async (topic, payload) => {
+export const handleEdge5Modbus: TopicHandler = async (topic, payload) => {
-  log(`${EDGE}.modbus.received ${topic} ${payload.length} bytes`)
+  console.log('edge5.modbus.received', { topic, bytes: payload.length })
 
   const values = parseValues(payload)
   const metrics = toMetrics(values, new Date().toISOString())

app/src/handlers/edge5-video.ts

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

app/src/handlers/edge5cdab-modbus.ts

@@ -1,69 +0,0 @@
-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())
-  log(`${EDGE}.modbus.prepared ${JSON.stringify(metrics)}`)
-  await postMetrics(metrics)
-}

app/src/handlers/index.ts

@@ -1,22 +1,15 @@
-import { register } from '../registry.js';
+import { register } from '../registry.js'
+import { handleDemoModbus, TOPIC as demoModbus } from './demo-modbus.js'
+import { handleDemoPlc, TOPIC as demoPlc } from './demo-plc.js'
+import { handleEdge5Modbus, TOPIC as edge5Modbus } from './edge5-modbus.js'
+import {
+  handleEdge5ModbusV2,
+  TOPIC as edge5ModbusV2,
+} from './edge5-modbus-v2.js'
+import { handleEdge5Video, TOPIC as edge5Video } from './edge5-video.js'
 
-import { handleDemoPlc } from './demo-plc.js';
+register(demoModbus, handleDemoModbus)
-import { handleDemoModbus } from './demo-modbus.js';
+register(demoPlc, handleDemoPlc)
-import { handleEdge5ModbusV2 } from './edge5-modbus-v2.js';
+register(edge5Modbus, handleEdge5Modbus)
-import { handleEdge5Video } from './edge5-video.js';
+register(edge5ModbusV2, handleEdge5ModbusV2)
-
+register(edge5Video, handleEdge5Video)
-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';
-
-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);

app/src/helpers/log.ts

@@ -1,3 +0,0 @@
-export function log(...args: unknown[]): void {
-  console.log(`[${new Date().toISOString()}]`, ...args)
-}

app/src/http/server.ts

@@ -3,8 +3,6 @@ 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')
@@ -26,5 +24,5 @@ createServer(async (req, res) => {
 
   res.writeHead(404).end('Not Found')
 }).listen(httpPort, () => {
-  log('http.server_started', { port: httpPort })
+  console.log('http.server_started', { port: httpPort })
 })

app/src/index.ts

@@ -3,6 +3,4 @@ import './http/server.js'
 import './mqtt.js'
 import './stream/edge-chunk-relay.js'
 
-import { log } from './helpers/log.js'
+console.log('mqtt-worker.started')
-
-log('mqtt-worker.started')

app/src/mqtt.ts

@@ -1,6 +1,5 @@
 import mqtt from 'mqtt'
 
-import { log } from './helpers/log.js'
 import { getRegisteredTopics } from './registry.js'
 import { routeMessage } from './router.js'
 
@@ -9,7 +8,7 @@ const mqttUrl = process.env.MQTT_URL as string
 export const mqttClient = mqtt.connect(mqttUrl)
 
 mqttClient.on('connect', () => {
-  log('mqtt.connected', { url: mqttUrl })
+  console.log('mqtt.connected', { url: mqttUrl })
 
   const topics = getRegisteredTopics()
   if (topics.length === 0) {
@@ -22,7 +21,7 @@ mqttClient.on('connect', () => {
       console.error('mqtt.subscribe_error', err)
       return
     }
-    log('mqtt.subscribed', { topics })
+    console.log('mqtt.subscribed', { topics })
   })
 })
 
@@ -35,5 +34,5 @@ mqttClient.on('error', (err) => {
 })
 
 mqttClient.on('reconnect', () => {
-  log('mqtt.reconnect')
+  console.log('mqtt.reconnect')
 })

app/src/stream/edge-chunk-relay.ts

@@ -1,8 +1,6 @@
 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 })
@@ -10,19 +8,19 @@ const clients = new Set<WebSocket>()
 
 wss.on('connection', (ws: WebSocket, req: IncomingMessage) => {
   clients.add(ws)
-  log('ws.client_connected', {
+  console.log('ws.client_connected', {
     clients: clients.size,
     ip: req.socket.remoteAddress,
   })
 
   ws.on('close', () => {
     clients.delete(ws)
-    log('ws.client_closed', { clients: clients.size })
+    console.log('ws.client_closed', { clients: clients.size })
   })
 })
 
 wss.on('listening', () => {
-  log('ws.server_started', { port: wsPort })
+  console.log('ws.server_started', { port: wsPort })
 })
 
 export function onEdgeChunk(chunk: Buffer): void {

docker-compose.yml

@@ -13,7 +13,7 @@ services:
     depends_on:
       - mosquitto
     ports:
-      # - "80:80"
+      - "80:80"
       - "9090:9090"
     environment:
       MQTT_URL: mqtt://mosquitto:1883

package-lock.json

@@ -1,6 +0,0 @@
-{
-  "name": "mqtt-ingest",
-  "lockfileVersion": 3,
-  "requires": true,
-  "packages": {}
-}