""" faults.py Motore di simulazione dei guasti e delle anomalie per il simulatore di campo fotovoltaico. Il modulo permette di simulare eventi a diversi livelli: PVPanel | +-- degradation +-- soiling +-- partial_shading +-- intermittent_fault +-- permanent_fault | v CombinerBox | +-- string_loss +-- fuse_fault +-- communication_fault +-- complete_failure | v Inverter | +-- derating +-- overheating +-- shutdown +-- clipping +-- communication_fault Ogni fault produce un fattore di riduzione della produzione. Esempio: normal power = 1000 W soiling factor = 0.90 shading factor = 0.80 effective power = 1000 * 0.90 * 0.80 = 720 W Il modulo non modifica direttamente i componenti PVPanel, CombinerBox o Inverter. Fornisce invece un livello indipendente di simulazione dei fault che può essere interrogato durante ogni timestep. """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Dict, List, Optional import random # ====================================================================== # ENUMERAZIONE DEI TIPI DI FAULT # ====================================================================== class FaultType(str, Enum): """ Tipologie di guasto disponibili. """ # -------------------------------------------------------------- # Pannello # -------------------------------------------------------------- PANEL_DEGRADATION = ( "panel_degradation" ) SOILING = ( "soiling" ) PARTIAL_SHADING = ( "partial_shading" ) PANEL_INTERMITTENT = ( "panel_intermittent" ) PANEL_FAILURE = ( "panel_failure" ) # -------------------------------------------------------------- # Combiner # -------------------------------------------------------------- COMBINER_STRING_LOSS = ( "combiner_string_loss" ) COMBINER_FUSE_FAULT = ( "combiner_fuse_fault" ) COMBINER_COMMUNICATION = ( "combiner_communication" ) COMBINER_FAILURE = ( "combiner_failure" ) # -------------------------------------------------------------- # Inverter # -------------------------------------------------------------- INVERTER_DERATING = ( "inverter_derating" ) INVERTER_OVERHEATING = ( "inverter_overheating" ) INVERTER_SHUTDOWN = ( "inverter_shutdown" ) INVERTER_CLIPPING = ( "inverter_clipping" ) INVERTER_COMMUNICATION = ( "inverter_communication" ) # ====================================================================== # LIVELLO DEL COMPONENTE # ====================================================================== class ComponentLevel(str, Enum): """ Livello gerarchico del componente affetto. """ PANEL = "panel" COMBINER = "combiner" INVERTER = "inverter" # ====================================================================== # EVENTO DI FAULT # ====================================================================== @dataclass class FaultEvent: """ Rappresenta un singolo evento di guasto. Parameters ---------- fault_id: Identificativo univoco del fault. fault_type: Tipo di guasto. component_level: Livello gerarchico del componente. component_id: ID del componente interessato. start_time: Inizio del fault. end_time: Fine del fault. None = fault permanente. severity: Severità del fault da 0.0 a 1.0. enabled: Indica se il fault è attivo. description: Descrizione opzionale. """ fault_id: str fault_type: FaultType component_level: ComponentLevel component_id: str start_time: datetime end_time: Optional[datetime] = None severity: float = 1.0 enabled: bool = True description: str = "" # -------------------------------------------------------------- # Validazione # -------------------------------------------------------------- def __post_init__( self ) -> None: self.severity = min( max( self.severity, 0.0 ), 1.0 ) # -------------------------------------------------------------- # Stato # -------------------------------------------------------------- def is_active( self, timestamp: datetime ) -> bool: """ Verifica se il fault è attivo al timestamp indicato. """ if not self.enabled: return False if timestamp < self.start_time: return False if ( self.end_time is not None and timestamp >= self.end_time ): return False return True # -------------------------------------------------------------- # Fattore di riduzione # -------------------------------------------------------------- def reduction_factor( self ) -> float: """ Restituisce il fattore di produzione associato alla severità del fault. severity = 0.0 -> nessuna perdita severity = 1.0 -> perdita completa Returns ------- float Fattore compreso tra 0.0 e 1.0. """ return 1.0 - self.severity # ====================================================================== # CONFIGURAZIONE FAULT # ====================================================================== @dataclass class FaultConfig: """ Configurazione per la generazione automatica dei fault. """ enabled: bool = True panel_fault_probability: float = 0.0001 combiner_fault_probability: float = 0.00002 inverter_fault_probability: float = 0.00001 intermittent_probability: float = 0.30 default_fault_duration_hours: float = 4.0 random_seed: Optional[int] = None @classmethod def from_config( cls, config: Dict[str, object] ) -> FaultConfig: """ Crea un'istanza di FaultConfig a partire da un dizionario di configurazione. """ fault_config = config["faults"] return cls( enabled= fault_config["enabled"], panel_fault_probability= fault_config["panel"]["failure_probability_daily"], combiner_fault_probability= fault_config["combiner"]["failure_probability_daily"], inverter_fault_probability= fault_config["inverter"]["failure_probability_daily"], intermittent_probability= fault_config.get( "intermittent_probability", 0.30 ), default_fault_duration_hours= fault_config.get( "default_fault_duration_hours", 4.0 ), random_seed= fault_config.get( "random_seed", None ) ) @classmethod def from_json( cls, json_path: str ) -> FaultConfig: """ Crea un'istanza di FaultConfig a partire da un file JSON. """ import json with open( json_path, "r" ) as f: config = json.load( f ) return cls.from_config( config ) # ====================================================================== # FAULT MANAGER # ====================================================================== @dataclass class FaultManager: """ Gestore centrale dei fault dell'impianto. Mantiene tutti gli eventi attivi e storici. Il FaultManager può: - aggiungere fault manualmente; - rimuovere fault; - verificare i fault attivi; - calcolare il fattore di produzione; - generare fault casuali; - fornire informazioni diagnostiche. """ config: FaultConfig = field( default_factory=FaultConfig ) faults: List[FaultEvent] = field( default_factory=list ) _fault_counter: int = 0 # ------------------------------------------------------------------ # Inizializzazione # ------------------------------------------------------------------ def __post_init__( self ) -> None: if ( self.config.random_seed is not None ): random.seed( self.config.random_seed ) # ------------------------------------------------------------------ # Generazione ID # ------------------------------------------------------------------ def _generate_fault_id( self ) -> str: self._fault_counter += 1 return ( f"FAULT_" f"{self._fault_counter:06d}" ) # ------------------------------------------------------------------ # Aggiunta fault # ------------------------------------------------------------------ def add_fault( self, fault_type: FaultType, component_level: ComponentLevel, component_id: str, start_time: datetime, severity: float = 1.0, duration_hours: Optional[float] = None, description: str = "" ) -> FaultEvent: """ Crea e registra un nuovo fault. """ end_time = None if duration_hours is not None: from datetime import timedelta end_time = ( start_time + timedelta( hours= duration_hours ) ) fault = FaultEvent( fault_id= self._generate_fault_id(), fault_type= fault_type, component_level= component_level, component_id= component_id, start_time= start_time, end_time= end_time, severity= severity, description= description ) self.faults.append( fault ) return fault # ------------------------------------------------------------------ # Rimozione fault # ------------------------------------------------------------------ def remove_fault( self, fault_id: str ) -> bool: """ Rimuove un fault tramite ID. """ for fault in self.faults: if ( fault.fault_id == fault_id ): self.faults.remove( fault ) return True return False # ------------------------------------------------------------------ # Disabilitazione fault # ------------------------------------------------------------------ def disable_fault( self, fault_id: str ) -> bool: """ Disabilita un fault senza eliminarlo dallo storico. """ for fault in self.faults: if ( fault.fault_id == fault_id ): fault.enabled = False return True return False # ------------------------------------------------------------------ # Fault attivi # ------------------------------------------------------------------ def get_active_faults( self, timestamp: datetime ) -> List[FaultEvent]: """ Restituisce tutti i fault attivi. """ return [ fault for fault in self.faults if fault.is_active( timestamp ) ] # ------------------------------------------------------------------ # Fault di un componente # ------------------------------------------------------------------ def get_component_faults( self, component_id: str, timestamp: datetime ) -> List[FaultEvent]: """ Restituisce i fault attivi associati a uno specifico componente. """ return [ fault for fault in self.faults if ( fault.component_id == component_id and fault.is_active( timestamp ) ) ] # ------------------------------------------------------------------ # Fattore componente # ------------------------------------------------------------------ def get_component_factor( self, component_id: str, timestamp: datetime ) -> float: """ Calcola il fattore di produzione complessivo di un componente. Se sono presenti più fault contemporaneamente, i fattori vengono moltiplicati. Esempio: Soiling = 0.90 Shading = 0.80 Factor = 0.90 * 0.80 = 0.72 """ active_faults = ( self.get_component_faults( component_id, timestamp ) ) factor = 1.0 for fault in active_faults: factor *= ( fault.reduction_factor() ) return min( max( factor, 0.0 ), 1.0 ) # ------------------------------------------------------------------ # Stato componente # ------------------------------------------------------------------ def get_component_status( self, component_id: str, timestamp: datetime ) -> Dict[str, object]: """ Restituisce lo stato diagnostico di un componente. """ faults = ( self.get_component_faults( component_id, timestamp ) ) factor = ( self.get_component_factor( component_id, timestamp ) ) return { "component_id": component_id, "healthy": len( faults ) == 0, "fault_count": len( faults ), "production_factor": factor, "faults": faults } # ------------------------------------------------------------------ # Generazione fault casuale # ------------------------------------------------------------------ @staticmethod def _daily_to_timestep_probability( daily_probability: float, timestep_minutes: float ) -> float: """ Converte una probabilità di fault giornaliera nella probabilità equivalente per un singolo timestep. Le probabilità configurate (*_probability_daily) sono riferite a un'intera giornata, ma generate_random_fault viene chiamato una volta per timestep: senza questa conversione la probabilità giornaliera verrebbe applicata una volta per ogni timestep, inflazionando il tasso di fault di un fattore pari al numero di timestep al giorno (ad es. 288x per timestep da 5 minuti). """ timesteps_per_day = ( 24 * 60 ) / timestep_minutes return 1 - ( 1 - daily_probability ) ** ( 1 / timesteps_per_day ) def generate_random_fault( self, component_id: str, component_level: ComponentLevel, timestamp: datetime, timestep_minutes: float = 5.0 ) -> Optional[FaultEvent]: """ Genera casualmente un fault. Il metodo è pensato per essere chiamato durante ogni timestep della simulazione. La probabilità viene scelta in base al livello gerarchico, e convertita da probabilità giornaliera a probabilità per il singolo timestep. """ if not self.config.enabled: return None if ( component_level == ComponentLevel.PANEL ): daily_probability = ( self.config .panel_fault_probability ) elif ( component_level == ComponentLevel.COMBINER ): daily_probability = ( self.config .combiner_fault_probability ) else: daily_probability = ( self.config .inverter_fault_probability ) probability = ( self._daily_to_timestep_probability( daily_probability, timestep_minutes ) ) # Nessun fault if ( random.random() >= probability ): return None # -------------------------------------------------------------- # Selezione tipologia # -------------------------------------------------------------- if ( component_level == ComponentLevel.PANEL ): fault_type = random.choice([ FaultType.SOILING, FaultType.PARTIAL_SHADING, FaultType.PANEL_INTERMITTENT, FaultType.PANEL_FAILURE ]) elif ( component_level == ComponentLevel.COMBINER ): fault_type = random.choice([ FaultType.COMBINER_STRING_LOSS, FaultType.COMBINER_FUSE_FAULT, FaultType.COMBINER_COMMUNICATION, FaultType.COMBINER_FAILURE ]) else: fault_type = random.choice([ FaultType.INVERTER_DERATING, FaultType.INVERTER_OVERHEATING, FaultType.INVERTER_SHUTDOWN, FaultType.INVERTER_CLIPPING, FaultType.INVERTER_COMMUNICATION ]) # -------------------------------------------------------------- # Intermittent fault # -------------------------------------------------------------- is_intermittent = ( random.random() < self.config .intermittent_probability ) if is_intermittent: duration = ( random.uniform( 0.5, self.config .default_fault_duration_hours ) ) else: duration = None # -------------------------------------------------------------- # Severità # -------------------------------------------------------------- if ( fault_type in [ FaultType.PANEL_FAILURE, FaultType.COMBINER_FAILURE, FaultType.INVERTER_SHUTDOWN ] ): severity = 1.0 elif ( fault_type == FaultType.SOILING ): severity = random.uniform( 0.05, 0.30 ) elif ( fault_type == FaultType.PARTIAL_SHADING ): severity = random.uniform( 0.10, 0.70 ) elif ( fault_type == FaultType.INVERTER_DERATING ): severity = random.uniform( 0.10, 0.50 ) else: severity = random.uniform( 0.10, 0.40 ) return self.add_fault( fault_type= fault_type, component_level= component_level, component_id= component_id, start_time= timestamp, severity= severity, duration_hours= duration, description= "Automatically generated fault" ) # ------------------------------------------------------------------ # Simulazione automatica # ------------------------------------------------------------------ def simulate_random_faults( self, timestamp: datetime, panel_ids: List[str], combiner_ids: List[str], inverter_ids: List[str], timestep_minutes: float = 5.0 ) -> List[FaultEvent]: """ Esegue la generazione casuale dei fault per tutti i componenti dell'impianto. Returns ------- List[FaultEvent] Fault generati durante il timestep. """ generated = [] for panel_id in panel_ids: fault = ( self.generate_random_fault( component_id= panel_id, component_level= ComponentLevel.PANEL, timestamp= timestamp, timestep_minutes= timestep_minutes ) ) if fault is not None: generated.append( fault ) for combiner_id in combiner_ids: fault = ( self.generate_random_fault( component_id= combiner_id, component_level= ComponentLevel.COMBINER, timestamp= timestamp, timestep_minutes= timestep_minutes ) ) if fault is not None: generated.append( fault ) for inverter_id in inverter_ids: fault = ( self.generate_random_fault( component_id= inverter_id, component_level= ComponentLevel.INVERTER, timestamp= timestamp, timestep_minutes= timestep_minutes ) ) if fault is not None: generated.append( fault ) return generated # ------------------------------------------------------------------ # Diagnostica # ------------------------------------------------------------------ def diagnostics( self, timestamp: datetime ) -> Dict[str, object]: """ Restituisce una panoramica diagnostica dei fault attivi. """ active = ( self.get_active_faults( timestamp ) ) by_level = { "panel": 0, "combiner": 0, "inverter": 0 } by_type = {} for fault in active: by_level[ fault.component_level.value ] += 1 fault_name = ( fault.fault_type.value ) by_type[ fault_name ] = ( by_type.get( fault_name, 0 ) + 1 ) return { "timestamp": timestamp, "active_faults": len( active ), "by_level": by_level, "by_type": by_type, "faults": active } # ------------------------------------------------------------------ # Pulizia fault scaduti # ------------------------------------------------------------------ def cleanup_expired_faults( self, timestamp: datetime ) -> int: """ Rimuove i fault terminati. Returns ------- int Numero di fault rimossi. """ before = len( self.faults ) self.faults = [ fault for fault in self.faults if ( fault.end_time is None or fault.end_time > timestamp ) ] return ( before - len( self.faults ) ) # ------------------------------------------------------------------ # Rappresentazione # ------------------------------------------------------------------ def __repr__( self ) -> str: """ Rappresentazione leggibile. """ active = sum( 1 for fault in self.faults if fault.enabled ) return ( f"FaultManager(" f"total_faults=" f"{len(self.faults)}, " f"active=" f"{active})" )