data/edge5/modbus/v3

app/Dockerfile

@@ -10,9 +10,7 @@ COPY src ./src
 
 RUN npm run build && npm prune --omit=dev
 
-# app/public/index.html has content for video preview
+COPY assets ./assets
-# TODO move all static files (html, mappers) to a separate folder "assets"
-COPY public ./public 
 
 ENV NODE_ENV=production
 CMD ["node", "dist/index.js"]

app/VIDEO.md

@@ -0,0 +1,96 @@
+Финальные требования и настройки у нас сейчас такие.
+
+## Финальный пресет чанкования ffmpeg (RTSP → MPEG‑TS → stdout)
+
+Мы остановились на сценарии, где ffmpeg:
+
+- тянет RTSP **по TCP** с камер,  
+- минимизирует внутреннюю буферизацию и время старта,  
+- не перекодирует видео (копирует H.264 как есть); если возникнут проблемы с нагрузкой на сеть, можно поискать варианты снизить; но на данном этапе важнее "накормить" видеобуфер воспроизведения хоть чем-нибудь;
+- выдаёт результат в **stdout (`pipe:1`)**, откуда его забирает mqtt транспорт.
+
+Финальный набор аргументов:
+
+```js
+[
+  '-hide_banner',
+  '-fflags', 'nobuffer',          // не буферировать при probe stream info
+  '-flags', 'low_delay',          // включить low-delay на уровне декодера
+  '-analyzeduration', '100000',   // ограничить анализ потока 100 ms (уменьшить стартовую задержку)
+  '-probesize', '32768',          // зондировать только 32 KB (быстрее старт, меньше буфера)
+  '-flush_packets', '1',          // сразу сбрасывать пакеты, не копить их внутри muxer'а
+
+  '-rtsp_transport', 'tcp',       // тянуть RTSP по TCP (надёжнее на плохой сети)
+  '-i', 'rtsp://admin:admin@192.168.0.11:554/live/sub', // sub-поток камеры
+
+  '-an',                          // отключить аудио — нужен только видеофрейм
+  '-c:v', 'copy',                 // не перекодировать видео, просто копировать кадры
+  '-f', 'mpegts',                 // контейнер MPEG-TS для дальнейшего трансмультиплексирования
+  '-mpegts_flags', 'resend_headers', // периодически повторять заголовки для надёжного декода
+  'pipe:1'                        // выводить поток в stdout (дальше забирает твой транспорт)
+]
+```
+
+Ключевые моменты:
+
+- `-c:v copy` сохраняет оригинальный кадр и GOP камеры — минимум задержки и вычислительной нагрузки.  
+- `-fflags nobuffer + -flags low_delay + -flush_packets 1` минимизируют буферизацию и увеличивают «прозрачность» для твоего транспорта.  
+- `-rtsp_transport tcp` это про **вход** (камера → ffmpeg), а `pipe:1` — про **выход** (ffmpeg → mqtt).
+
+***
+
+## Финальный пресет воспроизведения (mpegts.js + HTML5 video)
+
+На клиенте мы на данном этапе определяемся работать как с «live‑видео с буфером», а не как с чистым real‑time:
+
+- формат источника: **MPEG‑TS/H.264** по WebSocket;  
+- плеер: `mpegts.js`, режим `type: 'mpegts', isLive: true`;  
+- буфер: небольшой stash + несколько секунд latency для сглаживания провалов.
+
+Финальные параметры инициализации:
+
+```js
+const wsUrl = 'wss://...';
+
+const player = mpegts.createPlayer(
+  {
+    type: 'mpegts',         // MPEG-TS поток по WS
+    isLive: true,
+    url: wsUrl,
+    hasAudio: false,        // видео без аудио
+  },
+  {
+    enableWorker: true,
+
+    // Включаем stash, чтобы сглаживать редкий / рваный поток
+    enableStashBuffer: true,
+    stashInitialSize: 512,              // немного увеличенный stash для накопления
+
+    autoCleanupSourceBuffer: true,
+    autoCleanupMaxBackwardDuration: 10, // держим минимум истории
+    autoCleanupMinBackwardDuration: 5,
+
+    // Разрешаем latency-chasing с «комфортной» задержкой
+    liveBufferLatencyChasing: true,
+    liveBufferLatencyMaxLatency: 5.0,   // целевая задержка до ~5 секунд
+    liveBufferLatencyMinRemain: 2.0,    // оставляем ~2 секунды буфера после сброса
+
+    liveSync: false,                    // не пытаемся мягко догонять — приоритет стабильности
+    lazyLoad: false,
+    deferLoadAfterSourceOpen: false,
+  }
+);
+
+player.attachMediaElement(document.getElementById('v'));
+player.load();
+player.play();
+```
+
+Смысл этих настроек:
+
+- **Stash‑buffer включён**: mpegts.js может аккумулировать некоторое количество данных, прежде чем кормить MSE (браузеру), что хорошо при «дырявом» канале.  
+- **Latency‑chasing с MaxLatency ≈ 5 s**: браузер допускает несколько секунд отставания, но не позволяет задержке расти бесконечно; при накоплении лишнего буфера он «подпрыгивает» ближе к live‑позиции, сохраняя `MinRemain` ~2 секунды запаса.  
+- `liveSync: false` — мы не гонимся за минимально возможной задержкой любой ценой, а принимаем небольшой «запас» как разумный компромисс.  
+
+При улучшении канала или выполнениии работ по снижению нагрузки на сеть (уменьшения битрейта, fps) 
+нужно грамотно управлять буфером на стороне клиента (stash + latency).

