control/utils.py - master - cosmicfuturetoday/PSO

control/utils.py
459 строк18 KB
Vladimir Goncharov
create control/utils.py
12 янв 2026, 17:26
12 янв 2026, 17:26ba86318
"""
Вспомогательные функции для системы управления
"""

import numpy as np
from typing import Dict, List, Tuple, Optional, Any
import warnings
from dataclasses import dataclass
from enum import Enum


class ControlStrategy(Enum):
    """Стратегии управления"""
    CONSERVATIVE = "conservative"      # Минимальное вмешательство
    BALANCED = "balanced"              # Баланс между эффективностью и стабильностью
    AGGRESSIVE = "aggressive"          # Максимизация производительности
    SAFETY_FIRST = "safety_first"      # Приоритет безопасности


@dataclass
class ControlLimits:
    """Ограничения системы управления"""
    
    # Ограничения по мощности
    max_total_power: float = 10e6  # [W]
    min_total_power: float = 0.1e6  # [W]
    max_power_rate: float = 1e6  # Максимальное изменение мощности в секунду [W/s]
    
    # Ограничения по току
    max_plasma_current: float = 2e6  # [A]
    min_plasma_current: float = 0.1e6  # [A]
    
    # Ограничения по плотности
    max_density: float = 1e20  # [m^-3]
    min_density: float = 1e18  # [m^-3]
    
    # Ограничения по положению
    max_vertical_displacement: float = 0.1  # [m]
    max_radial_displacement: float = 0.05  # [m]
    
    # Временные ограничения
    min_response_time: float = 0.001  # [s]
    max_decision_time: float = 0.01  # [s]


def validate_control_commands(commands: List[Dict], 
                            limits: ControlLimits,
                            current_state: Dict) -> Tuple[List[Dict], List[str]]:
    """
    Валидация команд управления
    
    Args:
        commands: Список команд для проверки
        limits: Ограничения системы
        current_state: Текущее состояние системы
        
    Returns:
        (валидные_команды, список_ошибок)
    """
    valid_commands = []
    errors = []
    
    current_power = current_state.get('total_power', 0)
    
    for cmd in commands:
        cmd_valid = True
        cmd_errors = []
        
        # Проверка типа команды
        cmd_type = cmd.get('type', '')
        if not cmd_type:
            cmd_errors.append("Отсутствует тип команды")
            cmd_valid = False
        
        # Проверка параметров
        value = cmd.get('value', 0)
        
        if 'power' in cmd_type.lower():
            # Проверка ограничений по мощности
            if value < limits.min_total_power:
                cmd_errors.append(f"Мощность {value:.1e} ниже минимальной {limits.min_total_power:.1e}")
                cmd_valid = False
            elif value > limits.max_total_power:
                cmd_errors.append(f"Мощность {value:.1e} превышает максимальную {limits.max_total_power:.1e}")
                cmd_valid = False
            
            # Проверка скорости изменения мощности
            power_change = abs(value - current_power)
            if power_change > limits.max_power_rate:
                cmd_errors.append(f"Изменение мощности {power_change:.1e} превышает допустимое {limits.max_power_rate:.1e}")
                cmd_valid = False
        
        elif 'current' in cmd_type.lower():
            # Проверка ограничений по току
            if value < limits.min_plasma_current:
                cmd_errors.append(f"Ток {value:.1e} ниже минимального {limits.min_plasma_current:.1e}")
                cmd_valid = False
            elif value > limits.max_plasma_current:
                cmd_errors.append(f"Ток {value:.1e} превышает максимальный {limits.max_plasma_current:.1e}")
                cmd_valid = False
        
        elif 'density' in cmd_type.lower() or 'fueling' in cmd_type.lower():
            # Проверка ограничений по плотности
            if value < limits.min_density:
                cmd_errors.append(f"Плотность {value:.1e} ниже минимальной {limits.min_density:.1e}")
                cmd_valid = False
            elif value > limits.max_density:
                cmd_errors.append(f"Плотность {value:.1e} превышает максимальную {limits.max_density:.1e}")
                cmd_valid = False
        
        if cmd_valid:
            valid_commands.append(cmd)
        else:
            errors.extend([f"{cmd_type}: {err}" for err in cmd_errors])
    
    return valid_commands, errors


