hub/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:
  def __init__(self, building=None, pv_system=None, battery=None, electricity_demand=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):
    """
    :param building:
    :param tilt_angle:
    :param solar_angles:
    :param simulation_model_type:
    :param module_model_name:
    :param inverter_efficiency:
    :param system_catalogue_handler:
    :param roof_percentage_coverage:
    :param facade_coverage_percentage:
    """
    self.building = building
    self.electricity_demand = electricity_demand
    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):
    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, roof):
    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 = 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
    roof.installed_solar_collector_area = total_pv_area
    return panels_per_row, number_of_rows

  def system_assignation(self):
    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):
    if self.electricity_demand is not None:
      electricity_demand = self.electricity_demand
    else:
      electricity_demand = [demand*1000 for demand in
                            HourlyElectricityDemand(self.building).calculate()]
    rooftops_pv_output = [0] * len(electricity_demand)
    facades_pv_output = [0] * len(electricity_demand)
    rooftop_number_of_panels = 0
    if 'rooftop' in self.pv_installation_type.lower():
      for roof in self.building.roofs:
        if roof.perimeter_area > 40:
          np, ns = self.rooftop_sizing(roof)
          single_roof_number_of_panels = np * ns
          rooftop_number_of_panels += single_roof_number_of_panels
          if self.simulation_model_type == 'explicit':
            single_roof_pv_output = self.explicit_model(pv_system=self.pv_system,
                                                        inverter_efficiency=self.inverter_efficiency,
                                                        number_of_panels=single_roof_number_of_panels,
                                                        irradiance=roof.global_irradiance_tilted[cte.HOUR],
                                                        outdoor_temperature=self.building.external_temperature[
                                                          cte.HOUR])
            for i in range(len(rooftops_pv_output)):
              rooftops_pv_output[i] += single_roof_pv_output[i]
    total_hourly_pv_output = [rooftops_pv_output[i] + facades_pv_output[i] for i in range(8760)]
    imported_electricity = [0] * 8760
    exported_electricity = [0] * 8760
    self.building.self_sufficiency['hour'] = []
    for i in range(len(electricity_demand)):
      transfer = total_hourly_pv_output[i] - electricity_demand[i]
      self.building.self_sufficiency['hour'].append(transfer)
      if transfer > 0:
        exported_electricity[i] = transfer
      else:
        imported_electricity[i] = abs(transfer)
    self.building.self_sufficiency['year'] = sum(self.building.self_sufficiency['hour'])
    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(rooftops_pv_output) / 1000,
               'yearly_total_pv_production_kWh': sum(total_hourly_pv_output) / 1000,
               'specific_pv_production_kWh/kWp': sum(rooftops_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': rooftops_pv_output, 'T_out': self.building.external_temperature[cte.HOUR],
               'building_electricity_demand_W': 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()
    for energy_system in self.building.energy_systems:
      for generation_system in energy_system.generation_systems:
        if generation_system.system_type == cte.PHOTOVOLTAIC:
          generation_system.installed_capacity = (self.results['rooftop_panels'] *
                                                  float(generation_system.standard_test_condition_maximum_power))
    hourly_pv_output = self.results['total_hourly_pv_system_output_W']
    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)]
    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)