app/assets/index.html

@@ -0,0 +1,45 @@
+<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="utf-8" />
+  <title>edge5 video relay</title>
+  <script src="https://cdn.jsdelivr.net/npm/mpegts.js/dist/mpegts.js"></script>
+  <style>
+    body { margin: 0; background: #000; }
+    #v { display: block; width: 100vw; height: 100vh; object-fit: contain; }
+  </style>
+</head>
+<body>
+  <video id="v" controls autoplay muted></video>
+  <script>
+    const cameraId = new URLSearchParams(location.search).get('camera') ?? 'v1';
+    const wsUrl = `${location.protocol === 'https:' ? 'wss' : 'ws'}://${location.host}/ws/${cameraId}`;
+    const player = mpegts.createPlayer(
+      { type: 'mpegts', isLive: true, url: wsUrl, hasAudio: false },
+      {
+        enableWorker: true,
+
+        // Включаем stash, чтобы сглаживать редкий / рваный поток
+        enableStashBuffer: true,
+        stashInitialSize: 512,              // немного увеличенный stash для накопления
+
+        autoCleanupSourceBuffer: true,
+        autoCleanupMaxBackwardDuration: 10, // держим минимум истории
+        autoCleanupMinBackwardDuration: 5,
+
+        // Разрешаем latency-chasing с «комфортной» задержкой
+        liveBufferLatencyChasing: true,
+        liveBufferLatencyMaxLatency: 5.0,   // целевая задержка до ~5 секунд
+        liveBufferLatencyMinRemain: 2.0,    // оставляем ~2 секунды буфера после сброса
+
+        liveSync: false,                    // не пытаемся мягко догонять — приоритет стабильности
+        lazyLoad: false,
+        deferLoadAfterSourceOpen: false,
+      }
+    );
+    player.attachMediaElement(document.getElementById('v'))
+    player.load()
+    player.play()
+  </script>
+</body>
+</html>

app/public/index.html

@@ -1,45 +0,0 @@
-<!DOCTYPE html>
-<html>
-<head>
-  <meta charset="utf-8" />
-  <title>edge5 video relay</title>
-  <script src="https://cdn.jsdelivr.net/npm/mpegts.js/dist/mpegts.js"></script>
-  <style>
-    body { margin: 0; background: #000; }
-    #v { display: block; width: 100vw; height: 100vh; object-fit: contain; }
-  </style>
-</head>
-<body>
-  <video id="v" controls autoplay muted></video>
-  <script>
-    const cameraId = new URLSearchParams(location.search).get('camera') ?? 'v1';
-    const wsUrl = `${location.protocol === 'https:' ? 'wss' : 'ws'}://${location.host}/ws/${cameraId}`;
-    const player = mpegts.createPlayer(
-      { type: 'mpegts', isLive: true, url: wsUrl, hasAudio: false },
-      {
-        enableWorker: true,             // transmuxing в отдельном треде
-        enableStashBuffer: true,        // включить stash
-        stashInitialSize: 384,
-
-        autoCleanupSourceBuffer: true,  // не копить старые буферы в MSE
-        autoCleanupMaxBackwardDuration: 10,
-        autoCleanupMinBackwardDuration: 5,
-
-        // liveSync — плавное догоняние через playbackRate, а не жёсткий seek
-        liveSync: true,
-        liveSyncTargetLatency: 4.0,    // target после выравнивания ← дать буферу вырасти
-        liveSyncMaxLatency: 10.0,      // при плохом интернете даём 10с ← не трогать буфер до 10s
-        liveSyncPlaybackRate: 1.05,    // ускорение не более 1.2x ← очень мягкое догоняние
-
-        liveBufferLatencyChasing: false,  // включить stash ← убрать конфликтующий механизм
-
-        lazyLoad: false,
-        deferLoadAfterSourceOpen: false,
-      }
-    );
-    player.attachMediaElement(document.getElementById('v'))
-    player.load()
-    player.play()
-  </script>
-</body>
-</html>

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

@@ -1,3 +1,6 @@
+import { dirname, join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+
 import { postCloudIngest } from '../cloud-ingest.js'
 import { log } from '../helpers/log.js'
 import { createMetricMapper, type MappedMetric } from '../mapping.js'
@@ -5,8 +8,13 @@ import type { TopicHandler } from '../types.js'
 
 const REGISTER_COUNT = 125
 
+const defaultMappingFile = join(
+  dirname(fileURLToPath(import.meta.url)),
+  '../../assets/edge5-modbus.json',
+)
+
 const mapper = createMetricMapper({
-  filePath: process.env.EDGE5_MODBUS_V2_MAPPING_FILE as string,
+  filePath: process.env.EDGE5_MODBUS_V2_MAPPING_FILE ?? defaultMappingFile,
 })
 
 function parseValues(payload: Buffer): number[] {

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

@@ -0,0 +1,121 @@
+import { postCloudIngest } from '../cloud-ingest.js'
+import { log } from '../helpers/log.js'
+import type { TopicHandler } from '../types.js'
+
+const EDGE = 'edge5-v3'
+const REGISTER_COUNT = 24
+
+// Фиксируем интерпретацию
+const TAGS = [
+  'edge5-v3-wk',
+  'edge5-v3-pin',
+  'edge5-v3-hk',
+  'edge5-v3-mk',
+  'edge5-v3-mrot',
+  'edge5-v3-h',
+  'edge5-v3-vw',
+  'edge5-v3-pdk',
+  'edge5-v3-h2s',
+  'edge5-v3-t',
+  'edge5-v3-nrot',
+  'edge5-v3-vsp',
+] as const
+
+// из них выводим на дашборд семь виджетов
+const CLOUD_INGEST_TAGS = new Set<string>([
+  'edge5-v3-wk',
+  'edge5-v3-hk',
+  'edge5-v3-vw',
+  'edge5-v3-pdk',
+  'edge5-v3-h2s',
+  'edge5-v3-nrot',
+  'edge5-v3-vsp',
+])
+
+type Edge5ModbusV3Metric = {
+  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): Edge5ModbusV3Metric[] {
+  const metrics: Edge5ModbusV3Metric[] = []
+
+  for (let i = 0; i < values.length; i += 2) {
+    const tag = TAGS[i / 2]
+    const value = toFloatCDAB(values[i], values[i + 1])
+    if (isNaN(value)) {
+      if (CLOUD_INGEST_TAGS.has(tag)) {
+        log(
+          `edge5.modbus.v3: не удалось собрать число для ${tag}, ` +
+            `исходные values[${i}]=${values[i]}, values[${i + 1}]=${values[i + 1]}, пропуск`,
+        )
+      }
+      continue
+    }
+
+    metrics.push({
+      edge: EDGE,
+      timestamp,
+      tag,
+      value,
+    })
+  }
+
+  return metrics
+}
+
+async function postMetrics(metrics: Edge5ModbusV3Metric[]): Promise<void> {
+  const toPost = metrics.filter((metric) => CLOUD_INGEST_TAGS.has(metric.tag))
+  const errors: Error[] = []
+
+  for (const metric of toPost) {
+    try {
+      const response = await postCloudIngest(metric)
+
+      if (!response.ok) {
+        const text = await response.text()
+        throw new Error(text || `${response.status} ${response.statusText}`)
+      }
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error)
+      errors.push(new Error(`${metric.tag}: ${message}`))
+      console.error('edge5.modbus.v3.metric.failed', metric.tag, message)
+    }
+  }
+
+  log('edge5.modbus.v3.posted', { total: toPost.length });
+
+  if (errors.length) {
+    throw new Error(`Failed to post ${errors.length}/${toPost.length} edge5 modbus v3 metrics`)
+  }
+}
+
+export const handleEdge5ModbusV3: TopicHandler = async (topic, payload) => {
+  const timestamp = new Date().toISOString();
+  log('edge5.modbus.v3.received', { topic, bytes: payload.length, timestamp })
+
+  const values = parseValues(payload);
+  const metrics = toMetrics(values, timestamp);
+  await postMetrics(metrics);
+}

app/src/handlers/index.ts

@@ -3,6 +3,7 @@ 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 { handleEdge5ModbusV3 } from './edge5-modbus-v3.js';
 import { handleEdge5Video } from './edge5-video.js';
 import { handleEdge5VideoV2 } from './edge5-video-v2.js';
 
@@ -22,3 +23,4 @@ register(EDGE5_VIDEO_TOPIC, handleEdge5Video);
 register('data/edge5i/modbus', interpretation);
 register('data/edge5cdab/modbus', cdab);
 register('data/edge5/video/v2/+', handleEdge5VideoV2);
+register('data/edge5/modbus/v3', handleEdge5ModbusV3);

app/src/http/server.ts

@@ -6,8 +6,8 @@ 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 assetsDir = join(dirname(fileURLToPath(import.meta.url)), '../../assets')
-const indexPath = join(publicDir, 'index.html')
+const indexPath = join(assetsDir, 'index.html')
 
 createServer(async (req, res) => {
   const path = req.url?.split('?')[0]

app/src/index.ts

@@ -2,7 +2,6 @@ import './handlers/index.js'
 import './http/server.js'
 import './mqtt.js'
 import './stream/edge-chunk-relay.js'
-import './stream/ws-pipe-relay.js'
 
 import { log } from './helpers/log.js'
 

app/src/mapping.ts

@@ -19,7 +19,7 @@ export type MetricMapper = {
 
 type MetricMapperOptions = {
   // filePath contract:
-  // - built-in mappings are passed as URL via new URL('../mappings/*.json', import.meta.url);
+  // - built-in mappings live in app/assets/*.json (resolved relative to dist/ at runtime);
   // - env overrides are expected as absolute paths inside the runtime container.
   filePath: string | URL
 }

app/src/mqtt.ts

@@ -6,7 +6,12 @@ import { routeMessage } from './router.js'
 
 const mqttUrl = process.env.MQTT_URL as string
 
-export const mqttClient = mqtt.connect(mqttUrl)
+export const mqttClient = mqtt.connect(mqttUrl, {
+  clean: true,
+  queueQoSZero: false,
+  reconnectPeriod: 1000,
+  connectTimeout: 10000,
+});
 
 mqttClient.on('connect', () => {
   log('mqtt.connected', { url: mqttUrl })

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

@@ -21,6 +21,7 @@ function getClients(cameraId: string): Set<WebSocket> {
 
 wss.on('connection', (ws: WebSocket, req: IncomingMessage) => {
   log(`ws.connection ${req.url}`);
+  req.socket.setNoDelay(true)
   const cameraId = req.url ? lastTopicSegment(req.url) : 'v1';
 
   const clients = getClients(cameraId);

app/src/stream/ws-pipe-relay.ts

@@ -1,94 +0,0 @@
-import type { IncomingMessage } from 'http'
-import { WebSocket, WebSocketServer } from 'ws'
-
-import { log } from '../helpers/log.js'
-import { lastTopicSegment } from '../helpers/topic-match.js'
-
-const wsPipePort = Number(process.env.WS_PIPE_PORT)
-const LOG_INTERVAL_MS = 10_000
-
-const wss = new WebSocketServer({ port: wsPipePort, perMessageDeflate: false })
-
-const producersByChannel = new Map<string, WebSocket>()
-const viewersByChannel = new Map<string, Set<WebSocket>>()
-const lastLogByChannel = new Map<string, number>()
-
-function getViewers(channel: string): Set<WebSocket> {
-  let viewers = viewersByChannel.get(channel)
-  if (!viewers) {
-    viewers = new Set()
-    viewersByChannel.set(channel, viewers)
-  }
-  return viewers
-}
-
-function broadcast(channel: string, chunk: Buffer): void {
-  const clients = viewersByChannel.get(channel)
-  if (!clients) return
-  for (const client of clients) {
-    if (client.readyState === WebSocket.OPEN) {
-      client.send(chunk)
-    }
-  }
-}
-
-function logThroughput(channel: string, bytes: number): void {
-  const now = Date.now()
-  const lastLog = lastLogByChannel.get(channel) ?? 0
-  if (now - lastLog >= LOG_INTERVAL_MS) {
-    log(`pipe.in [${channel}] ↑ ${bytes}B`)
-    lastLogByChannel.set(channel, now)
-  }
-}
-
-function handleProducer(ws: WebSocket, channel: string): void {
-  const existing = producersByChannel.get(channel)
-  if (existing?.readyState === WebSocket.OPEN) {
-    existing.close(1000, 'replaced')
-  }
-  producersByChannel.set(channel, ws)
-
-  ws.on('message', (data, isBinary) => {
-    if (!isBinary) return
-    const chunk = Buffer.isBuffer(data) ? data : Buffer.from(data as ArrayBuffer)
-    logThroughput(channel, chunk.length)
-    broadcast(channel, chunk)
-  })
-
-  ws.on('close', () => {
-    if (producersByChannel.get(channel) === ws) {
-      producersByChannel.delete(channel)
-    }
-  })
-}
-
-function handleViewer(ws: WebSocket, channel: string, path: string): void {
-  log(`pipe.out.connection ${path}`)
-  const viewers = getViewers(channel)
-  viewers.add(ws)
-
-  ws.on('close', () => {
-    viewers.delete(ws)
-    if (viewers.size === 0) viewersByChannel.delete(channel)
-  })
-}
-
-wss.on('connection', (ws: WebSocket, req: IncomingMessage) => {
-  const path = req.url?.split('?')[0] ?? ''
-
-  if (path.startsWith('/in/')) {
-    handleProducer(ws, lastTopicSegment(path))
-    return
-  }
-
-  if (path.startsWith('/out/')) {
-    handleViewer(ws, lastTopicSegment(path), path)
-    return
-  }
-
-  ws.close(1008, 'unknown path')
-})
-
-wss.on('listening', () => {
-  log('pipe.server_started', { port: wsPipePort })
-})

docker-compose.yml

@@ -17,16 +17,14 @@ services:
     expose:
       - 80 
       - 9090 
-      - 9091
     environment:
       MQTT_URL: mqtt://mosquitto:1883
-      HTTP_PORT: 80 # http port for video previewiong page
+      HTTP_PORT: 80 # http port for video previewing page
       WS_PORT: 9090 # WebSocket port for video streaming
-      WS_PIPE_PORT: 9091 # WebSocket port for incoming and outgoing binary data
       CLOUD_INGEST_URL: ${CLOUD_INGEST_URL}
       CLOUD_INGEST_API_KEY: ${CLOUD_INGEST_API_KEY}
       DEMO_CLOUD_INGEST_URL: ${DEMO_CLOUD_INGEST_URL}
-      EDGE5_MODBUS_V2_MAPPING_FILE: ${EDGE5_MODBUS_V2_MAPPING_FILE:-/app/src/mappings/edge5-modbus.json}
+      EDGE5_MODBUS_V2_MAPPING_FILE: ${EDGE5_MODBUS_V2_MAPPING_FILE:-/app/assets/edge5-modbus.json}
     networks:
       - default
       - proxy

mosquitto/config/mosquitto.conf

@@ -1,6 +1,8 @@
 listener 1883
 allow_anonymous true
 
+set_tcp_nodelay true          # выключает Nagle на всех сокетах брокера
+
 persistence true
 persistence_location /mosquitto/data/