def prioritize_commands(commands: List[Dict], 
                       strategy: ControlStrategy = ControlStrategy.BALANCED) -> List[Dict]:
    """
    Приоритизация команд управления
    
    Args:
        commands: Список команд
        strategy: Стратегия приоритизации
        
    Returns:
        Отсортированный список команд по приоритету
    """
    if not commands:
        return []
    
    # Присваиваем приоритеты в зависимости от стратегии
    prioritized = []
    
    for cmd in commands:
        cmd_type = cmd.get('type', '')
        priority = 0
        
        # Определение приоритета в зависимости от типа команды и стратегии
        if strategy == ControlStrategy.SAFETY_FIRST:
            # Приоритет безопасности
            if 'emergency' in cmd_type.lower():
                priority = 100
            elif 'risk' in cmd_type.lower() or 'stabil' in cmd_type.lower():
                priority = 80
            elif 'optimize' in cmd_type.lower():
                priority = 20
            else:
                priority = 50
                
        elif strategy == ControlStrategy.CONSERVATIVE:
            # Консервативный подход
            if 'maintain' in cmd_type.lower():
                priority = 90
            elif 'adjust' in cmd_type.lower():
                priority = 60
            elif 'set' in cmd_type.lower():
                priority = 40
            else:
                priority = 30
                
        elif strategy == ControlStrategy.AGGRESSIVE:
            # Агрессивный подход
            if 'optimize' in cmd_type.lower() or 'improve' in cmd_type.lower():
                priority = 100
            elif 'increase' in cmd_type.lower():
                priority = 80
            else:
                priority = 40
                
        else:  # BALANCED
            # Сбалансированный подход
            if 'emergency' in cmd_type.lower():
                priority = 100
            elif 'optimize' in cmd_type.lower():
                priority = 70
            elif 'maintain' in cmd_type.lower():
                priority = 60
            elif 'adjust' in cmd_type.lower():
                priority = 50
            else:
                priority = 40
        
        prioritized.append((priority, cmd))
    
    # Сортировка по приоритету (по убыванию)
    prioritized.sort(key=lambda x: x[0], reverse=True)
    
    # Возвращаем только команды
    return [cmd for _, cmd in prioritized]


def merge_similar_commands(commands: List[Dict]) -> List[Dict]:
    """
    Объединение похожих команд управления
    
    Args:
        commands: Список команд
        
    Returns:
        Объединённый список команд
    """
    if not commands:
        return []
    
    # Группируем команды по типу и актуатору
    command_groups = {}
    
    for cmd in commands:
        cmd_type = cmd.get('type', 'unknown')
        actuator = cmd.get('actuator', 'general')
        
        key = (cmd_type, actuator)
        
        if key not in command_groups:
            command_groups[key] = []
        
        command_groups[key].append(cmd)
    
    # Объединяем команды в группах
    merged_commands = []
    
    for (cmd_type, actuator), group in command_groups.items():
        if len(group) == 1:
            # Одна команда - оставляем как есть
            merged_commands.append(group[0])
        else:
            # Несколько команд - объединяем
            merged_cmd = group[0].copy()
            
            # Для числовых значений берём среднее
            values = [c.get('value', 0) for c in group if 'value' in c]
            if values:
                merged_cmd['value'] = np.mean(values)
            
            # Объединяем причины
            reasons = [c.get('reason', '') for c in group if 'reason' in c]
            if reasons:
                merged_cmd['reason'] = '; '.join(set(reasons))
            
            merged_cmd['merged_count'] = len(group)
            merged_commands.append(merged_cmd)
    
    return merged_commands


