diff --git a/deploy/factory/nodered-flows.json b/deploy/factory/nodered-flows.json new file mode 100644 index 0000000..ad384af --- /dev/null +++ b/deploy/factory/nodered-flows.json @@ -0,0 +1,287 @@ +[ + { + "id": "mascarade-factory-tab", + "type": "tab", + "label": "Mascarade Factory 4.0", + "disabled": false, + "info": "Sample flows: MQTT sensor → maintenance-predictor, operator query → factory-copilot, periodic log analysis → log-analyst" + }, + + {"id": "comment-flow1", "type": "comment", "z": "mascarade-factory-tab", "name": "── Flow 1: MQTT Sensor → Maintenance Predictor → Alert ──", "x": 350, "y": 40, "wires": []}, + + { + "id": "mqtt-broker-cfg", + "type": "mqtt-broker", + "name": "Factory MQTT Broker", + "broker": "localhost", + "port": "1883", + "clientid": "nodered-mascarade", + "autoConnect": true, + "keepalive": "60", + "cleansession": true + }, + + { + "id": "mqtt-sensor-in", + "type": "mqtt in", + "z": "mascarade-factory-tab", + "name": "Sensor Data (vibration/temp)", + "topic": "factory/sensors/+/telemetry", + "qos": "1", + "datatype": "json", + "broker": "mqtt-broker-cfg", + "x": 180, + "y": 100, + "wires": [["format-sensor-msg"]] + }, + + { + "id": "format-sensor-msg", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Format for Maintenance Predictor", + "func": "// Extract sensor readings and build a prompt for the maintenance-predictor agent\nvar sensor = msg.topic.split('/')[2];\nvar data = msg.payload;\n\nmsg.topic = 'maintenance-predictor';\nmsg.payload = {\n text: `Analyze sensor ${sensor} readings: vibration=${data.vibration || 'N/A'} mm/s, temperature=${data.temperature || 'N/A'}°C, current=${data.current || 'N/A'}A, rpm=${data.rpm || 'N/A'}. Assess failure risk 0-100 and recommend action.`\n};\nmsg.context = {\n sensor_id: sensor,\n raw_data: data,\n timestamp: new Date().toISOString()\n};\nreturn msg;", + "outputs": 1, + "x": 470, + "y": 100, + "wires": [["mascarade-send-maintenance"]] + }, + + { + "id": "mascarade-send-maintenance", + "type": "http request", + "z": "mascarade-factory-tab", + "name": "→ Mascarade (maintenance-predictor)", + "method": "POST", + "ret": "obj", + "paytoqs": "ignore", + "url": "http://localhost:7880/nodered/send", + "x": 760, + "y": 100, + "wires": [["check-risk-level"]] + }, + + { + "id": "check-risk-level", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Extract risk score", + "func": "// Parse agent response for risk score\nvar response = msg.payload.payload || msg.payload;\nvar riskMatch = String(response).match(/(\\d+)\\s*\\/\\s*100|risk[:\\s]*(\\d+)/i);\nvar risk = riskMatch ? parseInt(riskMatch[1] || riskMatch[2]) : 0;\n\nmsg.risk = risk;\nmsg.payload = {\n risk: risk,\n agent_response: response,\n sensor_id: (msg.payload.mascarade || {}).context ? msg.payload.mascarade.context.sensor_id : 'unknown',\n timestamp: new Date().toISOString()\n};\n\n// Route: output 0 = risk > 70 (alert), output 1 = risk <= 70 (log only)\nif (risk > 70) {\n return [msg, null];\n} else {\n return [null, msg];\n}", + "outputs": 2, + "x": 1040, + "y": 100, + "wires": [["alert-high-risk", "mqtt-alert-out"], ["debug-low-risk"]] + }, + + { + "id": "alert-high-risk", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Build alert notification", + "func": "msg.payload = {\n level: 'CRITICAL',\n title: `Maintenance Alert — Sensor ${msg.payload.sensor_id}`,\n risk: msg.payload.risk,\n message: msg.payload.agent_response,\n timestamp: msg.payload.timestamp\n};\nmsg.topic = 'factory/alerts/maintenance';\nreturn msg;", + "outputs": 1, + "x": 1300, + "y": 80, + "wires": [["debug-alert"]] + }, + + { + "id": "mqtt-alert-out", + "type": "mqtt out", + "z": "mascarade-factory-tab", + "name": "Publish alert to MQTT", + "topic": "factory/alerts/maintenance", + "qos": "1", + "retain": false, + "broker": "mqtt-broker-cfg", + "x": 1300, + "y": 120, + "wires": [] + }, + + { + "id": "debug-alert", + "type": "debug", + "z": "mascarade-factory-tab", + "name": "⚠ HIGH RISK ALERT", + "active": true, + "tosidebar": true, + "console": true, + "x": 1540, + "y": 80, + "wires": [] + }, + + { + "id": "debug-low-risk", + "type": "debug", + "z": "mascarade-factory-tab", + "name": "Low risk (log)", + "active": true, + "tosidebar": true, + "console": false, + "x": 1300, + "y": 160, + "wires": [] + }, + + {"id": "comment-flow2", "type": "comment", "z": "mascarade-factory-tab", "name": "── Flow 2: Operator Query → Factory Copilot → Response ──", "x": 350, "y": 240, "wires": []}, + + { + "id": "http-operator-in", + "type": "http in", + "z": "mascarade-factory-tab", + "name": "Operator query endpoint", + "url": "/api/operator/ask", + "method": "post", + "x": 180, + "y": 300, + "wires": [["format-copilot-msg"]] + }, + + { + "id": "format-copilot-msg", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Format for Factory Copilot", + "func": "var query = msg.payload.question || msg.payload.text || msg.payload;\nmsg.topic = 'factory-copilot';\nmsg.payload = {\n text: String(query)\n};\nmsg.context = {\n operator: msg.payload.operator || 'anonymous',\n station: msg.payload.station || 'unknown'\n};\nreturn msg;", + "outputs": 1, + "x": 460, + "y": 300, + "wires": [["mascarade-send-copilot"]] + }, + + { + "id": "mascarade-send-copilot", + "type": "http request", + "z": "mascarade-factory-tab", + "name": "→ Mascarade (factory-copilot)", + "method": "POST", + "ret": "obj", + "paytoqs": "ignore", + "url": "http://localhost:7880/nodered/send", + "x": 740, + "y": 300, + "wires": [["format-copilot-response"]] + }, + + { + "id": "format-copilot-response", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Format HTTP response", + "func": "msg.payload = {\n answer: msg.payload.payload || msg.payload,\n agent: 'factory-copilot',\n timestamp: new Date().toISOString()\n};\nreturn msg;", + "outputs": 1, + "x": 1020, + "y": 300, + "wires": [["http-operator-out"]] + }, + + { + "id": "http-operator-out", + "type": "http response", + "z": "mascarade-factory-tab", + "name": "Send response", + "statusCode": "200", + "x": 1260, + "y": 300, + "wires": [] + }, + + {"id": "comment-flow3", "type": "comment", "z": "mascarade-factory-tab", "name": "── Flow 3: Periodic Log Analysis → Log Analyst → Shift Report ──", "x": 370, "y": 400, "wires": []}, + + { + "id": "cron-shift-end", + "type": "inject", + "z": "mascarade-factory-tab", + "name": "Every 8h (shift end)", + "props": [{"p": "payload"}, {"p": "topic", "vt": "str"}], + "repeat": "", + "crontab": "00 06,14,22 * * *", + "once": false, + "onceDelay": "0.1", + "topic": "log-analyst", + "payload": "", + "payloadType": "date", + "x": 180, + "y": 460, + "wires": [["fetch-shift-logs"]] + }, + + { + "id": "fetch-shift-logs", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Collect logs for shift period", + "func": "// Calculate shift window (last 8 hours)\nvar now = new Date();\nvar shiftStart = new Date(now.getTime() - 8 * 60 * 60 * 1000);\n\nmsg.topic = 'log-analyst';\nmsg.payload = {\n text: `Generate a shift report for the period ${shiftStart.toISOString()} to ${now.toISOString()}. Summarize: production counts, downtime events, quality alerts, safety incidents. Format as a structured shift handover report.`\n};\nmsg.context = {\n shift_start: shiftStart.toISOString(),\n shift_end: now.toISOString(),\n report_type: 'shift_handover'\n};\nreturn msg;", + "outputs": 1, + "x": 470, + "y": 460, + "wires": [["mascarade-send-loganalyst"]] + }, + + { + "id": "mascarade-send-loganalyst", + "type": "http request", + "z": "mascarade-factory-tab", + "name": "→ Mascarade (log-analyst)", + "method": "POST", + "ret": "obj", + "paytoqs": "ignore", + "url": "http://localhost:7880/nodered/send", + "x": 740, + "y": 460, + "wires": [["format-shift-report"]] + }, + + { + "id": "format-shift-report", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Format shift report", + "func": "var report = msg.payload.payload || msg.payload;\nmsg.payload = {\n report: report,\n generated_at: new Date().toISOString(),\n shift_end: (msg.payload.mascarade || {}).context ? msg.payload.mascarade.context.shift_end : new Date().toISOString()\n};\nmsg.topic = 'factory/reports/shift';\nreturn msg;", + "outputs": 1, + "x": 1020, + "y": 460, + "wires": [["mqtt-report-out", "debug-report", "email-report"]] + }, + + { + "id": "mqtt-report-out", + "type": "mqtt out", + "z": "mascarade-factory-tab", + "name": "Publish report to MQTT", + "topic": "factory/reports/shift", + "qos": "1", + "retain": true, + "broker": "mqtt-broker-cfg", + "x": 1300, + "y": 440, + "wires": [] + }, + + { + "id": "debug-report", + "type": "debug", + "z": "mascarade-factory-tab", + "name": "Shift report", + "active": true, + "tosidebar": true, + "console": true, + "x": 1280, + "y": 480, + "wires": [] + }, + + { + "id": "email-report", + "type": "function", + "z": "mascarade-factory-tab", + "name": "Prepare email (configure e-mail node)", + "func": "// Placeholder: connect an e-mail node downstream to send the shift report\nmsg.topic = 'Shift Report — ' + new Date().toLocaleDateString();\nmsg.payload = msg.payload.report;\nreturn msg;", + "outputs": 1, + "x": 1340, + "y": 520, + "wires": [[]] + } +] diff --git a/docs/WEB_RESEARCH_FACTORY_4_0_OSS_2026-03-25.md b/docs/WEB_RESEARCH_FACTORY_4_0_OSS_2026-03-25.md new file mode 100644 index 0000000..f14a76f --- /dev/null +++ b/docs/WEB_RESEARCH_FACTORY_4_0_OSS_2026-03-25.md @@ -0,0 +1,190 @@ +# Web Research — Factory 4.0 Open-Source Tools + +Date: 2026-03-25 + +--- + +## 1. Predictive Maintenance & Time-Series Frameworks + +### PatchTST (via Time-Series-Library) +- **URL**: https://github.com/thuml/Time-Series-Library +- **Stars**: ~7k (Time-Series-Library umbrella) +- **Last update**: Active (2025-2026) +- **Description**: Patch-based Transformer for long-term time-series forecasting. Treats series as sequences of patches rather than individual time steps. Channel-independent design reduces computational cost. +- **Relevance**: Direct fit for maintenance-predictor agent — vibration/temperature trend forecasting. Can predict degradation curves from InfluxDB data. +- **Integration difficulty**: Medium. Requires PyTorch, training pipeline. Best used via NeuralForecast wrapper. + +### TimesNet (via Time-Series-Library) +- **URL**: https://github.com/thuml/Time-Series-Library +- **Stars**: Same umbrella repo +- **Last update**: Active +- **Description**: Temporal 2D-variation modeling for general time-series analysis (ICLR 2023). Handles forecasting, classification, imputation, and anomaly detection in a single architecture. +- **Relevance**: Multi-task capability ideal for factory: forecast + anomaly detection from the same model. Classification mode useful for failure-type identification. +- **Integration difficulty**: Medium. Same as PatchTST — PyTorch dependency, training pipeline needed. + +### NeuralForecast (Nixtla) +- **URL**: https://github.com/Nixtla/neuralforecast +- **Stars**: ~4,000 +- **Last update**: Active (v3.1.5, 2025-2026) +- **Description**: Unified interface for 30+ state-of-the-art neural forecasting models including PatchTST and TimesNet. Integrates with Ray/Optuna for hyperparameter optimization. Transfer learning support. +- **Relevance**: **Best entry point** for our stack. Single API wrapping PatchTST, TimesNet, NHITS, etc. Transfer learning means we can fine-tune on small factory datasets. +- **Integration difficulty**: Low-Medium. pip install, pandas DataFrame interface. Pair with InfluxDB export. + +### PyOD (Python Outlier Detection) +- **URL**: https://github.com/yzhao062/pyod +- **Stars**: ~8,500 +- **Last update**: Active +- **Description**: 20+ outlier/anomaly detection algorithms: isolation forest, autoencoders, LOF, ECOD, deep learning models. Unified API. +- **Relevance**: Ideal for real-time anomaly detection on sensor data. Complements time-series forecasting with point-anomaly detection. Lightweight, no training required for unsupervised methods. +- **Integration difficulty**: Low. pip install, sklearn-like API. Can run in maintenance-predictor agent. + +### ADTK (Anomaly Detection Toolkit) +- **URL**: https://github.com/arundo/adtk +- **Stars**: ~1,100 +- **Last update**: Maintenance mode (last release 0.6.2) +- **Description**: Rule-based and unsupervised anomaly detection specifically for time series. Built by Arundo for industrial IoT. Detectors, transformers, aggregators with pipe API. +- **Relevance**: Perfect for rule-based industrial thresholds (vibration > X mm/s). Simple to deploy, no ML training needed. Good complement to PyOD for structured rules. +- **Integration difficulty**: Low. pip install, pandas-native. Caveat: maintenance mode, may need forking for long-term use. + +--- + +## 2. Vision Inspection + +### Ultralytics YOLO (YOLOv8 / YOLO11 / YOLO26) +- **URL**: https://github.com/ultralytics/ultralytics +- **Stars**: ~55,000+ +- **Last update**: Very active (2026) +- **Description**: State-of-the-art object detection, segmentation, classification, pose estimation. YOLOv8 is the stable industrial workhorse; YOLO11 and upcoming YOLO26 add architectural improvements. +- **Relevance**: Core of quality-inspector agent. Detect defects, missing components, label errors on production line. Runs on Jetson Nano/Xavier for edge deployment. +- **Integration difficulty**: Low. pip install ultralytics, train with labeled images. ONNX/TensorRT export for edge. + +### Grounding DINO +- **URL**: https://github.com/IDEA-Research/GroundingDINO +- **Stars**: ~7,000+ +- **Last update**: Active (ECCV 2024 paper) +- **Description**: Open-set object detection with text prompts. No training needed — describe what to find in natural language. +- **Relevance**: Zero-shot defect detection: operator types "scratch on surface" and model finds it. Useful for rare defects where training data is insufficient. +- **Integration difficulty**: Medium. PyTorch + transformers. Heavier than YOLO, better suited for offline analysis or GPU-equipped stations. + +### Grounded-SAM (Grounding DINO + SAM2) +- **URL**: https://github.com/IDEA-Research/Grounded-Segment-Anything +- **Stars**: ~16,000+ +- **Last update**: Active (2025-2026, SAM2 integration) +- **Description**: Combines Grounding DINO detection with SAM2 segmentation. Text-prompted detection + pixel-perfect masks. Autodistill integration for auto-labeling YOLOv8 training data. +- **Relevance**: **Key pipeline**: use Grounded-SAM to auto-label defect images, then train lightweight YOLOv8 for production edge. Also useful for measuring defect area/dimensions. +- **Integration difficulty**: Medium-High. Requires GPU, multi-model pipeline. Best as offline labeling/analysis tool, not real-time edge. + +### SAM2 (Segment Anything Model 2) +- **URL**: https://github.com/facebookresearch/sam2 +- **Stars**: ~12,000+ +- **Last update**: Active +- **Description**: Meta's universal segmentation model. Video-capable. Supports prompted segmentation with points, boxes, or masks. +- **Relevance**: Video inspection on production lines (conveyor tracking). Segment parts in motion for counting, dimensional analysis. +- **Integration difficulty**: Medium. PyTorch, GPU recommended. Pairs with ComfyUI for visual pipeline building. + +--- + +## 3. MES / SCADA / OPC-UA + +### open62541 +- **URL**: https://github.com/open62541/open62541 +- **Stars**: ~3,000+ +- **Last update**: Active (2025-2026) +- **Description**: OPC-UA stack in pure C. Platform-independent, certified for Standard Server 2017 Profile. MPLv2 license (commercial-friendly). Suitable for embedded systems. +- **Relevance**: If we need a lightweight OPC-UA server on edge devices (Jetson, RPi). Our opcua_mcp.py uses asyncua (Python), but open62541 is the reference for embedded C deployments. +- **Integration difficulty**: High (C library). Use only if Python asyncua is insufficient for edge performance. + +### Eclipse Milo +- **URL**: https://github.com/eclipse-milo/milo +- **Stars**: ~1,100+ +- **Last update**: Active +- **Description**: Java OPC-UA client/server SDK. Reference implementation for Eclipse IoT. Used as basis for PLC4X OPC-UA integration. +- **Relevance**: Relevant if factory runs Java/JVM stack. Our Python stack prefers asyncua, but Milo is the go-to for JVM-based SCADA integration. +- **Integration difficulty**: Medium (Java). Not directly useful for our Python stack unless bridging via PLC4X. + +### Apache PLC4X +- **URL**: https://github.com/apache/plc4x / https://plc4x.apache.org/ +- **Stars**: ~1,200+ +- **Last update**: Active (Apache incubator graduate) +- **Description**: Universal PLC communication library. Supports S7 (Siemens), Modbus, ADS (Beckhoff), EtherNet/IP, OPC-UA, BACnet, KNX, and more. Java primary, Go secondary, C in progress. +- **Relevance**: **High value** for multi-vendor factories. Single library to talk to Siemens, Allen-Bradley, Beckhoff PLCs. Could replace multiple protocol-specific tools. +- **Integration difficulty**: Medium. Java-first (needs JVM). Python bindings limited. Could be called via HTTP gateway or used alongside our MCP servers. + +### OpenMES / Open Source MES +- **URL**: Various — no single dominant OSS MES project +- **Last update**: Fragmented landscape +- **Description**: Open source Manufacturing Execution Systems are rare. Closest options: Odoo Manufacturing module, ERPNext manufacturing, or custom builds on top of MQTT/InfluxDB/Grafana. +- **Relevance**: Our stack (Mascarade + MQTT + InfluxDB + Grafana) effectively acts as a lightweight MES. Better to extend our own stack than adopt a half-maintained OSS MES. +- **Integration difficulty**: N/A — recommend building on our existing stack instead. + +--- + +## 4. Node-RED Industrial Nodes + +### node-red-contrib-opcua +- **URL**: https://flows.nodered.org/node/node-red-contrib-opcua +- **Stars**: Most popular OPC-UA package for Node-RED +- **Last update**: Active +- **Description**: OPC-UA client/server nodes for Node-RED. Browse, read, write, subscribe to OPC-UA servers directly from flows. +- **Relevance**: Direct complement to our nodered_connector.py. Allows flows that read OPC-UA data AND send to Mascarade agents in the same pipeline. +- **Integration difficulty**: Low. npm install in Node-RED. + +### node-red-contrib-modbus +- **URL**: https://flows.nodered.org/node/node-red-contrib-modbus +- **Stars**: Most popular Modbus package +- **Last update**: Active +- **Description**: Full Modbus TCP/RTU/ASCII support. Read coils, registers, write outputs. Well-documented. +- **Relevance**: Essential for legacy PLC communication. Many older factory machines only support Modbus. +- **Integration difficulty**: Low. npm install. + +### node-red-contrib-s7 +- **URL**: https://flows.nodered.org/node/node-red-contrib-s7 +- **Last update**: Active +- **Description**: Direct Siemens S7 PLC communication (S7-300, S7-400, S7-1200, S7-1500). Read/write PLC variables. +- **Relevance**: Critical for Siemens-heavy factories. Bypasses OPC-UA overhead for direct S7 protocol access. +- **Integration difficulty**: Low. npm install. Requires S7 PLC network access. + +### node-red-contrib-mqtt-broker +- **URL**: Built into Node-RED core +- **Last update**: Always current +- **Description**: MQTT client nodes (subscribe, publish) are built into Node-RED core. No extra install needed. +- **Relevance**: Foundation of our Flow 1 (sensor data → Mascarade). Already used in nodered-flows.json. +- **Integration difficulty**: None. Built-in. + +### FlowFuse (Node-RED management) +- **URL**: https://flowfuse.com / https://github.com/FlowFuse/flowfuse +- **Stars**: ~600+ +- **Last update**: Active (2025-2026) +- **Description**: Enterprise Node-RED management platform. Multi-instance, team collaboration, DevOps pipelines for Node-RED flows. Open-source core. +- **Relevance**: Useful for scaling Node-RED across multiple factory sites. Manages flow deployment, version control, access control. +- **Integration difficulty**: Medium. Docker deployment. Adds operational overhead but valuable at scale. + +--- + +## Summary — Recommended Integration Priority + +| Priority | Tool | Use Case | Effort | +|----------|------|----------|--------| +| 1 | NeuralForecast (PatchTST/TimesNet) | Predictive maintenance forecasting | Medium | +| 1 | PyOD | Real-time anomaly detection | Low | +| 1 | Ultralytics YOLOv8 | Quality inspection edge | Low | +| 1 | node-red-contrib-opcua + modbus | Node-RED industrial protocol nodes | Low | +| 2 | Grounded-SAM | Auto-labeling pipeline for YOLOv8 | Medium | +| 2 | Apache PLC4X | Multi-vendor PLC gateway | Medium | +| 2 | ADTK | Rule-based threshold alerts | Low | +| 3 | FlowFuse | Multi-site Node-RED management | Medium | +| 3 | open62541 | Embedded OPC-UA (C) | High | + +Sources: +- [Time-Series-Library (PatchTST, TimesNet)](https://github.com/thuml/Time-Series-Library) +- [NeuralForecast](https://github.com/Nixtla/neuralforecast) +- [PyOD](https://github.com/yzhao062/pyod) +- [ADTK](https://github.com/arundo/adtk) +- [Ultralytics YOLO](https://github.com/ultralytics/ultralytics) +- [Grounding DINO](https://github.com/IDEA-Research/GroundingDINO) +- [Grounded-SAM](https://github.com/IDEA-Research/Grounded-Segment-Anything) +- [SAM2](https://github.com/facebookresearch/sam2) +- [open62541](https://github.com/open62541/open62541) +- [Eclipse Milo](https://github.com/eclipse-milo/milo) +- [Apache PLC4X](https://plc4x.apache.org/) +- [FlowFuse](https://github.com/FlowFuse/flowfuse) diff --git a/docs/plans/27_todo_factory_4_0_mcp_opcua_mqtt.md b/docs/plans/27_todo_factory_4_0_mcp_opcua_mqtt.md index de908e7..6d12888 100644 --- a/docs/plans/27_todo_factory_4_0_mcp_opcua_mqtt.md +++ b/docs/plans/27_todo_factory_4_0_mcp_opcua_mqtt.md @@ -33,7 +33,7 @@ ## P1 — Pipeline données - [ ] Pipeline InfluxDB → PatchTST/TimesNet pour maintenance prédictive -- [ ] Connecteur Node-RED → Mascarade (HTTP nodes) +- [x] Connecteur Node-RED → Mascarade (HTTP nodes) — `tools/industrial/nodered_connector.py` + `deploy/factory/nodered-flows.json` - [ ] Connecteur OpenMES/Odoo → MCP server - [ ] Dashboard Grafana template industriel (vibrations, température, courant) diff --git a/tools/industrial/nodered_connector.py b/tools/industrial/nodered_connector.py new file mode 100644 index 0000000..1c2538b --- /dev/null +++ b/tools/industrial/nodered_connector.py @@ -0,0 +1,315 @@ +#!/usr/bin/env python3 +"""Node-RED HTTP connector for Mascarade — expose agents as Node-RED flow nodes. + +Provides HTTP endpoints that Node-RED HTTP-request nodes can call: + POST /nodered/send — translate Node-RED msg → Mascarade send, return result as msg + GET /nodered/agents — list available Mascarade agents + GET /nodered/health — health check + +Run with: + python tools/industrial/nodered_connector.py + # or via uvicorn: + uvicorn tools.industrial.nodered_connector:app --host 0.0.0.0 --port 7880 +""" + +from __future__ import annotations + +import asyncio +import json +import os +import time +import uuid +from typing import Any + +# --------------------------------------------------------------------------- +# Config +# --------------------------------------------------------------------------- + +MASCARADE_URL = os.getenv("MASCARADE_URL", "http://localhost:8000") +LISTEN_HOST = os.getenv("NODERED_CONNECTOR_HOST", "0.0.0.0") +LISTEN_PORT = int(os.getenv("NODERED_CONNECTOR_PORT", "7880")) + +# Default agents exposed to Node-RED +DEFAULT_AGENTS = [ + { + "id": "factory-copilot", + "name": "Factory Copilot", + "description": "Operator assistant — queries machine data via OPC-UA/MQTT", + }, + { + "id": "maintenance-predictor", + "name": "Maintenance Predictor", + "description": "Time-series analysis, predictive maintenance alerts", + }, + { + "id": "log-analyst", + "name": "Log Analyst", + "description": "MES/ERP log reader, automatic shift report generation", + }, + { + "id": "quality-inspector", + "name": "Quality Inspector", + "description": "Vision-based quality control with YOLOv8/SAM2", + }, +] + +# --------------------------------------------------------------------------- +# HTTP client for Mascarade API +# --------------------------------------------------------------------------- + +try: + import httpx + + HAS_HTTPX = True +except ImportError: + HAS_HTTPX = False + +# Try lightweight server options +try: + from aiohttp import web + + HAS_AIOHTTP = True +except ImportError: + HAS_AIOHTTP = False + + +async def _mascarade_send(agent: str, message: str, context: dict | None = None) -> dict: + """Forward a message to a Mascarade agent and return its response.""" + payload = { + "agent": agent, + "message": message, + "stream": False, + } + if context: + payload["context"] = context + + if not HAS_HTTPX: + # Stub mode — return a simulated response + return { + "agent": agent, + "response": f"[stub] Agent '{agent}' received: {message[:120]}", + "timestamp": time.time(), + "stub": True, + } + + async with httpx.AsyncClient(timeout=60.0) as client: + resp = await client.post(f"{MASCARADE_URL}/v1/send", json=payload) + resp.raise_for_status() + return resp.json() + + +async def _mascarade_agents() -> list[dict]: + """Fetch live agent list from Mascarade, fall back to defaults.""" + if not HAS_HTTPX: + return DEFAULT_AGENTS + + try: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get(f"{MASCARADE_URL}/v1/agents") + resp.raise_for_status() + return resp.json().get("agents", DEFAULT_AGENTS) + except Exception: + return DEFAULT_AGENTS + + +# --------------------------------------------------------------------------- +# Node-RED msg format helpers +# --------------------------------------------------------------------------- + + +def nodered_msg_to_mascarade(msg: dict) -> tuple[str, str, dict | None]: + """Extract agent, message and optional context from a Node-RED msg object. + + Node-RED msg convention: + msg.topic → agent id (e.g. "maintenance-predictor") + msg.payload → user message (string or dict with "text" key) + msg.context → optional dict forwarded as Mascarade context + """ + agent = msg.get("topic", "factory-copilot") + raw_payload = msg.get("payload", "") + if isinstance(raw_payload, dict): + message = raw_payload.get("text", json.dumps(raw_payload)) + else: + message = str(raw_payload) + context = msg.get("context") + return agent, message, context + + +def mascarade_to_nodered_msg(result: dict, original_msg: dict | None = None) -> dict: + """Wrap a Mascarade response into a Node-RED msg object. + + Output msg: + msg.payload → agent response text + msg.topic → agent id + msg._msgid → unique id + msg.mascarade → full raw response + """ + out = { + "_msgid": str(uuid.uuid4()).replace("-", "")[:16], + "topic": result.get("agent", "unknown"), + "payload": result.get("response", ""), + "mascarade": result, + } + # Preserve any extra fields from the original msg + if original_msg: + for key in ("_msgid", "parts", "rate", "reset"): + if key in original_msg and key not in out: + out[key] = original_msg[key] + return out + + +# --------------------------------------------------------------------------- +# aiohttp web application +# --------------------------------------------------------------------------- + + +def _build_app() -> "web.Application": + """Create the aiohttp web application with Node-RED routes.""" + app = web.Application() + + async def handle_health(request: web.Request) -> web.Response: + return web.json_response( + { + "status": "ok", + "service": "nodered-mascarade-connector", + "mascarade_url": MASCARADE_URL, + "timestamp": time.time(), + } + ) + + async def handle_agents(request: web.Request) -> web.Response: + agents = await _mascarade_agents() + return web.json_response({"agents": agents}) + + async def handle_send(request: web.Request) -> web.Response: + """POST /nodered/send — main bridge endpoint. + + Accepts a Node-RED msg (JSON body) and returns a Node-RED msg. + """ + try: + body = await request.json() + except Exception: + return web.json_response( + {"error": "Invalid JSON body"}, status=400 + ) + + # Support both single msg and array of msgs (Node-RED batch) + msgs = body if isinstance(body, list) else [body] + results = [] + + for msg in msgs: + agent, message, context = nodered_msg_to_mascarade(msg) + try: + result = await _mascarade_send(agent, message, context) + out_msg = mascarade_to_nodered_msg(result, original_msg=msg) + results.append(out_msg) + except Exception as exc: + results.append( + { + "_msgid": str(uuid.uuid4()).replace("-", "")[:16], + "topic": agent, + "payload": f"Error: {exc}", + "error": str(exc), + } + ) + + # Return single msg or array depending on input + if isinstance(body, list): + return web.json_response(results) + return web.json_response(results[0]) + + app.router.add_get("/nodered/health", handle_health) + app.router.add_get("/nodered/agents", handle_agents) + app.router.add_post("/nodered/send", handle_send) + + return app + + +# --------------------------------------------------------------------------- +# Fallback: stdlib http server (no dependencies) +# --------------------------------------------------------------------------- + + +def _run_stdlib_server() -> None: + """Minimal stdlib HTTP server for environments without aiohttp.""" + from http.server import BaseHTTPRequestHandler, HTTPServer + + class Handler(BaseHTTPRequestHandler): + def _json(self, data: Any, status: int = 200) -> None: + body = json.dumps(data, ensure_ascii=False).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: # noqa: N802 + if self.path == "/nodered/health": + self._json( + { + "status": "ok", + "service": "nodered-mascarade-connector", + "mascarade_url": MASCARADE_URL, + "timestamp": time.time(), + } + ) + elif self.path == "/nodered/agents": + self._json({"agents": DEFAULT_AGENTS}) + else: + self._json({"error": "Not found"}, 404) + + def do_POST(self) -> None: # noqa: N802 + if self.path != "/nodered/send": + self._json({"error": "Not found"}, 404) + return + + length = int(self.headers.get("Content-Length", 0)) + raw = self.rfile.read(length) + try: + body = json.loads(raw) + except Exception: + self._json({"error": "Invalid JSON"}, 400) + return + + msgs = body if isinstance(body, list) else [body] + results = [] + for msg in msgs: + agent, message, context = nodered_msg_to_mascarade(msg) + result = asyncio.run(_mascarade_send(agent, message, context)) + out_msg = mascarade_to_nodered_msg(result, original_msg=msg) + results.append(out_msg) + + if isinstance(body, list): + self._json(results) + else: + self._json(results[0]) + + server = HTTPServer((LISTEN_HOST, LISTEN_PORT), Handler) + print(f"[nodered-connector] stdlib server on http://{LISTEN_HOST}:{LISTEN_PORT}") + server.serve_forever() + + +# --------------------------------------------------------------------------- +# ASGI app (for uvicorn) +# --------------------------------------------------------------------------- + +app = _build_app() if HAS_AIOHTTP else None + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + + +def main() -> None: + print(f"[nodered-connector] Mascarade URL: {MASCARADE_URL}") + print(f"[nodered-connector] Listening on {LISTEN_HOST}:{LISTEN_PORT}") + print(f"[nodered-connector] httpx={'yes' if HAS_HTTPX else 'STUB'}, aiohttp={'yes' if HAS_AIOHTTP else 'stdlib'}") + + if HAS_AIOHTTP: + web.run_app(_build_app(), host=LISTEN_HOST, port=LISTEN_PORT) + else: + _run_stdlib_server() + + +if __name__ == "__main__": + main()