""" Simplified Protocol Signal Models Migration from complex ID system to simple signal names + tags """ from typing import List, Optional, Dict, Any from pydantic import BaseModel, validator from enum import Enum import uuid import logging logger = logging.getLogger(__name__) class ProtocolType(str, Enum): """Supported protocol types""" OPCUA = "opcua" MODBUS_TCP = "modbus_tcp" MODBUS_RTU = "modbus_rtu" REST_API = "rest_api" class ProtocolSignal(BaseModel): """ Simplified protocol signal with human-readable name and tags Replaces the complex station_id/equipment_id/data_type_id system """ signal_id: str signal_name: str tags: List[str] protocol_type: ProtocolType protocol_address: str db_source: str # 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 # Metadata created_at: Optional[str] = None updated_at: Optional[str] = None created_by: Optional[str] = None enabled: bool = True @validator('signal_id') def validate_signal_id(cls, v): """Validate signal ID format""" if not v.replace('_', '').replace('-', '').isalnum(): raise ValueError("Signal ID must be alphanumeric with underscores and hyphens") return v @validator('signal_name') def validate_signal_name(cls, v): """Validate signal name is not empty""" if not v or not v.strip(): raise ValueError("Signal name cannot be empty") return v.strip() @validator('tags') def validate_tags(cls, v): """Validate tags format""" if not isinstance(v, list): raise ValueError("Tags must be a list") # Remove empty tags and normalize cleaned_tags = [] for tag in v: if tag and isinstance(tag, str) and tag.strip(): cleaned_tags.append(tag.strip().lower()) return cleaned_tags @validator('protocol_address') def validate_protocol_address(cls, v, values): """Validate protocol address based on protocol type""" if 'protocol_type' not in values: return v protocol_type = values['protocol_type'] if protocol_type == ProtocolType.MODBUS_TCP or protocol_type == ProtocolType.MODBUS_RTU: # Modbus addresses should be numeric if not v.isdigit(): raise ValueError(f"Modbus address must be numeric, got: {v}") address = int(v) if address < 0 or address > 65535: raise ValueError(f"Modbus address must be between 0 and 65535, got: {address}") elif protocol_type == ProtocolType.OPCUA: # OPC UA addresses should follow NodeId format if not v.startswith(('ns=', 'i=', 's=')): raise ValueError(f"OPC UA address should start with ns=, i=, or s=, got: {v}") elif protocol_type == ProtocolType.REST_API: # REST API addresses should be URLs or paths if not v.startswith('/'): raise ValueError(f"REST API address should start with /, got: {v}") return v class ProtocolSignalCreate(BaseModel): """Model for creating new protocol signals""" signal_name: str tags: List[str] protocol_type: ProtocolType protocol_address: str db_source: str 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 modbus_config: Optional[Dict[str, Any]] = None opcua_config: Optional[Dict[str, Any]] = None def generate_signal_id(self) -> str: """Generate a unique signal ID from the signal name""" base_id = self.signal_name.lower().replace(' ', '_').replace('/', '_') base_id = ''.join(c for c in base_id if c.isalnum() or c in ['_', '-']) # Add random suffix to ensure uniqueness random_suffix = uuid.uuid4().hex[:8] return f"{base_id}_{random_suffix}" class ProtocolSignalUpdate(BaseModel): """Model for updating existing protocol signals""" signal_name: Optional[str] = None tags: Optional[List[str]] = None protocol_type: Optional[ProtocolType] = None protocol_address: Optional[str] = None db_source: Optional[str] = None preprocessing_enabled: Optional[bool] = None preprocessing_rules: Optional[List[Dict[str, Any]]] = None min_output_value: Optional[float] = None max_output_value: Optional[float] = None default_output_value: Optional[float] = None modbus_config: Optional[Dict[str, Any]] = None opcua_config: Optional[Dict[str, Any]] = None enabled: Optional[bool] = None class ProtocolSignalFilter(BaseModel): """Model for filtering protocol signals""" tags: Optional[List[str]] = None protocol_type: Optional[ProtocolType] = None signal_name_contains: Optional[str] = None enabled: Optional[bool] = True class SignalDiscoveryResult(BaseModel): """Model for discovery results that can be converted to protocol signals""" device_name: str protocol_type: ProtocolType protocol_address: str data_point: str device_address: Optional[str] = None device_port: Optional[int] = None def to_protocol_signal_create(self) -> ProtocolSignalCreate: """Convert discovery result to protocol signal creation data""" signal_name = f"{self.device_name} {self.data_point}" # Generate meaningful tags from discovery data tags = [ f"device:{self.device_name.lower().replace(' ', '_')}", f"protocol:{self.protocol_type.value}", f"data_point:{self.data_point.lower().replace(' ', '_')}" ] if self.device_address: tags.append(f"address:{self.device_address}") return ProtocolSignalCreate( signal_name=signal_name, tags=tags, protocol_type=self.protocol_type, protocol_address=self.protocol_address, db_source=f"measurements.{self.device_name.lower().replace(' ', '_')}_{self.data_point.lower().replace(' ', '_')}" ) # Example usage: # discovery_result = SignalDiscoveryResult( # device_name="Water Pump Controller", # protocol_type=ProtocolType.MODBUS_TCP, # protocol_address="40001", # data_point="Speed", # device_address="192.168.1.100" # ) # # signal_create = discovery_result.to_protocol_signal_create() # print(signal_create.signal_name) # "Water Pump Controller Speed" # print(signal_create.tags) # ["device:water_pump_controller", "protocol:modbus_tcp", "data_point:speed", "address:192.168.1.100"]