src/services/simulation_service.py - master - devsec/anothertap

src/services/simulation_service.py
454 строки17 KB
devsec
clean code gitignore
24 авг 2025, 21:27
24 авг 2025, 21:2775a9527
"""
Device Simulation Service for testing without real hardware

This service provides simulation capabilities for various types of measurement devices,
allowing development and testing without requiring actual hardware connections.
"""

import time
import random
import logging
from typing import Optional, Dict, Any, Tuple, List
from threading import Lock
from datetime import datetime

from models.device import Device, DeviceStatus


class SimulatedDevice:
    """
    Base class for simulated devices that respond to SCPI commands
    """
    
    def __init__(self, device: Device):
        self.device = device
        self.logger = logging.getLogger(f"{self.__class__.__name__}_{device.id[:8]}")
        self.state: Dict[str, Any] = {}
        self.error_queue: List[str] = []
        self.last_command_time = time.time()
        
        # Initialize default state
        self._initialize_state()
    
    def _initialize_state(self) -> None:
        """Initialize device state with default values"""
        self.state = {
            'output_enabled': False,
            'voltage': 0.0,
            'current': 0.0,
            'frequency': 1000.0,
            'amplitude': 1.0,
            'measurement_data': [],
            'error_count': 0,
            'operation_complete': True
        }
    
    def process_command(self, command: str) -> Tuple[bool, Optional[str]]:
        """
        Process a SCPI command and return response
        
        Args:
            command: SCPI command string
            
        Returns:
            Tuple of (success, response). Response is None for commands, string for queries
        """
        self.last_command_time = time.time()
        command = command.strip()
        
        try:
            # Standard IEEE 488.2 commands
            if command.upper() == "*IDN?":
                return self._handle_idn_query()
            elif command.upper() == "*RST":
                return self._handle_reset()
            elif command.upper() == "*OPC?":
                return True, "1"
            elif command.upper() == "*OPC":
                self.state['operation_complete'] = True
                return True, None
            elif command.upper() == "*TST?":
                return True, "0"  # Self-test passed
            elif command.upper() == "*ESR?":
                return True, "0"  # No errors
            elif command.upper() == "*STB?":
                return True, "0"  # Status byte
            elif command.upper().startswith("*ESE"):
                return True, None  # Event status enable
            elif command.upper().startswith("*SRE"):
                return True, None  # Service request enable
            elif command.upper() == "*CLS":
                self.error_queue.clear()
                return True, None
            elif command.upper() == "SYST:ERR?":
                return self._handle_system_error()
            
            # Device-specific commands
            return self._handle_device_specific_command(command)
            
        except Exception as e:
            self.logger.error(f"Error processing command '{command}': {e}")
            self.error_queue.append(f"Execution error: {str(e)}")
            return False, None
    
    def _handle_idn_query(self) -> Tuple[bool, str]:
        """Handle *IDN? query"""
        manufacturer = self.device.manufacturer or "Simulated"
        model = self.device.model or "Device"
        serial = self.device.serial_number or "SIM123456"
        firmware = self.device.simulation_config.get('firmware_version', '1.0.0')
        
        return True, f"{manufacturer},{model},{serial},{firmware}"
    
    def _handle_reset(self) -> Tuple[bool, None]:
        """Handle *RST command"""
        self._initialize_state()
        return True, None
    
    def _handle_system_error(self) -> Tuple[bool, str]:
        """Handle SYST:ERR? query"""
        if self.error_queue:
            error = self.error_queue.pop(0)
            return True, f'"{error}"'
        else:
            return True, '"No error"'
    
    def _handle_device_specific_command(self, command: str) -> Tuple[bool, Optional[str]]:
        """
        Handle device-specific SCPI commands
        Override this method in device-specific simulators
        """
        # Default response for unknown commands
        self.error_queue.append(f"Undefined header: {command}")
        return False, None


