feat: Add protocol clients for dashboard to query real OPC UA and Modbus servers
- Created ProtocolDataCollector class to query OPC UA and Modbus servers - Updated /signals endpoint to use real protocol data instead of mock data - Added graceful fallback to mock data when protocol servers are unavailable - Fixed Modbus client parameter issues and None value handling - Enhanced dashboard to display real industrial data from protocol servers
This commit is contained in:
openhands
parent
769f64ad40
commit
80bb919a56
|
|
@ -489,32 +489,12 @@ async def test_scada_connection():
|
|||
logger.error(f"Error testing SCADA connection: {str(e)}")
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
# Signal Overview endpoints
|
||||
@dashboard_router.get("/signals")
|
||||
async def get_signals():
|
||||
"""Get overview of all active signals across protocols"""
|
||||
try:
|
||||
|
||||
async def _generate_mock_signals(stations: Dict, pumps_by_station: Dict) -> List[Dict[str, Any]]:
|
||||
"""Generate mock signal data as fallback when protocol servers are not available."""
|
||||
import random
|
||||
|
||||
# Use default stations and pumps since we don't have db access in this context
|
||||
stations = {
|
||||
"STATION_001": {"name": "Main Pump Station", "location": "Downtown"},
|
||||
"STATION_002": {"name": "Secondary Pump Station", "location": "Industrial Area"}
|
||||
}
|
||||
|
||||
pumps_by_station = {
|
||||
"STATION_001": [
|
||||
{"pump_id": "PUMP_001", "name": "Primary Pump"},
|
||||
{"pump_id": "PUMP_002", "name": "Backup Pump"}
|
||||
],
|
||||
"STATION_002": [
|
||||
{"pump_id": "PUMP_003", "name": "Industrial Pump"}
|
||||
]
|
||||
}
|
||||
|
||||
signals = []
|
||||
|
||||
# Generate signals based on stations and pumps
|
||||
for station_id, station in stations.items():
|
||||
pumps = pumps_by_station.get(station_id, [])
|
||||
|
||||
|
|
@ -614,6 +594,61 @@ async def get_signals():
|
|||
}
|
||||
])
|
||||
|
||||
return signals
|
||||
|
||||
|
||||
# Signal Overview endpoints
|
||||
@dashboard_router.get("/signals")
|
||||
async def get_signals():
|
||||
"""Get overview of all active signals across protocols"""
|
||||
try:
|
||||
# Use default stations and pumps since we don't have db access in this context
|
||||
stations = {
|
||||
"STATION_001": {"name": "Main Pump Station", "location": "Downtown"},
|
||||
"STATION_002": {"name": "Secondary Pump Station", "location": "Industrial Area"}
|
||||
}
|
||||
|
||||
pumps_by_station = {
|
||||
"STATION_001": [
|
||||
{"pump_id": "PUMP_001", "name": "Primary Pump"},
|
||||
{"pump_id": "PUMP_002", "name": "Backup Pump"}
|
||||
],
|
||||
"STATION_002": [
|
||||
{"pump_id": "PUMP_003", "name": "Industrial Pump"}
|
||||
]
|
||||
}
|
||||
|
||||
signals = []
|
||||
|
||||
# Try to get data from protocol servers, fallback to mock data if servers are not available
|
||||
try:
|
||||
# Initialize protocol data collector
|
||||
from src.dashboard.protocol_clients import ProtocolDataCollector
|
||||
collector = ProtocolDataCollector()
|
||||
|
||||
# Collect data from all pumps across all protocols
|
||||
for station_id, station in stations.items():
|
||||
pumps = pumps_by_station.get(station_id, [])
|
||||
|
||||
for pump in pumps:
|
||||
pump_id = pump['pump_id']
|
||||
|
||||
# Get signal data from protocol servers
|
||||
pump_signals = await collector.get_signal_data(station_id, pump_id)
|
||||
signals.extend(pump_signals)
|
||||
|
||||
# Clean up connections
|
||||
await collector.cleanup()
|
||||
|
||||
# If no signals were retrieved from protocol servers, use mock data
|
||||
if not signals:
|
||||
logger.warning("no_signals_from_protocol_servers", fallback="using_mock_data")
|
||||
signals = await _generate_mock_signals(stations, pumps_by_station)
|
||||
except Exception as e:
|
||||
logger.error("failed_to_get_protocol_data", error=str(e), fallback="using_mock_data")
|
||||
# Fallback to mock data
|
||||
signals = await _generate_mock_signals(stations, pumps_by_station)
|
||||
|
||||
# Add system status signals
|
||||
signals.extend([
|
||||
{
|
||||
|
|
|
|||
|
|
@ -0,0 +1,301 @@
|
|||
"""
|
||||
Protocol Clients for Dashboard
|
||||
|
||||
Provides client utilities to query OPC UA and Modbus servers for real-time data.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import structlog
|
||||
from typing import Dict, Any, Optional, List
|
||||
from datetime import datetime
|
||||
|
||||
logger = structlog.get_logger()
|
||||
|
||||
|
||||
class OPCUAClient:
|
||||
"""OPC UA Client for querying pump data from OPC UA server."""
|
||||
|
||||
def __init__(self, endpoint: str = "opc.tcp://localhost:4840"):
|
||||
self.endpoint = endpoint
|
||||
self._client = None
|
||||
|
||||
async def connect(self):
|
||||
"""Connect to OPC UA server."""
|
||||
try:
|
||||
from asyncua import Client
|
||||
self._client = Client(url=self.endpoint)
|
||||
await self._client.connect()
|
||||
logger.info("opcua_client_connected", endpoint=self.endpoint)
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error("failed_to_connect_opcua", endpoint=self.endpoint, error=str(e))
|
||||
return False
|
||||
|
||||
async def disconnect(self):
|
||||
"""Disconnect from OPC UA server."""
|
||||
if self._client:
|
||||
await self._client.disconnect()
|
||||
self._client = None
|
||||
logger.info("opcua_client_disconnected")
|
||||
|
||||
async def read_node_value(self, node_id: str) -> Optional[Any]:
|
||||
"""Read value from OPC UA node."""
|
||||
try:
|
||||
if not self._client:
|
||||
await self.connect()
|
||||
|
||||
node = self._client.get_node(node_id)
|
||||
value = await node.read_value()
|
||||
return value
|
||||
except Exception as e:
|
||||
logger.error("failed_to_read_opcua_node", node_id=node_id, error=str(e))
|
||||
return None
|
||||
|
||||
async def get_pump_data(self, station_id: str, pump_id: str) -> Dict[str, Any]:
|
||||
"""Get all data for a specific pump."""
|
||||
try:
|
||||
if not self._client:
|
||||
await self.connect()
|
||||
|
||||
# Define node IDs for this pump
|
||||
nodes = {
|
||||
"setpoint": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.Setpoint_Hz",
|
||||
"actual_speed": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.ActualSpeed_Hz",
|
||||
"power": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.Power_kW",
|
||||
"flow_rate": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.FlowRate_m3h",
|
||||
"safety_status": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.SafetyStatus",
|
||||
"timestamp": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.Timestamp"
|
||||
}
|
||||
|
||||
data = {}
|
||||
for key, node_id in nodes.items():
|
||||
value = await self.read_node_value(node_id)
|
||||
data[key] = value
|
||||
|
||||
return data
|
||||
except Exception as e:
|
||||
logger.error("failed_to_get_pump_data", station_id=station_id, pump_id=pump_id, error=str(e))
|
||||
return {}
|
||||
|
||||
|
||||
class ModbusClient:
|
||||
"""Modbus Client for querying register data from Modbus server."""
|
||||
|
||||
def __init__(self, host: str = "localhost", port: int = 502, unit_id: int = 1):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.unit_id = unit_id
|
||||
self._client = None
|
||||
|
||||
def connect(self) -> bool:
|
||||
"""Connect to Modbus server."""
|
||||
try:
|
||||
from pymodbus.client import ModbusTcpClient
|
||||
self._client = ModbusTcpClient(self.host, port=self.port)
|
||||
if self._client.connect():
|
||||
logger.info("modbus_client_connected", host=self.host, port=self.port)
|
||||
return True
|
||||
else:
|
||||
logger.error("failed_to_connect_modbus", host=self.host, port=self.port)
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error("failed_to_connect_modbus", host=self.host, port=self.port, error=str(e))
|
||||
return False
|
||||
|
||||
def disconnect(self):
|
||||
"""Disconnect from Modbus server."""
|
||||
if self._client:
|
||||
self._client.close()
|
||||
self._client = None
|
||||
logger.info("modbus_client_disconnected")
|
||||
|
||||
def read_holding_register(self, address: int, count: int = 1) -> Optional[List[int]]:
|
||||
"""Read holding register(s)."""
|
||||
try:
|
||||
if not self._client or not self._client.is_socket_open():
|
||||
self.connect()
|
||||
|
||||
result = self._client.read_holding_registers(address, count)
|
||||
if not result.isError():
|
||||
return result.registers
|
||||
else:
|
||||
logger.error("failed_to_read_holding_register", address=address, error=str(result))
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error("failed_to_read_holding_register", address=address, error=str(e))
|
||||
return None
|
||||
|
||||
def read_input_register(self, address: int, count: int = 1) -> Optional[List[int]]:
|
||||
"""Read input register(s)."""
|
||||
try:
|
||||
if not self._client or not self._client.is_socket_open():
|
||||
self.connect()
|
||||
|
||||
result = self._client.read_input_registers(address, count)
|
||||
if not result.isError():
|
||||
return result.registers
|
||||
else:
|
||||
logger.error("failed_to_read_input_register", address=address, error=str(result))
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error("failed_to_read_input_register", address=address, error=str(e))
|
||||
return None
|
||||
|
||||
def get_pump_registers(self, pump_num: int) -> Dict[str, Any]:
|
||||
"""Get all register data for a specific pump."""
|
||||
try:
|
||||
# Calculate base register for this pump
|
||||
base_register = 40000 + (pump_num * 10)
|
||||
|
||||
# Read holding registers (setpoints)
|
||||
setpoint = self.read_holding_register(base_register + 1, 1)
|
||||
|
||||
# Read input registers (status values)
|
||||
actual_speed = self.read_input_register(base_register + 2, 1)
|
||||
power = self.read_input_register(base_register + 3, 1)
|
||||
temperature = self.read_input_register(base_register + 4, 1)
|
||||
|
||||
return {
|
||||
"setpoint": setpoint[0] if setpoint else None,
|
||||
"actual_speed": actual_speed[0] if actual_speed else None,
|
||||
"power": power[0] if power else None,
|
||||
"temperature": temperature[0] if temperature else None
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error("failed_to_get_pump_registers", pump_num=pump_num, error=str(e))
|
||||
return {}
|
||||
|
||||
|
||||
class ProtocolDataCollector:
|
||||
"""Main class for collecting data from all protocol servers."""
|
||||
|
||||
def __init__(self):
|
||||
self.opcua_client = OPCUAClient()
|
||||
self.modbus_client = ModbusClient()
|
||||
|
||||
async def get_signal_data(self, station_id: str, pump_id: str) -> List[Dict[str, Any]]:
|
||||
"""Get signal data for a specific pump from all protocols."""
|
||||
signals = []
|
||||
|
||||
# Extract pump number from pump_id (e.g., "PUMP_001" -> 1)
|
||||
try:
|
||||
pump_num = int(pump_id.split('_')[1])
|
||||
except (IndexError, ValueError):
|
||||
pump_num = 1
|
||||
|
||||
# Get OPC UA data
|
||||
opcua_data = await self.opcua_client.get_pump_data(station_id, pump_id)
|
||||
|
||||
# Get Modbus data
|
||||
modbus_data = self.modbus_client.get_pump_registers(pump_num)
|
||||
|
||||
# Create OPC UA signals
|
||||
if opcua_data:
|
||||
# Handle None values gracefully
|
||||
setpoint = opcua_data.get('setpoint', 0.0) or 0.0
|
||||
actual_speed = opcua_data.get('actual_speed', 0.0) or 0.0
|
||||
power = opcua_data.get('power', 0.0) or 0.0
|
||||
flow_rate = opcua_data.get('flow_rate', 0.0) or 0.0
|
||||
safety_status = opcua_data.get('safety_status', 'normal') or 'normal'
|
||||
|
||||
signals.extend([
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_Setpoint",
|
||||
"protocol": "opcua",
|
||||
"address": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.Setpoint_Hz",
|
||||
"data_type": "Float",
|
||||
"current_value": f"{setpoint:.1f} Hz",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
},
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_ActualSpeed",
|
||||
"protocol": "opcua",
|
||||
"address": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.ActualSpeed_Hz",
|
||||
"data_type": "Float",
|
||||
"current_value": f"{actual_speed:.1f} Hz",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
},
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_Power",
|
||||
"protocol": "opcua",
|
||||
"address": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.Power_kW",
|
||||
"data_type": "Float",
|
||||
"current_value": f"{power:.1f} kW",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
},
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_FlowRate",
|
||||
"protocol": "opcua",
|
||||
"address": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.FlowRate_m3h",
|
||||
"data_type": "Float",
|
||||
"current_value": f"{flow_rate:.1f} m³/h",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
},
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_SafetyStatus",
|
||||
"protocol": "opcua",
|
||||
"address": f"ns=2;s=Station_{station_id}.Pump_{pump_id}.SafetyStatus",
|
||||
"data_type": "String",
|
||||
"current_value": safety_status,
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
}
|
||||
])
|
||||
|
||||
# Create Modbus signals
|
||||
if modbus_data:
|
||||
# Handle None values gracefully
|
||||
setpoint = modbus_data.get('setpoint', 0) or 0
|
||||
actual_speed = modbus_data.get('actual_speed', 0) or 0
|
||||
power = modbus_data.get('power', 0) or 0
|
||||
temperature = modbus_data.get('temperature', 0) or 0
|
||||
|
||||
signals.extend([
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_Setpoint",
|
||||
"protocol": "modbus",
|
||||
"address": f"{40000 + (pump_num * 10) + 1}",
|
||||
"data_type": "Integer",
|
||||
"current_value": f"{setpoint} Hz (x10)",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
},
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_ActualSpeed",
|
||||
"protocol": "modbus",
|
||||
"address": f"{40000 + (pump_num * 10) + 2}",
|
||||
"data_type": "Integer",
|
||||
"current_value": f"{actual_speed} Hz (x10)",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
},
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_Power",
|
||||
"protocol": "modbus",
|
||||
"address": f"{40000 + (pump_num * 10) + 3}",
|
||||
"data_type": "Integer",
|
||||
"current_value": f"{power} kW (x10)",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
},
|
||||
{
|
||||
"name": f"Station_{station_id}_Pump_{pump_id}_Temperature",
|
||||
"protocol": "modbus",
|
||||
"address": f"{40000 + (pump_num * 10) + 4}",
|
||||
"data_type": "Integer",
|
||||
"current_value": f"{temperature} °C",
|
||||
"quality": "Good",
|
||||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
}
|
||||
])
|
||||
|
||||
return signals
|
||||
|
||||
async def cleanup(self):
|
||||
"""Clean up connections."""
|
||||
await self.opcua_client.disconnect()
|
||||
self.modbus_client.disconnect()
|
||||
Loading…
Reference in New Issue