/
devsec
/
anothertap
Обзор
Документация
Войти
/
devsec
/
anothertap
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/services/device_manager.py
510 строк
16 KB
devsec
clean code gitignore
24 авг 2025, 21:27
24 авг 2025, 21:27
75a9527
Код
Авторство
О чём код?
""" Device Management Service for CRUD operations on measurement devices """ import json import os import logging from typing import List, Optional, Dict, Any from pathlib import Path from models.device import Device, DeviceStatus, ConnectionType from services.scpi_service import SCPIService from services.simulation_service import SimulationService class DeviceManager: """ Service for managing measurement devices with CRUD operations """ def __init__(self, data_directory: str = "data"): self.logger = logging.getLogger(__name__) self.data_directory = Path(data_directory) self.devices_file = self.data_directory / "devices.json" self._devices: Dict[str, Device] = {} self.scpi_service = SCPIService() # Ensure data directory exists self.data_directory.mkdir(exist_ok=True) # Load existing devices self.load_devices() def create_device(self, device: Device) -> bool: """ Create a new device Args: device: Device to create Returns: True if creation successful, False otherwise """ try: if device.id in self._devices: self.logger.warning(f"Device with ID {device.id} already exists") return False self._devices[device.id] = device self.save_devices() self.logger.info(f"Created device: {device.display_name}") return True except Exception as e: self.logger.error(f"Failed to create device: {e}") return False def get_device(self, device_id: str) -> Optional[Device]: """ Get a device by ID Args: device_id: ID of the device to retrieve Returns: Device if found, None otherwise """ return self._devices.get(device_id) def get_all_devices(self) -> List[Device]: """ Get all devices Returns: List of all devices """ return list(self._devices.values()) def update_device(self, device: Device) -> bool: """ Update an existing device Args: device: Device to update Returns: True if update successful, False otherwise """ try: if device.id not in self._devices: self.logger.warning(f"Device with ID {device.id} not found") return False self._devices[device.id] = device self.save_devices() self.logger.info(f"Updated device: {device.display_name}") return True except Exception as e: self.logger.error(f"Failed to update device: {e}") return False def delete_device(self, device_id: str) -> bool: """ Delete a device Args: device_id: ID of the device to delete Returns: True if deletion successful, False otherwise """ try: if device_id not in self._devices: self.logger.warning(f"Device with ID {device_id} not found") return False device = self._devices[device_id] # Disconnect if connected if device.is_connected(): self.disconnect_device(device_id) del self._devices[device_id] self.save_devices() self.logger.info(f"Deleted device: {device.display_name}") return True except Exception as e: self.logger.error(f"Failed to delete device: {e}") return False def find_devices(self, **criteria) -> List[Device]: """ Find devices matching criteria Args: **criteria: Search criteria (name, manufacturer, model, etc.) Returns: List of devices matching criteria """ results = [] for device in self._devices.values(): match = True for key, value in criteria.items(): if not hasattr(device, key): match = False break device_value = getattr(device, key) if isinstance(device_value, str): if value.lower() not in device_value.lower(): match = False break else: if device_value != value: match = False break if match: results.append(device) return results def connect_device(self, device_id: str) -> bool: """ Connect to a device Args: device_id: ID of the device to connect to Returns: True if connection successful, False otherwise """ device = self.get_device(device_id) if not device: self.logger.error(f"Device {device_id} not found") return False success = self.scpi_service.connect_device(device) if success: self.save_devices() # Save updated status return success def disconnect_device(self, device_id: str) -> bool: """ Disconnect from a device Args: device_id: ID of the device to disconnect from Returns: True if disconnection successful, False otherwise """ device = self.get_device(device_id) if not device: self.logger.error(f"Device {device_id} not found") return False success = self.scpi_service.disconnect_device(device) if success: self.save_devices() # Save updated status return success def test_device_connection(self, device_id: str) -> bool: """ Test device connection Args: device_id: ID of the device to test Returns: True if connection is working, False otherwise """ device = self.get_device(device_id) if not device: return False return self.scpi_service.test_device_connection(device) def discover_devices(self) -> List[str]: """ Discover available VISA resources Returns: List of available resource strings """ return self.scpi_service.list_available_resources() def auto_identify_device(self, connection_string: str) -> Optional[Device]: """ Automatically identify a device from its connection string Args: connection_string: VISA connection string Returns: Device with populated information if successful, None otherwise """ try: # Create temporary device for identification temp_device = Device( name="Unknown Device", connection_string=connection_string ) # Try to connect and identify if self.scpi_service.connect_device(temp_device): # Device should now have updated information self.scpi_service.disconnect_device(temp_device) return temp_device else: return None except Exception as e: self.logger.error(f"Failed to auto-identify device: {e}") return None def get_connected_devices(self) -> List[Device]: """ Get all currently connected devices Returns: List of connected devices """ return [device for device in self._devices.values() if device.is_connected()] def get_disconnected_devices(self) -> List[Device]: """ Get all disconnected devices Returns: List of disconnected devices """ return [device for device in self._devices.values() if not device.is_connected()] def export_devices(self, file_path: str) -> bool: """ Export devices to JSON file Args: file_path: Path to export file Returns: True if export successful, False otherwise """ try: devices_data = [device.to_dict() for device in self._devices.values()] with open(file_path, 'w', encoding='utf-8') as f: json.dump({ 'devices': devices_data, 'version': '1.0.0', 'exported_at': str(Path().cwd()) }, f, indent=2, ensure_ascii=False) self.logger.info(f"Exported {len(devices_data)} devices to {file_path}") return True except Exception as e: self.logger.error(f"Failed to export devices: {e}") return False def import_devices(self, file_path: str, overwrite: bool = False) -> bool: """ Import devices from JSON file Args: file_path: Path to import file overwrite: Whether to overwrite existing devices Returns: True if import successful, False otherwise """ try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) devices_data = data.get('devices', []) imported_count = 0 for device_data in devices_data: device = Device.from_dict(device_data) if device.id in self._devices and not overwrite: self.logger.warning(f"Device {device.id} already exists, skipping") continue self._devices[device.id] = device imported_count += 1 if imported_count > 0: self.save_devices() self.logger.info(f"Imported {imported_count} devices from {file_path}") return True except Exception as e: self.logger.error(f"Failed to import devices: {e}") return False def save_devices(self): """Save all devices to JSON file""" try: devices_data = [device.to_dict() for device in self._devices.values()] with open(self.devices_file, 'w', encoding='utf-8') as f: json.dump({ 'devices': devices_data, 'version': '1.0.0' }, f, indent=2, ensure_ascii=False) except Exception as e: self.logger.error(f"Failed to save devices: {e}") def load_devices(self): """Load devices from JSON file""" try: if not self.devices_file.exists(): self.logger.info("No devices file found, starting with empty device list") return with open(self.devices_file, 'r', encoding='utf-8') as f: data = json.load(f) devices_data = data.get('devices', []) for device_data in devices_data: device = Device.from_dict(device_data) # Reset connection status on load device.status = DeviceStatus.DISCONNECTED self._devices[device.id] = device self.logger.info(f"Loaded {len(devices_data)} devices") except Exception as e: self.logger.error(f"Failed to load devices: {e}") def disconnect_all(self): """Disconnect from all devices""" self.scpi_service.disconnect_all() # Update device statuses for device in self._devices.values(): device.status = DeviceStatus.DISCONNECTED self.save_devices() def create_simulated_device(self, device_type: str, name: str, manufacturer: str = "", model: str = "") -> Optional[Device]: """ Create a new simulated device Args: device_type: Type of device to simulate (power_supply, multimeter, etc.) name: Name for the device manufacturer: Manufacturer name (optional) model: Model name (optional) Returns: Created device if successful, None otherwise """ try: # Get simulation service from SCPI service simulation_service = self.scpi_service.simulation_service # Create default simulation config simulation_config = simulation_service.create_default_simulation_config(device_type) # Create device with simulation settings device = Device( name=name or f"Simulated {device_type.replace('_', ' ').title()}", manufacturer=manufacturer or simulation_config.get('manufacturer', 'Simulated'), model=model or simulation_config.get('model', device_type.replace('_', ' ').title()), connection_type=ConnectionType.SIMULATION, connection_string=f"SIM::{device_type.upper()}", description=f"Simulated {device_type.replace('_', ' ')} device", simulation_mode=True, simulation_config=simulation_config ) # Add to device list if self.create_device(device): self.logger.info(f"Created simulated {device_type}: {device.display_name}") return device else: return None except Exception as e: self.logger.error(f"Failed to create simulated device: {e}") return None def get_available_simulation_types(self) -> List[str]: """ Get list of available simulated device types Returns: List of device type names """ return self.scpi_service.simulation_service.get_available_device_types() def create_simulation_examples(self) -> List[Device]: """ Create example simulated devices for demonstration Returns: List of created devices """ examples = [] try: # Power Supply example power_supply = self.create_simulated_device( device_type="power_supply", name="Sim Power Supply", manufacturer="Test Equipment Inc.", model="TPS-3030" ) if power_supply: examples.append(power_supply) # Multimeter example multimeter = self.create_simulated_device( device_type="multimeter", name="Sim Multimeter", manufacturer="Measurement Corp.", model="DMM-1000" ) if multimeter: examples.append(multimeter) # Function Generator example func_gen = self.create_simulated_device( device_type="function_generator", name="Sim Function Gen", manufacturer="Signal Systems", model="FG-2500" ) if func_gen: examples.append(func_gen) self.logger.info(f"Created {len(examples)} simulation examples") except Exception as e: self.logger.error(f"Failed to create simulation examples: {e}") return examples def get_simulated_devices(self) -> List[Device]: """ Get all simulated devices Returns: List of simulated devices """ return [device for device in self._devices.values() if device.simulation_mode or device.connection_type == ConnectionType.SIMULATION] def get_real_devices(self) -> List[Device]: """ Get all real (non-simulated) devices Returns: List of real devices """ return [device for device in self._devices.values() if not device.simulation_mode and device.connection_type != ConnectionType.SIMULATION]