def calculate_control_efficiency(initial_state: Dict,
                               final_state: Dict,
                               commands_executed: List[Dict],
                               time_elapsed: float) -> Dict:
    """
    Расчёт эффективности управления
    
    Args:
        initial_state: Начальное состояние
        final_state: Конечное состояние
        commands_executed: Выполненные команды
        time_elapsed: Затраченное время
        
    Returns:
        Метрики эффективности
    """
    metrics = {}
    
    # 1. Эффективность по самоорганизации
    if 'xi_p' in initial_state and 'xi_p' in final_state:
        xi_improvement = initial_state['xi_p'] - final_state['xi_p']
        metrics['xi_improvement'] = xi_improvement
        metrics['xi_improvement_rate'] = xi_improvement / (time_elapsed + 1e-10)
    
    # 2. Эффективность по энергии
    if 'total_energy' in initial_state and 'total_energy' in final_state:
        energy_change = final_state['total_energy'] - initial_state['total_energy']
        total_power_used = sum(cmd.get('value', 0) for cmd in commands_executed 
                             if 'power' in cmd.get('type', '').lower())
        
        if total_power_used > 0:
            energy_efficiency = energy_change / (total_power_used * time_elapsed + 1e-10)
            metrics['energy_efficiency'] = energy_efficiency
    
    # 3. Эффективность по количеству команд
    metrics['commands_per_second'] = len(commands_executed) / (time_elapsed + 1e-10)
    
    # 4. Эффективность по стабильности
    if 'risk_score' in initial_state and 'risk_score' in final_state:
        risk_reduction = initial_state['risk_score'] - final_state['risk_score']
        metrics['risk_reduction'] = risk_reduction
    
    # 5. Общая оценка эффективности
    # Взвешенная сумма отдельных метрик
    weights = {
        'xi_improvement': 0.4,
        'energy_efficiency': 0.3,
        'risk_reduction': 0.3
    }
    
    total_score = 0
    for metric, weight in weights.items():
        if metric in metrics:
            # Нормировка метрики
            if metric == 'xi_improvement':
                norm_value = min(max(metrics[metric] * 10, 0), 1)
            elif metric == 'energy_efficiency':
                norm_value = min(max(metrics[metric], 0), 1)
            elif metric == 'risk_reduction':
                norm_value = min(max(metrics[metric], 0), 1)
            else:
                norm_value = 0
            
            total_score += weight * norm_value
    
    metrics['overall_efficiency'] = total_score
    
    return metrics


