pv_workflow/pv_assessment/pv_system_assessment.py

import math
import csv
import hub.helpers.constants as cte
from pv_assessment.electricity_demand_calculator import HourlyElectricityDemand
from hub.catalog_factories.energy_systems_catalog_factory import EnergySystemsCatalogFactory
from hub.helpers.monthly_values import MonthlyValues


class PvSystemAssessment:
  """
  A class to assess and size photovoltaic (PV) systems for buildings.

  This class provides methods for sizing PV systems on rooftops, calculating their energy output,
  assigning system components from a catalog, and evaluating grid-tied systems. It also allows for
  CSV output of system results.

  Attributes:
      building (Building): The building for which the PV system is being assessed.
      pv_system (PvSystem): The photovoltaic system used in the assessment.
      battery (EnergyStorageSystem): The energy storage system associated with the PV system, if any.
      tilt_angle (float): The tilt angle of the PV system.
      solar_angles (DataFrame): Solar angles used in simulation (e.g., solar altitude, azimuth).
      pv_installation_type (str): Type of installation (e.g., rooftop, facade).
      simulation_model_type (str): Type of simulation model (e.g., 'explicit').
      module_model_name (str): Model name of the PV module being used.
      inverter_efficiency (float): Efficiency of the inverter.
      system_catalogue_handler (str): Handler for the PV system catalog.
      roof_percentage_coverage (float): Percentage of the roof to be covered by PV panels.
      facade_coverage_percentage (float): Percentage of the facade to be covered by PV panels.
      csv_output (bool): Whether or not to generate CSV output of the results.
      output_path (str): Path to store the output CSV file.
      results (dict): Dictionary to store the results of the system assessment.
  """
  def __init__(self, building=None, pv_system=None, battery=None, tilt_angle=None, solar_angles=None,
               pv_installation_type=None, simulation_model_type=None, module_model_name=None,
               inverter_efficiency=None, system_catalogue_handler=None, roof_percentage_coverage=None,
               facade_coverage_percentage=None, csv_output=False, output_path=None):
    """
    Initializes the PvSystemAssessment instance with the provided parameters or defaults.

    :param building: The building object for which the PV system is assessed.
    :param pv_system: The photovoltaic system (optional, defaults to building's existing PV system).
    :param battery: The battery storage system (optional, defaults to building's associated storage).
    :param tilt_angle: Tilt angle of the PV system.
    :param solar_angles: DataFrame containing solar angles (e.g., solar altitude, azimuth).
    :param pv_installation_type: Type of installation ('rooftop', 'facade', etc.).
    :param simulation_model_type: Type of simulation model ('explicit').
    :param module_model_name: Model name of the PV module.
    :param inverter_efficiency: Efficiency of the inverter.
    :param system_catalogue_handler: Catalog handler to select PV module from a catalog.
    :param roof_percentage_coverage: Percentage of roof to cover with PV modules.
    :param facade_coverage_percentage: Percentage of facade to cover with PV modules.
    :param csv_output: Boolean indicating whether to generate CSV output.
    :param output_path: Path for saving the CSV output.
    """
    self.building = building
    self.tilt_angle = tilt_angle
    self.solar_angles = solar_angles
    self.pv_installation_type = pv_installation_type
    self.simulation_model_type = simulation_model_type
    self.module_model_name = module_model_name
    self.inverter_efficiency = inverter_efficiency
    self.system_catalogue_handler = system_catalogue_handler
    self.roof_percentage_coverage = roof_percentage_coverage
    self.facade_coverage_percentage = facade_coverage_percentage
    self.pv_hourly_generation = None
    self.t_cell = None
    self.results = {}
    self.csv_output = csv_output
    self.output_path = output_path
    if pv_system is not None:
      self.pv_system = pv_system
    else:
      for energy_system in self.building.energy_systems:
        for generation_system in energy_system.generation_systems:
          if generation_system.system_type == cte.PHOTOVOLTAIC:
            self.pv_system = generation_system
    if battery is not None:
      self.battery = battery
    else:
      for energy_system in self.building.energy_systems:
        for generation_system in energy_system.generation_systems:
          if generation_system.system_type == cte.PHOTOVOLTAIC and generation_system.energy_storage_systems is not None:
            for storage_system in generation_system.energy_storage_systems:
              if storage_system.type_energy_stored == cte.ELECTRICAL:
                self.battery = storage_system

  @staticmethod
  def explicit_model(pv_system, inverter_efficiency, number_of_panels, irradiance, outdoor_temperature):
    """
    Calculates the PV system's energy output using the explicit model.

    :param pv_system: The photovoltaic system.
    :param inverter_efficiency: Efficiency of the inverter.
    :param number_of_panels: Number of PV panels in the system.
    :param irradiance: Hourly irradiance values.
    :param outdoor_temperature: Hourly outdoor temperature values.
    :return: List of energy outputs for each hour.
    """
    inverter_efficiency = inverter_efficiency
    stc_power = float(pv_system.standard_test_condition_maximum_power)
    stc_irradiance = float(pv_system.standard_test_condition_radiation)
    cell_temperature_coefficient = float(pv_system.cell_temperature_coefficient) / 100 if (
        pv_system.cell_temperature_coefficient is not None) else None
    stc_t_cell = float(pv_system.standard_test_condition_cell_temperature)
    nominal_condition_irradiance = float(pv_system.nominal_radiation)
    nominal_condition_cell_temperature = float(pv_system.nominal_cell_temperature)
    nominal_t_out = float(pv_system.nominal_ambient_temperature)
    g_i = irradiance
    t_out = outdoor_temperature
    t_cell = []
    pv_output = []
    for i in range(len(g_i)):
      t_cell.append((t_out[i] + (g_i[i] / nominal_condition_irradiance) *
                     (nominal_condition_cell_temperature - nominal_t_out)))
      pv_output.append((inverter_efficiency * number_of_panels * (stc_power * (g_i[i] / stc_irradiance) *
                                                                  (1 - cell_temperature_coefficient *
                                                                   (t_cell[i] - stc_t_cell)))))
    return pv_output

  def rooftop_sizing(self):
    """
    Sizing of the rooftop PV system.

    :return: Number of panels per row and number of rows to be installed.
    """
    pv_system = self.pv_system
    if self.module_model_name is not None:
      self.system_assignation()
    # System Sizing
    module_width = float(pv_system.width)
    module_height = float(pv_system.height)
    roof_area = 0
    for roof in self.building.roofs:
      roof_area += roof.perimeter_area
    pv_module_area = module_width * module_height
    available_roof = (self.roof_percentage_coverage * roof_area)
    # Inter-Row Spacing
    winter_solstice = self.solar_angles[(self.solar_angles['AST'].dt.month == 12) &
                                        (self.solar_angles['AST'].dt.day == 21) &
                                        (self.solar_angles['AST'].dt.hour == 12)]
    solar_altitude = winter_solstice['solar altitude'].values[0]
    solar_azimuth = winter_solstice['solar azimuth'].values[0]
    distance = ((module_height * math.sin(math.radians(self.tilt_angle)) * abs(
      math.cos(math.radians(solar_azimuth)))) / math.tan(math.radians(solar_altitude)))
    distance = float(format(distance, '.2f'))
    # Calculation of the number of panels
    space_dimension = math.sqrt(available_roof)
    space_dimension = float(format(space_dimension, '.2f'))
    panels_per_row = math.ceil(space_dimension / module_width)
    number_of_rows = math.ceil(space_dimension / distance)
    total_number_of_panels = panels_per_row * number_of_rows
    total_pv_area = total_number_of_panels * pv_module_area
    self.building.roofs[0].installed_solar_collector_area = total_pv_area
    return panels_per_row, number_of_rows

  def system_assignation(self):
    """
    Assigns a PV system from the energy systems catalog based on the module model name.

    :raises ValueError: If no PV module with the provided model name exists in the catalog.
    """
    generation_units_catalogue = EnergySystemsCatalogFactory(self.system_catalogue_handler).catalog
    catalog_pv_generation_equipments = [component for component in
                                        generation_units_catalogue.entries('generation_equipments') if
                                        component.system_type == 'photovoltaic']
    selected_pv_module = None
    for pv_module in catalog_pv_generation_equipments:
      if self.module_model_name == pv_module.model_name:
        selected_pv_module = pv_module
    if selected_pv_module is None:
      raise ValueError("No PV module with the provided model name exists in the catalogue")
    for energy_system in self.building.energy_systems:
      for idx, generation_system in enumerate(energy_system.generation_systems):
        if generation_system.system_type == cte.PHOTOVOLTAIC:
          new_system = selected_pv_module
        # Preserve attributes that exist in the original but not in the new system
          for attr in dir(generation_system):
            # Skip private attributes and methods
            if not attr.startswith('__') and not callable(getattr(generation_system, attr)):
              if not hasattr(new_system, attr):
                setattr(new_system, attr, getattr(generation_system, attr))
        # Replace the old generation system with the new one
          energy_system.generation_systems[idx] = new_system

  def grid_tied_system(self):
    """
    Evaluates the performance and sizing of a grid-tied PV system.

    This method assesses energy production, storage, and demand matching for a grid-tied PV system.

    :return: The evaluation results as a dictionary.
    """
    building_hourly_electricity_demand = [demand / cte.WATTS_HOUR_TO_JULES for demand in
                                          HourlyElectricityDemand(self.building).calculate()]
    rooftop_pv_output = [0] * 8760
    facade_pv_output = [0] * 8760
    rooftop_number_of_panels = 0
    if 'rooftop' in self.pv_installation_type.lower():
      np, ns = self.rooftop_sizing()
      if self.simulation_model_type == 'explicit':
        rooftop_number_of_panels = np * ns
        rooftop_pv_output = self.explicit_model(pv_system=self.pv_system,
                                                inverter_efficiency=self.inverter_efficiency,
                                                number_of_panels=rooftop_number_of_panels,
                                                irradiance=self.building.roofs[0].global_irradiance_tilted[
                                                  cte.HOUR],
                                                outdoor_temperature=self.building.external_temperature[
                                                  cte.HOUR])

    total_hourly_pv_output = [rooftop_pv_output[i] + facade_pv_output[i] for i in range(8760)]
    imported_electricity = [0] * 8760
    exported_electricity = [0] * 8760
    for i in range(8760):
      transfer = total_hourly_pv_output[i] - building_hourly_electricity_demand[i]
      if transfer > 0:
        exported_electricity[i] = transfer
      else:
        imported_electricity[i] = abs(transfer)

    results = {'building_name': self.building.name,
               'total_floor_area_m2': self.building.thermal_zones_from_internal_zones[0].total_floor_area,
               'roof_area_m2': self.building.roofs[0].perimeter_area, 'rooftop_panels': rooftop_number_of_panels,
               'rooftop_panels_area_m2': self.building.roofs[0].installed_solar_collector_area,
               'yearly_rooftop_ghi_kW/m2': self.building.roofs[0].global_irradiance[cte.YEAR][0] / 1000,
               f'yearly_rooftop_tilted_radiation_{self.tilt_angle}_degree_kW/m2':
                 self.building.roofs[0].global_irradiance_tilted[cte.YEAR][0] / 1000,
               'yearly_rooftop_pv_production_kWh': sum(rooftop_pv_output) / 1000,
               'yearly_total_pv_production_kWh': sum(total_hourly_pv_output) / 1000,
               'specific_pv_production_kWh/kWp': sum(rooftop_pv_output) / (
                   float(self.pv_system.standard_test_condition_maximum_power) * rooftop_number_of_panels),
               'hourly_rooftop_poa_irradiance_W/m2': self.building.roofs[0].global_irradiance_tilted[cte.HOUR],
               'hourly_rooftop_pv_output_W': rooftop_pv_output, 'T_out': self.building.external_temperature[cte.HOUR],
               'building_electricity_demand_W': building_hourly_electricity_demand,
               'total_hourly_pv_system_output_W': total_hourly_pv_output, 'import_from_grid_W': imported_electricity,
               'export_to_grid_W': exported_electricity}
    return results

  def enrich(self):
    system_archetype_name = self.building.energy_systems_archetype_name
    archetype_name = '_'.join(system_archetype_name.lower().split())
    if 'grid_tied' in archetype_name:
      self.results = self.grid_tied_system()
    hourly_pv_output = self.results['total_hourly_pv_system_output_W']
    if self.building.onsite_electrical_production:
      self.building.pv_generation[cte.HOUR] = hourly_pv_output
      self.building.pv_generation[cte.MONTH] = MonthlyValues.get_total_month(hourly_pv_output)
      self.building.pv_generation[cte.YEAR] = [sum(hourly_pv_output)]
    else:
      self.building.onsite_electrical_production[cte.HOUR] = hourly_pv_output
      self.building.onsite_electrical_production[cte.MONTH] = MonthlyValues.get_total_month(hourly_pv_output)
      self.building.onsite_electrical_production[cte.YEAR] = [sum(hourly_pv_output)]
    if self.csv_output:
      self.save_to_csv(self.results, self.output_path, f'{self.building.name}_pv_system_analysis.csv')

  @staticmethod
  def save_to_csv(data, output_path, filename='rooftop_system_results.csv'):
    # Separate keys based on whether their values are single values or lists
    single_value_keys = [key for key, value in data.items() if not isinstance(value, list)]
    list_value_keys = [key for key, value in data.items() if isinstance(value, list)]

    # Check if all lists have the same length
    list_lengths = [len(data[key]) for key in list_value_keys]
    if not all(length == list_lengths[0] for length in list_lengths):
      raise ValueError("All lists in the dictionary must have the same length")

    # Get the length of list values (assuming all lists are of the same length, e.g., 8760 for hourly data)
    num_rows = list_lengths[0] if list_value_keys else 1

    # Open the CSV file for writing
    with open(output_path / filename, mode='w', newline='') as csv_file:
      writer = csv.writer(csv_file)
    # Write single-value data as a header section
      for key in single_value_keys:
        writer.writerow([key, data[key]])
    # Write an empty row for separation
      writer.writerow([])
    # Write the header for the list values
      writer.writerow(list_value_keys)
    # Write each row for the lists
      for i in range(num_rows):
        row = [data[key][i] for key in list_value_keys]
        writer.writerow(row)