""" Dashboard Configuration Manager Provides comprehensive SCADA and hardware configuration through the dashboard """ import json import logging from typing import Dict, List, Optional, Any from pydantic import BaseModel, validator from enum import Enum logger = logging.getLogger(__name__) class ProtocolType(str, Enum): OPC_UA = "opcua" MODBUS_TCP = "modbus_tcp" MODBUS_RTU = "modbus_rtu" REST_API = "rest_api" class SCADAProtocolConfig(BaseModel): """Base SCADA protocol configuration""" protocol_type: ProtocolType enabled: bool = True name: str description: str = "" class OPCUAConfig(SCADAProtocolConfig): """OPC UA protocol configuration""" protocol_type: ProtocolType = ProtocolType.OPC_UA endpoint: str = "opc.tcp://0.0.0.0:4840" security_policy: str = "Basic256Sha256" certificate_file: str = "/app/certs/server.pem" private_key_file: str = "/app/certs/server.key" @validator('endpoint') def validate_endpoint(cls, v): if not v.startswith("opc.tcp://"): raise ValueError("OPC UA endpoint must start with 'opc.tcp://'") return v class ModbusTCPConfig(SCADAProtocolConfig): """Modbus TCP protocol configuration""" protocol_type: ProtocolType = ProtocolType.MODBUS_TCP host: str = "0.0.0.0" port: int = 502 unit_id: int = 1 timeout: float = 5.0 @validator('port') def validate_port(cls, v): if not 1 <= v <= 65535: raise ValueError("Port must be between 1 and 65535") return v class DataPointMapping(BaseModel): """Data point mapping between protocol and internal representation""" protocol_type: ProtocolType station_id: str pump_id: str data_type: str # setpoint, actual_speed, status, etc. protocol_address: str # OPC UA node, Modbus register, etc. data_type_specific: Dict[str, Any] = {} class ProtocolMapping(BaseModel): """Unified protocol mapping configuration for all protocols""" id: str protocol_type: ProtocolType station_id: str equipment_id: str data_type_id: str protocol_address: str # register address or OPC UA node db_source: str # database table and column transformation_rules: List[Dict[str, Any]] = [] # Signal preprocessing configuration preprocessing_enabled: bool = False preprocessing_rules: List[Dict[str, Any]] = [] min_output_value: Optional[float] = None max_output_value: Optional[float] = None default_output_value: Optional[float] = None # Protocol-specific configurations modbus_config: Optional[Dict[str, Any]] = None opcua_config: Optional[Dict[str, Any]] = None @validator('id') def validate_id(cls, v): if not v.replace('_', '').isalnum(): raise ValueError("Mapping ID must be alphanumeric with underscores") return v @validator('station_id') def validate_station_id(cls, v): """Validate that station exists in tag metadata system""" from src.core.tag_metadata_manager import tag_metadata_manager if v and v not in tag_metadata_manager.stations: raise ValueError(f"Station '{v}' does not exist in tag metadata system") return v @validator('equipment_id') def validate_equipment_id(cls, v, values): """Validate that equipment exists in tag metadata system and belongs to station""" from src.core.tag_metadata_manager import tag_metadata_manager if v and v not in tag_metadata_manager.equipment: raise ValueError(f"Equipment '{v}' does not exist in tag metadata system") # Validate equipment belongs to station if 'station_id' in values and values['station_id']: equipment = tag_metadata_manager.equipment.get(v) if equipment and equipment.station_id != values['station_id']: raise ValueError(f"Equipment '{v}' does not belong to station '{values['station_id']}'") return v @validator('data_type_id') def validate_data_type_id(cls, v): """Validate that data type exists in tag metadata system""" from src.core.tag_metadata_manager import tag_metadata_manager if v and v not in tag_metadata_manager.data_types: raise ValueError(f"Data type '{v}' does not exist in tag metadata system") return v @validator('protocol_address') def validate_protocol_address(cls, v, values): if 'protocol_type' in values: if values['protocol_type'] == ProtocolType.MODBUS_TCP: try: address = int(v) if not (0 <= address <= 65535): raise ValueError("Modbus address must be between 0 and 65535") except ValueError: raise ValueError("Modbus address must be a valid integer") elif values['protocol_type'] == ProtocolType.MODBUS_RTU: try: address = int(v) if not (0 <= address <= 65535): raise ValueError("Modbus RTU address must be between 0 and 65535") except ValueError: raise ValueError("Modbus RTU address must be a valid integer") elif values['protocol_type'] == ProtocolType.OPC_UA: if not v.startswith('ns='): raise ValueError("OPC UA Node ID must start with 'ns='") elif values['protocol_type'] == ProtocolType.REST_API: if not v.startswith(('http://', 'https://')): raise ValueError("REST API endpoint must start with 'http://' or 'https://'") return v def apply_preprocessing(self, value: float) -> float: """Apply preprocessing rules to a value""" if not self.preprocessing_enabled: return value processed_value = value for rule in self.preprocessing_rules: rule_type = rule.get('type') params = rule.get('parameters', {}) if rule_type == 'scale': processed_value *= params.get('factor', 1.0) elif rule_type == 'offset': processed_value += params.get('offset', 0.0) elif rule_type == 'clamp': min_val = params.get('min', float('-inf')) max_val = params.get('max', float('inf')) processed_value = max(min_val, min(processed_value, max_val)) elif rule_type == 'linear_map': # Map from [input_min, input_max] to [output_min, output_max] input_min = params.get('input_min', 0.0) input_max = params.get('input_max', 1.0) output_min = params.get('output_min', 0.0) output_max = params.get('output_max', 1.0) if input_max == input_min: processed_value = output_min else: normalized = (processed_value - input_min) / (input_max - input_min) processed_value = output_min + normalized * (output_max - output_min) elif rule_type == 'deadband': # Apply deadband to prevent oscillation center = params.get('center', 0.0) width = params.get('width', 0.0) if abs(processed_value - center) <= width: processed_value = center # Apply final output limits if self.min_output_value is not None: processed_value = max(self.min_output_value, processed_value) if self.max_output_value is not None: processed_value = min(self.max_output_value, processed_value) return processed_value class HardwareDiscoveryResult(BaseModel): """Result from hardware auto-discovery""" success: bool discovered_stations: List[Dict[str, Any]] = [] discovered_pumps: List[Dict[str, Any]] = [] errors: List[str] = [] warnings: List[str] = [] class ConfigurationManager: """Manages comprehensive system configuration through dashboard""" def __init__(self, db_client=None): self.protocol_configs: Dict[ProtocolType, SCADAProtocolConfig] = {} self.data_mappings: List[DataPointMapping] = [] self.protocol_mappings: List[ProtocolMapping] = [] self.db_client = db_client # Load mappings from database if available if self.db_client: self._load_mappings_from_db() def _load_mappings_from_db(self): """Load protocol mappings from database""" try: query = """ SELECT mapping_id, station_id, equipment_id, protocol_type, protocol_address, data_type_id, db_source, enabled FROM protocol_mappings WHERE enabled = true ORDER BY station_id, equipment_id, protocol_type """ results = self.db_client.execute_query(query) logger.info(f"Database query returned {len(results)} rows") for row in results: try: # Convert protocol_type string to enum protocol_type = ProtocolType(row['protocol_type']) mapping = ProtocolMapping( id=row['mapping_id'], station_id=row['station_id'], equipment_id=row['equipment_id'], protocol_type=protocol_type, protocol_address=row['protocol_address'], data_type_id=row['data_type_id'], db_source=row['db_source'] ) self.protocol_mappings.append(mapping) logger.debug(f"Loaded mapping {row['mapping_id']}: {protocol_type}") except Exception as e: logger.error(f"Failed to create mapping for {row['mapping_id']}: {str(e)}") logger.info(f"Loaded {len(self.protocol_mappings)} protocol mappings from database") except Exception as e: logger.error(f"Failed to load protocol mappings from database: {str(e)}") def configure_protocol(self, config: SCADAProtocolConfig) -> bool: """Configure a SCADA protocol""" try: self.protocol_configs[config.protocol_type] = config logger.info(f"Configured {config.protocol_type.value} protocol: {config.name}") return True except Exception as e: logger.error(f"Failed to configure protocol {config.protocol_type}: {str(e)}") return False def map_data_point(self, mapping: DataPointMapping) -> bool: """Map a data point between protocol and internal representation""" try: # Verify protocol is configured if mapping.protocol_type not in self.protocol_configs: raise ValueError(f"Protocol {mapping.protocol_type} is not configured") # Verify pump exists if mapping.pump_id not in self.pumps: raise ValueError(f"Pump {mapping.pump_id} does not exist") self.data_mappings.append(mapping) logger.info(f"Mapped {mapping.data_type} for pump {mapping.pump_id} to {mapping.protocol_address}") return True except Exception as e: logger.error(f"Failed to map data point for {mapping.pump_id}: {str(e)}") return False def add_protocol_mapping(self, mapping: ProtocolMapping) -> bool: """Add a new protocol mapping with validation""" try: # Validate the mapping validation_result = self.validate_protocol_mapping(mapping) if not validation_result['valid']: raise ValueError(f"Mapping validation failed: {', '.join(validation_result['errors'])}") # # # Verify pump exists # if mapping.pump_id not in self.pumps: # raise ValueError(f"Pump {mapping.pump_id} does not exist") # # # Verify station exists # if mapping.station_id not in self.stations: # raise ValueError(f"Station {mapping.station_id} does not exist") # Save to database if available if self.db_client: query = """ INSERT INTO protocol_mappings (mapping_id, station_id, equipment_id, protocol_type, protocol_address, data_type_id, db_source, created_by, enabled) VALUES (:mapping_id, :station_id, :equipment_id, :protocol_type, :protocol_address, :data_type_id, :db_source, :created_by, :enabled) ON CONFLICT (mapping_id) DO UPDATE SET station_id = EXCLUDED.station_id, equipment_id = EXCLUDED.equipment_id, protocol_type = EXCLUDED.protocol_type, protocol_address = EXCLUDED.protocol_address, data_type_id = EXCLUDED.data_type_id, db_source = EXCLUDED.db_source, enabled = EXCLUDED.enabled, updated_at = CURRENT_TIMESTAMP """ params = { 'mapping_id': mapping.id, 'station_id': mapping.station_id, 'equipment_id': mapping.equipment_id, 'protocol_type': mapping.protocol_type.value, 'protocol_address': mapping.protocol_address, 'data_type_id': mapping.data_type_id, 'db_source': mapping.db_source, 'created_by': 'dashboard', 'enabled': True } self.db_client.execute(query, params) self.protocol_mappings.append(mapping) logger.info(f"Added protocol mapping {mapping.id}: {mapping.protocol_type} for {mapping.station_id}/{mapping.equipment_id}") return True except Exception as e: logger.error(f"Failed to add protocol mapping {mapping.id}: {str(e)}") return False def get_protocol_mappings(self, protocol_type: Optional[ProtocolType] = None, station_id: Optional[str] = None, equipment_id: Optional[str] = None) -> List[ProtocolMapping]: """Get mappings filtered by protocol/station/equipment""" filtered_mappings = self.protocol_mappings.copy() if protocol_type: filtered_mappings = [m for m in filtered_mappings if m.protocol_type == protocol_type] if station_id: filtered_mappings = [m for m in filtered_mappings if m.station_id == station_id] if equipment_id: filtered_mappings = [m for m in filtered_mappings if m.equipment_id == equipment_id] return filtered_mappings def update_protocol_mapping(self, mapping_id: str, updated_mapping: ProtocolMapping) -> bool: """Update an existing protocol mapping""" try: # Find the mapping to update for i, mapping in enumerate(self.protocol_mappings): if mapping.id == mapping_id: # Validate the updated mapping (exclude current mapping from conflict check) validation_result = self.validate_protocol_mapping(updated_mapping, exclude_mapping_id=mapping_id) if not validation_result['valid']: raise ValueError(f"Mapping validation failed: {', '.join(validation_result['errors'])}") # Update in database if available if self.db_client: query = """ UPDATE protocol_mappings SET station_id = :station_id, equipment_id = :equipment_id, protocol_type = :protocol_type, protocol_address = :protocol_address, data_type_id = :data_type_id, db_source = :db_source, updated_at = CURRENT_TIMESTAMP WHERE mapping_id = :mapping_id """ params = { 'mapping_id': mapping_id, 'station_id': updated_mapping.station_id, 'equipment_id': updated_mapping.equipment_id, 'protocol_type': updated_mapping.protocol_type.value, 'protocol_address': updated_mapping.protocol_address, 'data_type_id': updated_mapping.data_type_id, 'db_source': updated_mapping.db_source } self.db_client.execute(query, params) self.protocol_mappings[i] = updated_mapping logger.info(f"Updated protocol mapping {mapping_id}") return True raise ValueError(f"Protocol mapping {mapping_id} not found") except Exception as e: logger.error(f"Failed to update protocol mapping {mapping_id}: {str(e)}") return False def delete_protocol_mapping(self, mapping_id: str) -> bool: """Delete a protocol mapping""" try: initial_count = len(self.protocol_mappings) self.protocol_mappings = [m for m in self.protocol_mappings if m.id != mapping_id] if len(self.protocol_mappings) < initial_count: # Delete from database if available if self.db_client: query = "DELETE FROM protocol_mappings WHERE mapping_id = :mapping_id" self.db_client.execute(query, {'mapping_id': mapping_id}) logger.info(f"Deleted protocol mapping {mapping_id}") return True else: raise ValueError(f"Protocol mapping {mapping_id} not found") except Exception as e: logger.error(f"Failed to delete protocol mapping {mapping_id}: {str(e)}") return False def validate_protocol_mapping(self, mapping: ProtocolMapping, exclude_mapping_id: Optional[str] = None) -> Dict[str, Any]: """Validate protocol mapping for conflicts and protocol-specific rules""" errors = [] warnings = [] # Check for ID conflicts (exclude current mapping when updating) for existing in self.protocol_mappings: if existing.id == mapping.id and existing.id != exclude_mapping_id: errors.append(f"Mapping ID '{mapping.id}' already exists") break # Protocol-specific validation if mapping.protocol_type == ProtocolType.MODBUS_TCP: # Modbus validation try: address = int(mapping.protocol_address) if not (0 <= address <= 65535): errors.append("Modbus address must be between 0 and 65535") # Check for address conflicts within same protocol for existing in self.protocol_mappings: if (existing.id != mapping.id and existing.protocol_type == ProtocolType.MODBUS_TCP and existing.protocol_address == mapping.protocol_address): errors.append(f"Modbus address {mapping.protocol_address} already used by {existing.station_id}/{existing.equipment_id}") break except ValueError: errors.append("Modbus address must be a valid integer") elif mapping.protocol_type == ProtocolType.OPC_UA: # OPC UA validation if not mapping.protocol_address.startswith('ns='): errors.append("OPC UA Node ID must start with 'ns='") # Check for node conflicts within same protocol for existing in self.protocol_mappings: if (existing.id != mapping.id and existing.protocol_type == ProtocolType.OPC_UA and existing.protocol_address == mapping.protocol_address): errors.append(f"OPC UA node {mapping.protocol_address} already used by {existing.station_id}/{existing.equipment_id}") break elif mapping.protocol_type == ProtocolType.MODBUS_RTU: # Modbus RTU validation (same as Modbus TCP) try: address = int(mapping.protocol_address) if not (0 <= address <= 65535): errors.append("Modbus RTU address must be between 0 and 65535") # Check for address conflicts within same protocol for existing in self.protocol_mappings: if (existing.id != mapping.id and existing.protocol_type == ProtocolType.MODBUS_RTU and existing.protocol_address == mapping.protocol_address): errors.append(f"Modbus RTU address {mapping.protocol_address} already used by {existing.station_id}/{existing.equipment_id}") break except ValueError: errors.append("Modbus RTU address must be a valid integer") elif mapping.protocol_type == ProtocolType.REST_API: # REST API validation if not mapping.protocol_address.startswith(('http://', 'https://')): errors.append("REST API endpoint must start with 'http://' or 'https://'") # Check for endpoint conflicts within same protocol for existing in self.protocol_mappings: if (existing.id != mapping.id and existing.protocol_type == ProtocolType.REST_API and existing.protocol_address == mapping.protocol_address): errors.append(f"REST API endpoint {mapping.protocol_address} already used by {existing.station_id}/{existing.equipment_id}") break # Check database source format if '.' not in mapping.db_source: warnings.append("Database source should be in format 'table.column'") return { 'valid': len(errors) == 0, 'errors': errors, 'warnings': warnings } def auto_discover_hardware(self) -> HardwareDiscoveryResult: """Auto-discover connected hardware and SCADA systems""" result = HardwareDiscoveryResult(success=True) try: # This would integrate with actual hardware discovery # For now, provide mock discovery for demonstration # Mock OPC UA discovery if ProtocolType.OPC_UA in self.protocol_configs: logger.info("Performing OPC UA hardware discovery...") # Simulate discovering a station via OPC UA mock_station = { "station_id": "discovered_station_001", "name": "Discovered Pump Station", "location": "Building A", "max_pumps": 2, "power_capacity": 100.0 } result.discovered_stations.append(mock_station) # Simulate discovering pumps mock_pump = { "pump_id": "discovered_pump_001", "station_id": "discovered_station_001", "name": "Discovered Primary Pump", "type": "centrifugal", "power_rating": 55.0, "max_speed": 50.0, "min_speed": 20.0 } result.discovered_pumps.append(mock_pump) # Mock Modbus discovery if ProtocolType.MODBUS_TCP in self.protocol_configs: logger.info("Performing Modbus TCP hardware discovery...") result.warnings.append("Modbus discovery requires manual configuration") logger.info(f"Hardware discovery completed: {len(result.discovered_stations)} stations, {len(result.discovered_pumps)} pumps found") except Exception as e: result.success = False result.errors.append(f"Hardware discovery failed: {str(e)}") logger.error(f"Hardware discovery failed: {str(e)}") return result def validate_configuration(self) -> Dict[str, Any]: """Validate the complete configuration""" validation_result = { "valid": True, "errors": [], "warnings": [], "summary": {} } # Check protocol configurations if not self.protocol_configs: validation_result["warnings"].append("No SCADA protocols configured") # Check stations and pumps if not self.stations: validation_result["warnings"].append("No pump stations configured") # Check data mappings if not self.data_mappings: validation_result["warnings"].append("No data point mappings configured") # Check protocol mappings if not self.protocol_mappings: validation_result["warnings"].append("No protocol mappings configured") # Check safety limits pumps_without_limits = set(self.pumps.keys()) - set(limit.pump_id for limit in self.safety_limits.values()) if pumps_without_limits: validation_result["warnings"].append(f"Pumps without safety limits: {', '.join(pumps_without_limits)}") # Validate individual protocol mappings for mapping in self.protocol_mappings: mapping_validation = self.validate_protocol_mapping(mapping) if not mapping_validation['valid']: validation_result['errors'].extend([f"Mapping {mapping.id}: {error}" for error in mapping_validation['errors']]) validation_result['warnings'].extend([f"Mapping {mapping.id}: {warning}" for warning in mapping_validation['warnings']]) # Create summary validation_result["summary"] = { "protocols_configured": len(self.protocol_configs), "data_mappings": len(self.data_mappings), "protocol_mappings": len(self.protocol_mappings) } return validation_result def export_configuration(self) -> Dict[str, Any]: """Export complete configuration for backup""" return { "protocols": {pt.value: config.dict() for pt, config in self.protocol_configs.items()}, "data_mappings": [mapping.dict() for mapping in self.data_mappings], "protocol_mappings": [mapping.dict() for mapping in self.protocol_mappings] } def import_configuration(self, config_data: Dict[str, Any]) -> bool: """Import configuration from backup""" try: # Clear existing configuration self.protocol_configs.clear() self.data_mappings.clear() self.protocol_mappings.clear() # Import protocols for pt_str, config_dict in config_data.get("protocols", {}).items(): protocol_type = ProtocolType(pt_str) if protocol_type == ProtocolType.OPC_UA: config = OPCUAConfig(**config_dict) elif protocol_type == ProtocolType.MODBUS_TCP: config = ModbusTCPConfig(**config_dict) else: config = SCADAProtocolConfig(**config_dict) self.protocol_configs[protocol_type] = config # Import data mappings for mapping_dict in config_data.get("data_mappings", []): mapping = DataPointMapping(**mapping_dict) self.data_mappings.append(mapping) # Import protocol mappings for mapping_dict in config_data.get("protocol_mappings", []): mapping = ProtocolMapping(**mapping_dict) self.protocol_mappings.append(mapping) logger.info("Configuration imported successfully") return True except Exception as e: logger.error(f"Failed to import configuration: {str(e)}") return False # Global configuration manager instance configuration_manager = ConfigurationManager()