def generate_control_report(controller_stats: Dict,
                          plasma_stats: Dict,
                          time_period: Tuple[float, float]) -> str:
    """
    Генерация отчёта о работе системы управления
    
    Args:
        controller_stats: Статистика контроллера
        plasma_stats: Статистика плазмы
        time_period: Период времени (начало, конец)
        
    Returns:
        Текстовый отчёт
    """
    t_start, t_end = time_period
    
    report = f"""
ОТЧЁТ О РАБОТЕ СИСТЕМЫ УПРАВЛЕНИЯ
==================================
Период: {t_start:.3f} - {t_end:.3f} с (длительность: {t_end - t_start:.3f} с)

СТАТИСТИКА КОНТРОЛЛЕРА:
----------------------
• Всего циклов управления: {controller_stats.get('total_cycles', 0)}
• Средний риск: {controller_stats.get('avg_risk', 0):.3f}
• Максимальный риск: {controller_stats.get('max_risk', 0):.3f}
• Среднее количество команд за цикл: {controller_stats.get('avg_commands_per_cycle', 0):.1f}
• Количество предупреждений: {controller_stats.get('warnings_count', 0)}
• Размер очереди данных: {controller_stats.get('data_queue_size', 0)}
• Размер очереди команд: {controller_stats.get('command_queue_size', 0)}

СТАТИСТИКА ПЛАЗМЫ:
-----------------
• Среднее давление: {plasma_stats.get('avg_pressure', 0):.2e} Па
• Средняя плотность: {plasma_stats.get('avg_density', 0):.2e} м⁻³
• Средняя температура электронов: {plasma_stats.get('avg_Te', 0):.1f} эВ
• Средняя температура ионов: {plasma_stats.get('avg_Ti', 0):.1f} эВ
• Минимальный q: {plasma_stats.get('min_q', 0):.2f}
• Максимальный β: {plasma_stats.get('max_beta', 0):.3f}

ЭФФЕКТИВНОСТЬ УПРАВЛЕНИЯ:
-----------------------
• Улучшение самоорганизации (ξ): {plasma_stats.get('xi_improvement', 0):.3f}
• Изменение энергии: {plasma_stats.get('energy_change', 0):.2e} Дж
• Снижение риска: {plasma_stats.get('risk_reduction', 0):.3f}
• Общая эффективность: {plasma_stats.get('overall_efficiency', 0):.1%}

РЕКОМЕНДАЦИИ:
------------
"""
    
    # Добавляем рекомендации на основе статистики
    recommendations = []
    
    if controller_stats.get('avg_risk', 0) > 0.5:
        recommendations.append("Высокий средний риск. Рекомендуется перейти в режим SAFETY_FIRST.")
    
    if controller_stats.get('warnings_count', 0) > 10:
        recommendations.append("Большое количество предупреждений. Проверить датчики и алгоритмы.")
    
    if plasma_stats.get('xi_improvement', 0) < 0:
        recommendations.append("Самоорганизация ухудшилась. Пересмотреть стратегию управления.")
    
    if not recommendations:
        recommendations.append("Система работает нормально. Продолжить текущую стратегию.")
    
    for i, rec in enumerate(recommendations, 1):
        report += f"{i}. {rec}\n"
    
    report += "\n" + "=" * 50 + "\n"
    
    return report


# Пример использования вспомогательных функций
def example_utils_usage():
    """Пример использования утилит управления"""
    
    # Создаём тестовые команды
    commands = [
        {
            'type': 'set_power',
            'actuator': 'ECRH_1',
            'value': 1.5e6,
            'reason': 'Increase heating'
        },
        {
            'type': 'set_power',
            'actuator': 'ECRH_2',
            'value': 1.2e6,
            'reason': 'Balance heating'
        },
        {
            'type': 'emergency_shutdown',
            'actuator': 'all',
            'value': 0.0,
            'reason': 'High risk detected'
        }
    ]
    
    # Валидация
    limits = ControlLimits(max_total_power=2e6)
    current_state = {'total_power': 1e6}
    
    valid_commands, errors = validate_control_commands(commands, limits, current_state)
    
    print("Валидация команд:")
    print(f"  Валидных: {len(valid_commands)}")
    print(f"  Ошибок: {len(errors)}")
    for err in errors:
        print(f"    - {err}")
    
    # Приоритизация
    prioritized = prioritize_commands(valid_commands, ControlStrategy.SAFETY_FIRST)
    
    print("\nПриоритизация (SAFETY_FIRST):")
    for cmd in prioritized:
        print(f"  {cmd['type']}: {cmd.get('reason', '')}")
    
    # Объединение
    merged = merge_similar_commands(prioritized)
    
    print("\nОбъединённые команды:")
    for cmd in merged:
        print(f"  {cmd['type']} (merged: {cmd.get('merged_count', 1)}): {cmd.get('reason', '')}")
    
    # Расчёт эффективности
    initial = {'xi_p': 0.15, 'total_energy': 1e6, 'risk_score': 0.6}
    final = {'xi_p': 0.10, 'total_energy': 1.2e6, 'risk_score': 0.4}
    
    efficiency = calculate_control_efficiency(
        initial, final, valid_commands, time_elapsed=1.0
    )
    
    print("\nЭффективность управления:")
    for metric, value in efficiency.items():
        print(f"  {metric}: {value:.3f}")


if __name__ == "__main__":
    example_utils_usage()