class PowerSupplySimulator(SimulatedDevice):
    """Simulated power supply device"""
    
    def _initialize_state(self) -> None:
        super()._initialize_state()
        self.state.update({
            'voltage_setpoint': 0.0,
            'current_limit': 1.0,
            'output_enabled': False,
            'voltage_measured': 0.0,
            'current_measured': 0.0,
            'ovp_enabled': True,
            'ocp_enabled': True
        })
    
    def _handle_device_specific_command(self, command: str) -> Tuple[bool, Optional[str]]:
        cmd_upper = command.upper()
        
        # Voltage commands
        if cmd_upper.startswith("VOLT"):
            if "?" in cmd_upper:
                if "MEAS" in cmd_upper:
                    # Simulate measurement with small noise
                    noise = random.uniform(-0.01, 0.01)
                    measured = self.state['voltage_setpoint'] + noise
                    self.state['voltage_measured'] = measured
                    return True, f"{measured:.3f}"
                else:
                    return True, f"{self.state['voltage_setpoint']:.3f}"
            else:
                # Set voltage
                try:
                    value = float(cmd_upper.split()[1])
                    self.state['voltage_setpoint'] = value
                    return True, None
                except (IndexError, ValueError):
                    self.error_queue.append("Numeric data error")
                    return False, None
        
        # Current commands
        elif cmd_upper.startswith("CURR"):
            if "?" in cmd_upper:
                if "MEAS" in cmd_upper:
                    # Simulate current measurement
                    if self.state['output_enabled']:
                        current = min(self.state['current_limit'], 
                                    self.state['voltage_setpoint'] / 10.0)  # Assume 10 ohm load
                        noise = random.uniform(-0.001, 0.001)
                        measured = current + noise
                    else:
                        measured = 0.0
                    self.state['current_measured'] = measured
                    return True, f"{measured:.3f}"
                else:
                    return True, f"{self.state['current_limit']:.3f}"
            else:
                # Set current limit
                try:
                    value = float(cmd_upper.split()[1])
                    self.state['current_limit'] = value
                    return True, None
                except (IndexError, ValueError):
                    self.error_queue.append("Numeric data error")
                    return False, None
        
        # Output commands
        elif cmd_upper.startswith("OUTP"):
            if "?" in cmd_upper:
                return True, "1" if self.state['output_enabled'] else "0"
            else:
                if "ON" in cmd_upper or "1" in cmd_upper:
                    self.state['output_enabled'] = True
                elif "OFF" in cmd_upper or "0" in cmd_upper:
                    self.state['output_enabled'] = False
                return True, None
        
        return super()._handle_device_specific_command(command)


class MultimeterSimulator(SimulatedDevice):
    """Simulated digital multimeter"""
    
    def _initialize_state(self) -> None:
        super()._initialize_state()
        self.state.update({
            'function': 'VOLT:DC',
            'range': 'AUTO',
            'resolution': 'MAX',
            'trigger_source': 'IMM',
            'sample_count': 1
        })
    
    def _handle_device_specific_command(self, command: str) -> Tuple[bool, Optional[str]]:
        cmd_upper = command.upper()
        
        # Configure measurement function
        if cmd_upper.startswith("CONF"):
            if "VOLT" in cmd_upper:
                self.state['function'] = 'VOLT:DC'
            elif "CURR" in cmd_upper:
                self.state['function'] = 'CURR:DC'
            elif "RES" in cmd_upper:
                self.state['function'] = 'RES'
            elif "FREQ" in cmd_upper:
                self.state['function'] = 'FREQ'
            return True, None
        
        # Measure/Read commands
        elif cmd_upper.startswith(("MEAS", "READ", "FETC")):
            return self._simulate_measurement()
        
        # Function selection
        elif cmd_upper.startswith("FUNC"):
            if "?" in cmd_upper:
                return True, f'"{self.state["function"]}"'
            else:
                # Set function (simplified)
                return True, None
        
        return super()._handle_device_specific_command(command)
    
    def _simulate_measurement(self) -> Tuple[bool, str]:
        """Simulate a measurement based on current function"""
        function = self.state['function']
        
        if function == 'VOLT:DC':
            # Simulate DC voltage measurement
            base_value = random.uniform(-10.0, 10.0)
            noise = random.uniform(-0.001, 0.001)
            value = base_value + noise
        elif function == 'CURR:DC':
            # Simulate DC current measurement
            base_value = random.uniform(-1.0, 1.0)
            noise = random.uniform(-0.0001, 0.0001)
            value = base_value + noise
        elif function == 'RES':
            # Simulate resistance measurement
            base_value = random.uniform(100.0, 100000.0)
            noise = random.uniform(-1.0, 1.0)
            value = max(0, base_value + noise)
        elif function == 'FREQ':
            # Simulate frequency measurement
            base_value = random.uniform(50.0, 1000.0)
            noise = random.uniform(-0.1, 0.1)
            value = max(0, base_value + noise)
        else:
            value = 0.0
        
        return True, f"{value:.6f}"


