""" pvsim/simulator.py Motore principale della simulazione fotovoltaica. Gerarchia: PVPlant | +-- Inverter | | | +-- CombinerBox | | | +-- PVPanel | v SimulationResult | +-- SimulationStep | +-- WeatherData +-- PanelData[] +-- CombinerData[] +-- InverterData[] +-- PlantData +-- FaultData[] Responsabilità: simulator.py -> simula statistics.py -> analizza exporter.py -> esporta """ from __future__ import annotations from dataclasses import dataclass, field from datetime import datetime, timedelta from typing import Any, Dict, List, Optional import pandas as pd from .models import ( WeatherData, PanelData, CombinerData, InverterData, PlantData, FaultData, SimulationStep, ) from .plant import PVPlant from .sun import SunModel from .weather import WeatherModel from .faults import FaultManager # ====================================================================== # CONFIGURATION # ====================================================================== @dataclass class SimulationConfig: """ Configurazione temporale della simulazione. """ start: datetime end: datetime timestep_minutes: int = 5 generate_random_faults: bool = False store_panel_data: bool = True store_combiner_data: bool = True store_inverter_data: bool = True store_plant_data: bool = True store_weather_data: bool = True store_fault_data: bool = True def __post_init__( self, ) -> None: if self.end <= self.start: raise ValueError( "end deve essere " "maggiore di start" ) if self.timestep_minutes <= 0: raise ValueError( "timestep_minutes deve " "essere maggiore di zero" ) # ====================================================================== # RESULT # ====================================================================== @dataclass class SimulationResult: """ Risultato completo della simulazione. """ steps: List[ SimulationStep ] = field( default_factory=list ) # ------------------------------------------------------------------ @property def panel_records( self, ) -> List[PanelData]: records = [] for step in self.steps: records.extend( step.panels ) return records # ------------------------------------------------------------------ @property def combiner_records( self, ) -> List[CombinerData]: records = [] for step in self.steps: records.extend( step.combiners ) return records # ------------------------------------------------------------------ @property def inverter_records( self, ) -> List[InverterData]: records = [] for step in self.steps: records.extend( step.inverters ) return records # ------------------------------------------------------------------ @property def plant_records( self, ) -> List[PlantData]: records = [] for step in self.steps: if step.plant is not None: records.append( step.plant ) return records # ------------------------------------------------------------------ @property def weather_records( self, ) -> List[WeatherData]: records = [] for step in self.steps: if step.weather is not None: records.append( step.weather ) return records # ------------------------------------------------------------------ @property def fault_records( self, ) -> List[FaultData]: records = [] for step in self.steps: records.extend( step.faults ) return records # ================================================================== # DATAFRAME # ================================================================== @staticmethod def _to_dataframe( records: List[Any], ) -> pd.DataFrame: if not records: return pd.DataFrame() return pd.DataFrame( [ record.to_dict() for record in records ] ) # ------------------------------------------------------------------ def panel_dataframe( self, ) -> pd.DataFrame: return self._to_dataframe( self.panel_records ) # ------------------------------------------------------------------ def combiner_dataframe( self, ) -> pd.DataFrame: return self._to_dataframe( self.combiner_records ) # ------------------------------------------------------------------ def inverter_dataframe( self, ) -> pd.DataFrame: return self._to_dataframe( self.inverter_records ) # ------------------------------------------------------------------ def plant_dataframe( self, ) -> pd.DataFrame: return self._to_dataframe( self.plant_records ) # ------------------------------------------------------------------ def weather_dataframe( self, ) -> pd.DataFrame: return self._to_dataframe( self.weather_records ) # ------------------------------------------------------------------ def faults_dataframe( self, ) -> pd.DataFrame: return self._to_dataframe( self.fault_records ) # ================================================================== # SUMMARY # ================================================================== def summary( self, ) -> Dict[str, Any]: return { "timesteps": len( self.steps ), "weather_records": len( self.weather_records ), "panel_records": len( self.panel_records ), "combiner_records": len( self.combiner_records ), "inverter_records": len( self.inverter_records ), "plant_records": len( self.plant_records ), "fault_records": len( self.fault_records ), } # ====================================================================== # SIMULATOR # ====================================================================== class PVSimulator: """ Motore centrale di simulazione. """ def __init__( self, plant: PVPlant, sun: SunModel, weather: WeatherModel, fault_manager: FaultManager, ) -> None: self.plant = plant self.sun = sun self.weather = weather self.fault_manager = fault_manager # ================================================================== # WEATHER # ================================================================== def _build_weather_data( self, timestamp: datetime, ) -> WeatherData: conditions = ( self.weather.get_conditions( timestamp ) ) return WeatherData( timestamp= timestamp, ambient_temperature_C= conditions[ "ambient_temperature_C" ], solar_elevation_deg= conditions[ "solar_elevation_deg" ], solar_azimuth_deg= conditions[ "solar_azimuth_deg" ], solar_zenith_deg= conditions[ "solar_zenith_deg" ], angle_of_incidence_deg= conditions[ "angle_of_incidence_deg" ], ghi_Wm2= conditions[ "ghi_Wm2" ], dni_Wm2= conditions[ "dni_Wm2" ], dhi_Wm2= conditions[ "dhi_Wm2" ], poa_global_Wm2= conditions[ "poa_global_Wm2" ], poa_direct_Wm2= conditions[ "poa_direct_Wm2" ], poa_diffuse_Wm2= conditions[ "poa_diffuse_Wm2" ], cloud_factor= conditions[ "cloud_factor" ], rain_factor= conditions[ "rain_factor" ], rain_active= conditions[ "rain_active" ], daylight= conditions[ "daylight" ], ) # ================================================================== # COMPONENT IDS # ================================================================== def _component_ids( self, ) -> Dict[ str, List[str] ]: panel_ids = [] combiner_ids = [] inverter_ids = [] for inverter in self.plant.inverters: inverter_ids.append( inverter.inverter_id ) for combiner in inverter.combiners: combiner_ids.append( combiner.combiner_id ) for panel in combiner.panels: panel_ids.append( panel.panel_id ) return { "panel": panel_ids, "combiner": combiner_ids, "inverter": inverter_ids, } # ================================================================== # RANDOM FAULTS # ================================================================== def _generate_faults( self, timestamp: datetime, ) -> None: ids = self._component_ids() self.fault_manager.simulate_random_faults( timestamp= timestamp, panel_ids= ids[ "panel" ], combiner_ids= ids[ "combiner" ], inverter_ids= ids[ "inverter" ], ) # ================================================================== # PANEL # ================================================================== def _simulate_panels( self, timestamp: datetime, weather: WeatherData, ) -> List[PanelData]: records = [] for inverter in self.plant.inverters: for combiner in inverter.combiners: for panel in combiner.panels: fault_factor = ( self.fault_manager .get_component_factor( panel.panel_id, timestamp ) ) try: result = ( panel.simulate( irradiance= weather .poa_global_Wm2, ambient_temperature= weather .ambient_temperature_C, timestamp= timestamp ) ) dc_power = ( result[ "dc_power_W" ] ) panel_temperature = ( result.get( "panel_temperature_C", weather .ambient_temperature_C ) ) except ( AttributeError, TypeError, KeyError, ): dc_power = ( panel.nominal_power * ( weather .poa_global_Wm2 / 1000.0 ) ) panel_temperature = ( weather .ambient_temperature_C ) dc_power *= fault_factor record = PanelData( timestamp= timestamp, panel_id= panel.panel_id, combiner_id= combiner.combiner_id, inverter_id= inverter.inverter_id, dc_power_W= dc_power, nominal_power_W= panel.nominal_power, irradiance_Wm2= weather .poa_global_Wm2, panel_temperature_C= panel_temperature, fault_factor= fault_factor, enabled= panel.enabled, ) records.append( record ) return records # ================================================================== # COMBINER # ================================================================== def _simulate_combiners( self, timestamp: datetime, panels: List[PanelData], ) -> List[CombinerData]: records = [] for inverter in self.plant.inverters: for combiner in inverter.combiners: component_panels = [ panel for panel in panels if ( panel.combiner_id == combiner.combiner_id ) ] if not component_panels: continue raw_power = sum( panel.dc_power_W for panel in component_panels ) nominal_power = sum( panel.nominal_power_W for panel in component_panels ) active_panels = sum( 1 for panel in component_panels if panel.enabled and panel.dc_power_W > 0 ) fault_factor = ( self.fault_manager .get_component_factor( combiner.combiner_id, timestamp ) ) dc_power = ( raw_power * fault_factor ) dc_voltage = 400.0 dc_current = ( dc_power / dc_voltage if dc_voltage > 0 else 0.0 ) fault_active = ( fault_factor < 1.0 ) if fault_active: status = "fault" elif active_panels == 0: status = "idle" else: status = "normal" availability = ( active_panels / len( component_panels ) ) records.append( CombinerData( timestamp= timestamp, combiner_id= combiner.combiner_id, inverter_id= inverter.inverter_id, dc_power_W= dc_power, dc_voltage_V= dc_voltage, dc_current_A= dc_current, nominal_power_W= nominal_power, panel_count= len( component_panels ), active_panel_count= active_panels, fault_factor= fault_factor, fault_active= fault_active, availability= availability, status= status, ) ) return records # ================================================================== # INVERTER # ================================================================== def _simulate_inverters( self, timestamp: datetime, combiners: List[CombinerData], weather: WeatherData, ) -> List[InverterData]: records = [] for inverter in self.plant.inverters: component_combiners = [ combiner for combiner in combiners if ( combiner.inverter_id == inverter.inverter_id ) ] if not component_combiners: continue dc_power = sum( combiner.dc_power_W for combiner in component_combiners ) nominal_power = sum( combiner.nominal_power_W for combiner in component_combiners ) fault_factor = ( self.fault_manager .get_component_factor( inverter.inverter_id, timestamp ) ) dc_power *= fault_factor efficiency = 0.97 ac_power = ( dc_power * efficiency ) inverter_nominal = ( inverter.nominal_power_kW * 1000.0 ) clipping_loss = 0.0 if ac_power > inverter_nominal: clipping_loss = ( ac_power - inverter_nominal ) ac_power = ( inverter_nominal ) dc_voltage = 800.0 dc_current = ( dc_power / dc_voltage if dc_voltage > 0 else 0.0 ) ac_voltage = 400.0 ac_current = ( ac_power / ac_voltage if ac_voltage > 0 else 0.0 ) temperature = ( weather.ambient_temperature_C + ( ac_power / max( inverter_nominal, 1.0 ) ) * 20.0 ) active_combiners = sum( 1 for combiner in component_combiners if combiner.status != "fault" ) fault_active = ( fault_factor < 1.0 ) if fault_active: status = "fault" elif ac_power <= 0: status = "idle" else: status = "normal" availability = ( active_combiners / len( component_combiners ) ) records.append( InverterData( timestamp= timestamp, inverter_id= inverter.inverter_id, dc_power_W= dc_power, ac_power_W= ac_power, dc_voltage_V= dc_voltage, dc_current_A= dc_current, ac_voltage_V= ac_voltage, ac_current_A= ac_current, efficiency= efficiency, nominal_power_W= inverter_nominal, clipping_loss_W= clipping_loss, temperature_C= temperature, fault_factor= fault_factor, combiner_count= len( component_combiners ), active_combiner_count= active_combiners, fault_active= fault_active, availability= availability, status= status, ) ) return records # ================================================================== # PLANT # ================================================================== def _simulate_plant( self, timestamp: datetime, panels: List[PanelData], combiners: List[CombinerData], inverters: List[InverterData], config: SimulationConfig, ) -> PlantData: dc_power = sum( item.dc_power_W for item in inverters ) ac_power = sum( item.ac_power_W for item in inverters ) nominal_power = max( ( item.nominal_power_W for item in inverters ), default=0.0 ) timestep_hours = ( config.timestep_minutes / 60.0 ) energy_Wh = ( ac_power * timestep_hours ) panel_count = len( panels ) active_panels = sum( 1 for panel in panels if panel.enabled ) availability = ( active_panels / panel_count if panel_count > 0 else 0.0 ) efficiency = ( ac_power / dc_power if dc_power > 0 else 0.0 ) performance_ratio = ( ac_power / nominal_power if nominal_power > 0 else 0.0 ) fault_factor = ( sum( panel.fault_factor for panel in panels ) / panel_count if panel_count > 0 else 1.0 ) fault_active = ( fault_factor < 1.0 ) if fault_active: status = "fault" elif ac_power <= 0: status = "idle" else: status = "normal" return PlantData( timestamp= timestamp, plant_id= self.plant.plant_id, dc_power_W= dc_power, ac_power_W= ac_power, effective_ac_power_W= ac_power, nominal_power_W= nominal_power, energy_Wh= energy_Wh, cumulative_energy_Wh= 0.0, inverter_count= len( inverters ), active_inverter_count= sum( 1 for item in inverters if item.status != "fault" ), combiner_count= len( combiners ), active_combiner_count= sum( 1 for item in combiners if item.status != "fault" ), panel_count= panel_count, active_panel_count= active_panels, efficiency= efficiency, performance_ratio= performance_ratio, availability= availability, fault_factor= fault_factor, fault_active= fault_active, status= status, ) # ================================================================== # FAULTS # ================================================================== def _simulate_faults( self, timestamp: datetime, ) -> List[FaultData]: records = [] active_faults = ( self.fault_manager .get_active_faults( timestamp ) ) for fault in active_faults: records.append( FaultData( timestamp= timestamp, fault_id= fault.fault_id, fault_type= fault.fault_type.value, component_level= fault.component_level.value, component_id= fault.component_id, severity= fault.severity, reduction_factor= fault.reduction_factor(), description= fault.description, ) ) return records # ================================================================== # TIMESTEP # ================================================================== def simulate_timestep( self, timestamp: datetime, config: SimulationConfig, ) -> SimulationStep: weather = ( self._build_weather_data( timestamp ) ) if config.generate_random_faults: self._generate_faults( timestamp ) panels = ( self._simulate_panels( timestamp, weather ) if config.store_panel_data else [] ) combiners = ( self._simulate_combiners( timestamp, panels ) if config.store_combiner_data else [] ) inverters = ( self._simulate_inverters( timestamp, combiners, weather ) if config.store_inverter_data else [] ) plant = ( self._simulate_plant( timestamp, panels, combiners, inverters, config ) if config.store_plant_data else None ) faults = ( self._simulate_faults( timestamp ) if config.store_fault_data else [] ) return SimulationStep( timestamp= timestamp, weather=( weather if config.store_weather_data else None ), panels= panels, combiners= combiners, inverters= inverters, plant= plant, faults= faults, ) # ================================================================== # RUN # ================================================================== def run( self, config: SimulationConfig, ) -> SimulationResult: result = SimulationResult() timestamp = config.start while timestamp < config.end: step = ( self.simulate_timestep( timestamp, config ) ) result.steps.append( step ) timestamp += timedelta( minutes= config.timestep_minutes ) return result # ================================================================== # SINGLE DAY # ================================================================== def run_single_day( self, date: datetime, timestep_minutes: int = 5, generate_random_faults: bool = False, ) -> SimulationResult: start = datetime( date.year, date.month, date.day ) config = SimulationConfig( start= start, end= start + timedelta( days=1 ), timestep_minutes= timestep_minutes, generate_random_faults= generate_random_faults, ) return self.run( config ) # ================================================================== # YEAR # ================================================================== def run_year( self, year: int, timestep_minutes: int = 15, generate_random_faults: bool = False, ) -> SimulationResult: start = datetime( year, 1, 1 ) end = datetime( year + 1, 1, 1 ) config = SimulationConfig( start= start, end= end, timestep_minutes= timestep_minutes, generate_random_faults= generate_random_faults, ) return self.run( config ) # ====================================================================== __all__ = [ "SimulationConfig", "SimulationResult", "PVSimulator", ]