feat(vucm): add integration with NI-cDAQ

Author: anataty

examples/vucm/ni-daq/Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.10-alpine3.16
+
+WORKDIR /app
+
+RUN apk add build-base
+
+RUN python -m venv .venv
+COPY requirements.txt requirements.txt
+RUN .venv/bin/pip install -r requirements.txt
+
+COPY script.py script.py
+
+CMD [".venv/bin/python", "script.py"]

examples/vucm/ni-daq/docker_run.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+set -euo pipefail
+IFS=$'\n\t'
+
+SCRIPT_DIR="$(realpath "$(dirname "$0")")"
+IMAGE_TAG="${IMAGE_TAG:-"enapter-vucm-examples/$(basename "$SCRIPT_DIR"):latest"}"
+
+docker build --tag "$IMAGE_TAG" "$SCRIPT_DIR"
+
+docker run --rm -it \
+	--name "ni-daq" \
+	--network host \
+	-e ENAPTER_LOG_LEVEL="${ENAPTER_LOG_LEVEL:-info}" \
+	-e ENAPTER_VUCM_BLOB="$ENAPTER_VUCM_BLOB" \
+	-e LISTEN_TCP_PORT="$LISTEN_TCP_PORT" \
+	"$IMAGE_TAG"

examples/vucm/ni-daq/manifest.yml

@@ -0,0 +1,101 @@
+blueprint_spec: "device/1.0"
+
+display_name: ATS stack
+
+communication_module:
+  product: ENP-VIRTUAL
+
+properties:
+  model:
+    display_name: Model
+    type: string
+
+alerts:
+  parse_error:
+    display_name: Data processing failed
+    severity: error							
+telemetry:
+  status:
+    display_name: Status
+    type: string
+    enum:
+      - ok
+      - error
+      - no_data
+  T1:
+    display_name: T1
+    type: float
+  T2:
+    display_name: T2
+    type: float
+  T3:
+    display_name: T2
+    type: float
+  Current:
+    display_name: Current
+    type: float
+  PSU:
+    display_name: Current
+    type: float
+  P1:
+    display_name: P1
+    type: float
+  P2:
+    display_name: P2
+    type: float
+  P3:
+    display_name: P3
+    type: float
+  Flow:
+    display_name: Flow
+    type: float
+  Conductivity:
+    display_name: Conductivity
+    type: float
+  MFMH2:
+    display_name: MFMH2
+    type: float
+  Theoretical_h2:
+    display_name: MFMH2
+    type: float
+  MCM02:
+    display_name: MCM02
+    type: float
+  Refilling:
+    display_name: Refilling
+    type: float
+  PC:
+    display_name: PC
+    type: float
+  C1:
+    display_name: Cell 1
+    type: float
+  C2:
+    display_name: Cell 2
+    type: float
+  C3:
+    display_name: Cell 3
+    type: float
+  C4:
+    display_name: Cell 4
+    type: float
+  C5:
+    display_name: Cell 5
+    type: float
+  C6:
+    display_name: Cell 6
+    type: float
+  C7:
+    display_name: Cell 7
+    type: float
+  C8:
+    display_name: Cell 8
+    type: float
+  C9:
+    display_name: Cell 9
+    type: float
+  C10:
+    display_name: Cell 10
+    type: float
+
+commands: {}

examples/vucm/ni-daq/requirements.txt

@@ -0,0 +1 @@
+enapter==0.9.2

examples/vucm/ni-daq/script.py

@@ -0,0 +1,95 @@
+import asyncio
+import functools
+import json
+import os
+import socket
+import time
+from datetime import datetime
+
+import enapter
+
+def parse_json(bytes):
+    return json.loads(bytes.decode())
+
+
+async def main():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        device_factory = functools.partial(
+            NIDAQ,
+            socket=sock,
+            tcp_port=os.environ["LISTEN_TCP_PORT"],
+        )
+        await enapter.vucm.run(device_factory)
+
+
+class NIDAQ(enapter.vucm.Device):
+    def __init__(self, socket, tcp_port, **kwargs):
+        super().__init__(**kwargs)
+        self.socket = socket
+        self.tcp_port = tcp_port
+
+    async def task_properties_sender(self):
+        while True:
+            await self.send_properties(
+                {
+                    "model": "NI cDAQ 9178",
+                }
+            )
+            await asyncio.sleep(10)
+
+    async def task_telemetry_sender(self):
+        server_address = ('localhost', int(self.tcp_port))
+        self.socket.bind(server_address)
+        self.socket.setblocking(False)
+        self.socket.listen(1)
+        
+        while True:
+            try:
+                await self.log.info('waiting for a connection')
+                connection, client_address = await asyncio.get_event_loop().sock_accept(self.socket)
+
+                await self.log.info(f'connection from {client_address}')
+                data = bytearray()
+                
+                while True:
+                    try:
+                        received = await asyncio.get_event_loop().sock_recv(connection, 1024)
+                        if not received:
+                            await self.log.info(f'no more data from {client_address}')
+                            break
+                        data.extend(received)
+                        # await self.log.info(f'got data: {received}')
+                    except Exception as e:
+                        await self.log.error(f"Error receiving data: {e}")
+                        break
+                
+                try:
+                    telemetry = dict()
+
+                    if len(data) != 0:
+                        telemetry["status"] = "ok" # TODO: define status
+                        telemetry = parse_json(data)
+                        if telemetry['Date'] and telemetry['Time']:
+                            tt = telemetry['Date'] + ' ' + telemetry['Time']
+                            date = datetime.strptime(tt,'%d/%m/%Y %H:%M:%S')
+                            telemetry.pop('Date')
+                            telemetry.pop('Time')
+                            telemetry["timestamp"] = int(time.time())
+                       
+                        await self.log.info(f'data to send: {telemetry}')
+                    await self.send_telemetry(telemetry)
+                    self.alerts.clear()
+                except Exception as e:
+                    self.alerts.add("parse_error")
+                    await self.log.error(f"failed to process data: {e}")
+                    
+            except Exception as e:
+                await self.log.error(f"Connection error: {e}")
+            finally:
+                connection.close()
+                await asyncio.sleep(1)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+