hub/pv_assessment/pv_system_assessment_with_lcoe.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,
               run_lcoe=False):
    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.csv_output = csv_output
    self.output_path = output_path
    self.run_lcoe = run_lcoe

    # Default LCOE parameters
    self.inflation_rate = 0.03
    self.discount_rate = 0.05
    self.period = 25
    self.degradation_rate = 0.01
    self.year_of_replacement_list = [12]
    self.replacement_ratio = 0.1
    self.installation_cost = 0
    self.tax_deduct = 0
    self.incentive = 0

    self.pv_hourly_generation = None
    self.t_cell = None
    self.results = {}

    if pv_system is not None:
      self.pv_system = pv_system
    else:
      self.pv_system = None
      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:
      self.battery = None
      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):
    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 0.0)
    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)

    pv_output = []
    for i in range(len(irradiance)):
      g_i = irradiance[i]
      t_out = outdoor_temperature[i]
      t_cell = t_out + (g_i / nominal_condition_irradiance) * (nominal_condition_cell_temperature - nominal_t_out)
      power = (inverter_efficiency * number_of_panels *
               (stc_power * (g_i / stc_irradiance) *
                (1 - cell_temperature_coefficient * (t_cell - stc_t_cell))))
      pv_output.append(power)
    return pv_output

  def rooftop_sizing(self, roof):
    pv_system = self.pv_system
    if self.module_model_name is not None:
      self.system_assignation()
    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)

    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'))

    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
          for attr in dir(generation_system):
            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))
          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 = [d * 1000 for d 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 self.pv_installation_type is not None and 'rooftop' in self.pv_installation_type.lower():
      for roof in self.building.roofs:
        if roof.perimeter_area > 40:
          npanels_per_row, nrows = self.rooftop_sizing(roof)
          single_roof_number_of_panels = npanels_per_row * nrows
          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(len(electricity_demand))]
    imported_electricity = []
    exported_electricity = []
    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.append(transfer)
        imported_electricity.append(0)
      else:
        exported_electricity.append(0)
        imported_electricity.append(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,
      'yearly_rooftop_tilted_radiation_{}_degree_kW/m2'.format(self.tilt_angle):
        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)
      ) if rooftop_number_of_panels > 0 else 0,
      'pv_installation': 'possible' if ((
        sum(rooftops_pv_output) /
        (float(self.pv_system.standard_test_condition_maximum_power) * rooftop_number_of_panels)
      ) if rooftop_number_of_panels > 0 else 0) > 760 else 'not possible',
      '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
    }

    if self.run_lcoe:
      stc_power_w = float(self.pv_system.standard_test_condition_maximum_power)
      capacity_kW = (results['rooftop_panels'] * stc_power_w) / 1000.0

      # Parametric cost function
      cost_per_kW = 3086.8 * (capacity_kW ** (-0.061))
      initial_cost = capacity_kW * cost_per_kW

      first_year_generation_PV = sum(total_hourly_pv_output) / 1000.0

      building_hourly_consumption_kWh = [x / 1000.0 for x in electricity_demand]
      PV_hourly_generation_kWh = [x / 1000.0 for x in total_hourly_pv_output]

      lcoe_pv = self._calculate_lcoe(
        capacity=capacity_kW,
        cost_per_kW=cost_per_kW,
        first_year_generation=first_year_generation_PV,
        period=self.period,
        discount_rate=self.discount_rate,
        degradation_rate=self.degradation_rate,
        year_of_replacement_list=self.year_of_replacement_list,
        replacement_ratio=self.replacement_ratio,
        inflation_rate=self.inflation_rate,
        initial_cost=initial_cost
      )
      results['LCOE_PV'] = lcoe_pv

      # If needed, you can also calculate system LCOE with the given method:
      # (If you want to represent Zahra's code exactly, this is how you do it)
      # lcoe_system = self._calculate_system_lcoe(
      #   PV_hourly_generation=PV_hourly_generation_kWh,
      #   building_hourly_consumption=building_hourly_consumption_kWh,
      #   grid_current_tariff=0.06704, # Example or from building function logic
      #   capacity=capacity_kW,
      #   cost_per_kW=cost_per_kW,
      #   first_year_generation_PV=first_year_generation_PV,
      #   period=self.period,
      #   discount_rate=self.discount_rate,
      #   degradation_rate=self.degradation_rate,
      #   year_of_replacement_list=self.year_of_replacement_list,
      #   replacement_ratio=self.replacement_ratio,
      #   inflation_rate=self.inflation_rate,
      #   initial_cost=initial_cost,
      #   installation_cost=self.installation_cost,
      #   tax_deduct=self.tax_deduct,
      #   incentive=self.incentive
      # )
      # results['LCOE_system'] = lcoe_system

    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)]
    self.building.pv_generation['LCOE_PV'] = self.results['LCOE_PV']
    self.building.pv_generation['PV_Installation'] = self.results['pv_installation']
    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'):
    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)]
    if list_value_keys:
      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")
      num_rows = list_lengths[0]
    else:
      num_rows = 1

    with open(output_path / filename, mode='w', newline='') as csv_file:
      writer = csv.writer(csv_file)
      for key in single_value_keys:
        writer.writerow([key, data[key]])
      writer.writerow([])
      if list_value_keys:
        writer.writerow(list_value_keys)
        for i in range(num_rows):
          row = [data[key][i] for key in list_value_keys]
          writer.writerow(row)

  def _discounted_total_generation_pv(self, first_year_generation_PV,
                                      period, discount_rate,
                                      degradation_rate=0.01):
    discounted_total_generation = 0
    discounted_generation_per_year = {}
    for year in range(1, period + 1):
      generation = (first_year_generation_PV *
                    ((1 - degradation_rate) ** (year - 1)) /
                    ((1 + discount_rate) ** year))
      discounted_generation_per_year[year] = generation
      discounted_total_generation += generation
    return discounted_generation_per_year, discounted_total_generation

  def _discounted_total_cost_pv(self, capacity,
                                cost_per_kW,
                                discount_rate,
                                year_of_replacement_list,
                                period, replacement_ratio,
                                inflation_rate,
                                initial_cost):
    opex_annual = {}
    discounted_annual_cost = {}
    discounted_total_cost = 0

    replacement_cost = {
      y: capacity * cost_per_kW * replacement_ratio * ((1 + inflation_rate) ** y) / ((1 + discount_rate) ** y)
      for y in year_of_replacement_list
    }

    for y in range(period + 1):
      # Annual OPEX with inflation and discounting
      opex_annual[y] = initial_cost * 0.01 * ((1 + inflation_rate) ** y) / ((1 + discount_rate) ** y)

    for y in range(period + 1):
      # Annual OPEX
      opex_annual[y] = initial_cost * 0.01 * ((1 + inflation_rate) ** y) / ((1 + discount_rate) ** y)

      # Total annual cost (OPEX + replacement CAPEX if applicable)
      if y == 0:
        annual_cost = initial_cost
      elif y in year_of_replacement_list:
        annual_cost = opex_annual[y] + replacement_cost[y]
      else:
        annual_cost = opex_annual[y]

      discounted_annual_cost[y] = annual_cost
      discounted_total_cost += annual_cost

    return opex_annual, initial_cost, discounted_annual_cost, discounted_total_cost, replacement_cost

  def _estimate_system_income(self, building_hourly_consumption,
                              PV_hourly_generation,
                              grid_current_tariff,
                              inflation_rate,
                              discount_rate,
                              period,
                              initial_cost,
                              installation_cost,
                              tax_deduct,
                              incentive):
    total_discounted_income = 0
    annual_discounted_income_dict = {}

    def net_metering_income(PV_hourly_generation, building_hourly_consumption, grid_tariff):
      PV_hourly_export = [max(PV_hourly_generation[i] - building_hourly_consumption[i], 0)
                          for i in range(len(PV_hourly_generation))]
      building_hourly_purchase = [max(building_hourly_consumption[i] - PV_hourly_generation[i], 0)
                                  for i in range(len(PV_hourly_generation))]
      annual_PV_export = sum(PV_hourly_export)
      annual_grid_purchase = sum(building_hourly_purchase)
      return min(annual_PV_export, annual_grid_purchase) * grid_tariff

    for year in range(1, period + 1):
      inflated_grid_tariff = (grid_current_tariff * ((1 + inflation_rate) ** year) / ((1 + discount_rate) ** year))

      building_hourly_self_consumption = [min(PV_hourly_generation[i], building_hourly_consumption[i])
                                          for i in range(len(PV_hourly_generation))]
      PV_hourly_export = [max(PV_hourly_generation[i] - building_hourly_consumption[i], 0)
                          for i in range(len(PV_hourly_generation))]

      self_consumption_income = sum(building_hourly_self_consumption) * inflated_grid_tariff
      net_metering_revenue = net_metering_income(PV_hourly_generation, building_hourly_consumption, inflated_grid_tariff)

      annual_tax_deduction_income = (initial_cost * (1 + tax_deduct) * ((1 - tax_deduct) ** (year - 1)) * tax_deduct)

      annual_discounted_income = self_consumption_income + net_metering_revenue + annual_tax_deduction_income
      annual_discounted_income_dict[year] = annual_discounted_income

      total_discounted_income += annual_discounted_income

    total_discounted_income += incentive
    return total_discounted_income, annual_discounted_income_dict

  def _calculate_lcoe(self, capacity,
                      cost_per_kW,
                      first_year_generation,
                      period,
                      discount_rate,
                      degradation_rate,
                      year_of_replacement_list,
                      replacement_ratio,
                      inflation_rate,
                      initial_cost):
    _, _, _, discounted_total_cost, _ = self._discounted_total_cost_pv(
      capacity,
      cost_per_kW,
      discount_rate,
      year_of_replacement_list,
      period,
      replacement_ratio,
      inflation_rate,
      initial_cost
    )

    _, discounted_total_generation = self._discounted_total_generation_pv(
      first_year_generation, period, discount_rate, degradation_rate
    )

    if discounted_total_generation == 0:
      raise ValueError("Discounted generation is zero, cannot calculate LCOE.")

    lcoe = discounted_total_cost / discounted_total_generation
    return lcoe

  def _calculate_system_lcoe(self,
                             PV_hourly_generation,
                             building_hourly_consumption,
                             grid_current_tariff,
                             capacity,
                             cost_per_kW,
                             first_year_generation_PV,
                             period,
                             discount_rate,
                             degradation_rate,
                             year_of_replacement_list,
                             replacement_ratio,
                             inflation_rate,
                             initial_cost,
                             installation_cost,
                             tax_deduct,
                             incentive):
    PV_hourly_export = [max(PV_hourly_generation[i] - building_hourly_consumption[i], 0) for i in range(len(PV_hourly_generation))]
    building_hourly_purchase = [max(building_hourly_consumption[i] - PV_hourly_generation[i], 0) for i in range(len(PV_hourly_generation))]
    building_hourly_self_consumption = [min(PV_hourly_generation[i], building_hourly_consumption[i]) for i in range(len(PV_hourly_generation))]

    annual_PV_export = sum(PV_hourly_export)
    annual_grid_purchase = sum(building_hourly_purchase)
    annual_building_self_consumption = sum(building_hourly_self_consumption)

    total_energy = annual_building_self_consumption + annual_grid_purchase + annual_PV_export

    share_self = annual_building_self_consumption / total_energy if total_energy > 0 else 0
    share_grid = annual_grid_purchase / total_energy if total_energy > 0 else 0
    share_export = annual_PV_export / total_energy if total_energy > 0 else 0

    lcoe_pv = self._calculate_lcoe(
      capacity,
      cost_per_kW,
      first_year_generation_PV,
      period,
      discount_rate,
      degradation_rate,
      year_of_replacement_list,
      replacement_ratio,
      inflation_rate,
      initial_cost
    )

    lcoe_grid = grid_current_tariff

    pv_export_income, annual_discounted_income_dict = self._estimate_system_income(
      building_hourly_consumption,
      PV_hourly_generation,
      grid_current_tariff,
      inflation_rate,
      discount_rate,
      period,
      initial_cost,
      installation_cost,
      tax_deduct,
      incentive
    )

    lcoe_export = -pv_export_income / annual_PV_export if annual_PV_export > 0 else 0

    lcoe_system = (
      share_self * lcoe_pv +
      share_grid * lcoe_grid +
      share_export * lcoe_export
    )

    return lcoe_system