Кибернетика, автоматика, протоколы

В приложении, код, в котором я только число переносов на новую строку поменял.

Он полностью создан AI.

Его можно очень вольно преобразовывать, менять структуру и логику просто командами текстом.

Наверное, можно даже не проверять (но я проверил, все так как надо, вплоть до комментариев, некоторые из которых вытянуты из логики вызывающей программы самостоятельно!)

Общее ощущение - странное. Собственно кодирование - куда круче чем я писал бы сам. И очень-очень бысто.

Но в плане постановки задачи - ты получаешь ровно то, что попросил.
Любое усложнение - только по отдельному запросу.

С живым человеком... живой человек мне бы сразу предложил пойти покурить в коридоре, пока он будет конфетку делать :)

Возможно это в платных версиях есть, я не в курсе, но вот так - очень непривычно.

И... драйвово 8)

#!/usr/bin/env python3
"""
Modbus configuration module with JSON persistence and custom slave context
"""
import json, struct, logging
from datetime import datetime
from pathlib import Path
from pymodbus.datastore import ModbusSlaveContext

log = logging.getLogger(__name__)


def float_to_modbus_regs(value):
    """Convert float to two Modbus registers (IEEE 754)"""
    bytes_val = struct.pack('>f', value)
    regs = struct.unpack('>HH', bytes_val)
    return list(regs)


def modbus_regs_to_float(reg1, reg2):
    """Convert two Modbus registers to float (IEEE 754)"""
    try:
        bytes_val = struct.pack('>HH', reg1, reg2)
        return struct.unpack('>f', bytes_val)[0]
    except Exception as e:
        log.error(f"Failed to convert registers to float: {e}")
        return None


class JsonConfig:
    """Manage configuration parameters with JSON persistence"""
    
    def __init__(self, config_file, config_params):
        """
        Initialize configuration manager
        
        Args:
            config_file: Path to JSON configuration file
            config_params: Dictionary with parameter definitions
                Format: {
                    'param_name': {
                        'register': int,     # Starting Modbus register
                        'count': int,        # Number of registers (2 for float)
                        'min': float,        # Minimum value
                        'max': float,        # Maximum value
                        'default': float     # Default value
                    }
                }
        """
        self.config_file = Path(config_file)
        self.config_params = config_params
        
        # Initialize parameters with defaults
        for param_name, param_info in self.config_params.items():
            setattr(self, param_name, param_info['default'])
        
        self.load()
    
    def load(self):
        """Load configuration from JSON file"""
        try:
            if self.config_file.exists():
                with open(self.config_file, 'r') as f:
                    data = json.load(f)
                    for param_name, param_info in self.config_params.items():
                        value = data.get(param_name, param_info['default'])
                        setattr(self, param_name, value)
                    log.info(f"Loaded config: " + ", ".join(
                        f"{name}={getattr(self, name)}" for name in self.config_params.keys()
                    ))
            else:
                log.info("Config file not found, using defaults")
                self.save()
        except Exception as e:
            log.error(f"Failed to load config: {e}, using defaults")
    
    def save(self):
        """Save configuration to JSON file"""
        try:
            self.config_file.parent.mkdir(parents=True, exist_ok=True)
            
            data = {
                param_name: getattr(self, param_name) for param_name in self.config_params.keys()
            }
            data['last_updated'] = datetime.now().isoformat()
            
            with open(self.config_file, 'w') as f:
                json.dump(data, f, indent=4)
            
            log.info(f"Saved config: " + ", ".join(
                f"{name}={getattr(self, name)}" for name in self.config_params.keys()
            ))
        except Exception as e:
            log.error(f"Failed to save config: {e}")
    
    def update(self, **kwargs):
        """Update parameters with validation"""
        changed = False
        
        for param_name, value in kwargs.items():
            if param_name not in self.config_params:
                log.warning(f"Unknown parameter: {param_name}")
                continue
            
            if value is None:
                continue
            
            param_info = self.config_params[param_name]
            if param_info['min'] <= value <= param_info['max']:
                setattr(self, param_name, value)
                changed = True
            else:
                log.warning(
                    f"Invalid {param_name} value: {value} "
                    f"(range: {param_info['min']} to {param_info['max']})"
                )
        
        if changed:
            self.save()
        
        return changed
    
    def get_register_map(self):
        """Get human-readable register map"""
        lines = []
        for param_name, param_info in self.config_params.items():
            reg_start = param_info['register']
            reg_end = reg_start + param_info['count'] - 1
            lines.append(
                f"  Reg {reg_start}-{reg_end}: {param_name} "
                f"(IEEE 754 float, range: {param_info['min']} to {param_info['max']})"
            )
        return "\n".join(lines)


class JsonSlaveContext(ModbusSlaveContext):
    """Custom Modbus Slave Context with automatic config persistence"""
    
    def __init__(self, config, *args, **kwargs):
        """
        Initialize custom slave context
        
        Args:
            config: JsonConfig instance
            *args, **kwargs: Arguments passed to ModbusSlaveContext
        """
        super().__init__(*args, **kwargs)
        self.config = config
        self._initialize_config_registers()
    
    def _initialize_config_registers(self):
        """Initialize configuration registers with current values"""
        if not self.store['h']:
            return
        
        for param_name, param_info in self.config.config_params.items():
            value = getattr(self.config, param_name)
            regs = float_to_modbus_regs(value)
            self.store['h'].setValues(param_info['register'], regs)
            log.debug(f"Initialized {param_name}={value} at register {param_info['register']}")
        
        log.info("All config registers initialized")
    
    def setValues(self, fx, address, values):
        """Override setValues to intercept config register writes"""
        # Call parent method first
        super().setValues(fx, address, values)
        
        # Check if config registers were modified (holding registers only)
        if fx == 3:
            self._check_config_update(address, values)
    
    def _check_config_update(self, address, values):
        """Check if any config parameter was modified"""
        for param_name, param_info in self.config.config_params.items():
            reg_start = param_info['register']
            reg_count = param_info['count']
            reg_end = reg_start + reg_count - 1
            
            # Check if write overlaps with this parameter's registers
            write_end = address + len(values) - 1
            if not (write_end < reg_start or address > reg_end):
                self._handle_config_update(param_name, param_info)
    
    def _handle_config_update(self, param_name, param_info):
        """Handle configuration parameter update"""
        try:
            # Read registers for this parameter
            regs = self.store['h'].getValues(
                param_info['register'],
                param_info['count']
            )
            
            # Convert to float
            value = modbus_regs_to_float(regs[0], regs[1])
            if value is None:
                raise ValueError(f"Failed to convert {param_name} registers to float")
            
            # Validate range
            if not (param_info['min'] <= value <= param_info['max']):
                log.warning(
                    f"Invalid {param_name} value: {value} "
                    f"(range: {param_info['min']} to {param_info['max']})"
                )
                self._restore_param(param_name, param_info)
                return
            
            # Update config
            update_dict = {param_name: value}
            if self.config.update(**update_dict):
                log.info(f"Config updated via Modbus: {param_name}={value}")
            else:
                log.warning(f"Config update failed for {param_name}")
                self._restore_param(param_name, param_info)
                
        except Exception as e:
            log.error(f"Error updating {param_name}: {e}")
            self._restore_param(param_name, param_info)
    
    def _restore_param(self, param_name, param_info):
        """Restore parameter to its saved value"""
        try:
            value = getattr(self.config, param_name)
            regs = float_to_modbus_regs(value)
            self.store['h'].setValues(param_info['register'], regs)
            log.info(f"Restored {param_name}={value}")
        except Exception as e:
            log.error(f"Failed to restore {param_name}: {e}")