class FunctionGeneratorSimulator(SimulatedDevice):
    """Simulated function generator"""
    
    def _initialize_state(self) -> None:
        super()._initialize_state()
        self.state.update({
            'frequency': 1000.0,
            'amplitude': 1.0,
            'offset': 0.0,
            'waveform': 'SIN',
            'output_enabled': False,
            'burst_enabled': False,
            'trigger_source': 'IMM'
        })
    
    def _handle_device_specific_command(self, command: str) -> Tuple[bool, Optional[str]]:
        cmd_upper = command.upper()
        
        # Frequency commands
        if cmd_upper.startswith("FREQ"):
            if "?" in cmd_upper:
                return True, f"{self.state['frequency']:.3f}"
            else:
                try:
                    value = float(cmd_upper.split()[1])
                    self.state['frequency'] = value
                    return True, None
                except (IndexError, ValueError):
                    self.error_queue.append("Numeric data error")
                    return False, None
        
        # Amplitude commands
        elif cmd_upper.startswith("VOLT"):
            if "?" in cmd_upper:
                return True, f"{self.state['amplitude']:.3f}"
            else:
                try:
                    value = float(cmd_upper.split()[1])
                    self.state['amplitude'] = value
                    return True, None
                except (IndexError, ValueError):
                    self.error_queue.append("Numeric data error")
                    return False, None
        
        # Waveform commands
        elif cmd_upper.startswith("FUNC"):
            if "?" in cmd_upper:
                return True, self.state['waveform']
            else:
                waveform = cmd_upper.split()[1] if len(cmd_upper.split()) > 1 else 'SIN'
                if waveform in ['SIN', 'SQU', 'TRI', 'RAMP', 'NOIS']:
                    self.state['waveform'] = waveform
                    return True, None
                else:
                    self.error_queue.append("Illegal parameter value")
                    return False, None
        
        # Output commands
        elif cmd_upper.startswith("OUTP"):
            if "?" in cmd_upper:
                return True, "1" if self.state['output_enabled'] else "0"
            else:
                if "ON" in cmd_upper or "1" in cmd_upper:
                    self.state['output_enabled'] = True
                elif "OFF" in cmd_upper or "0" in cmd_upper:
                    self.state['output_enabled'] = False
                return True, None
        
        return super()._handle_device_specific_command(command)


class SimulationService:
    """
    Service for managing simulated devices
    """
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self._simulators: Dict[str, SimulatedDevice] = {}
        self._lock = Lock()
        
        # Device type mapping
        self._simulator_types = {
            'power_supply': PowerSupplySimulator,
            'multimeter': MultimeterSimulator,
            'function_generator': FunctionGeneratorSimulator,
            'generic': SimulatedDevice  # Default simulator
        }
    
    def create_simulator(self, device: Device) -> bool:
        """
        Create a simulator for the given device
        
        Args:
            device: Device to simulate
            
        Returns:
            True if simulator created successfully
        """
        with self._lock:
            try:
                # Determine simulator type from device configuration
                device_type = device.simulation_config.get('device_type', 'generic')
                simulator_class = self._simulator_types.get(device_type, SimulatedDevice)
                
                # Create simulator instance
                simulator = simulator_class(device)
                self._simulators[device.id] = simulator
                
                self.logger.info(f"Created {device_type} simulator for device {device.name}")
                return True
                
            except Exception as e:
                self.logger.error(f"Failed to create simulator for device {device.name}: {e}")
                return False
    
    def remove_simulator(self, device_id: str) -> None:
        """Remove simulator for device"""
        with self._lock:
            if device_id in self._simulators:
                del self._simulators[device_id]
                self.logger.info(f"Removed simulator for device {device_id}")
    
    def send_command(self, device: Device, command: str) -> Tuple[bool, Optional[str]]:
        """
        Send command to simulated device
        
        Args:
            device: Target device
            command: SCPI command
            
        Returns:
            Tuple of (success, response)
        """
        if device.id not in self._simulators:
            self.logger.error(f"No simulator found for device {device.id}")
            return False, None
        
        simulator = self._simulators[device.id]
        
        # Add small delay to simulate real device communication
        time.sleep(0.01)
        
        return simulator.process_command(command)
    
    def get_available_device_types(self) -> List[str]:
        """Get list of available simulated device types"""
        return list(self._simulator_types.keys())
    
    def create_default_simulation_config(self, device_type: str) -> Dict[str, Any]:
        """Create default simulation configuration for device type"""
        configs = {
            'power_supply': {
                'device_type': 'power_supply',
                'max_voltage': 30.0,
                'max_current': 3.0,
                'firmware_version': '1.2.3'
            },
            'multimeter': {
                'device_type': 'multimeter',
                'voltage_ranges': ['100mV', '1V', '10V', '100V', '1000V'],
                'current_ranges': ['10mA', '100mA', '1A', '3A'],
                'firmware_version': '2.1.0'
            },
            'function_generator': {
                'device_type': 'function_generator',
                'max_frequency': 25000000,  # 25 MHz
                'max_amplitude': 10.0,  # 10V
                'waveforms': ['SIN', 'SQU', 'TRI', 'RAMP', 'NOIS'],
                'firmware_version': '3.0.1'
            },
            'generic': {
                'device_type': 'generic',
                'firmware_version': '1.0.0'
            }
        }
        
        return configs.get(device_type, configs['generic'])