hub/exports/energy_systems/heat_pump_export.py

"""
HeatPumpExport exports heatpump outputs into several files after insel execution
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2022 Concordia CERC group
Project Coder Peter Yefi peteryefi@gmail.com
"""
import os
from typing import List, Union, Dict
import yaml
from string import Template
import pandas as pd


class HeatPumpExport:
  """
  Exports heat pump values as coefficients
  of some defined function
  """

  def __init__(self, base_path, city, output_path, template, demand_path=None, water_temp=None):
    self._template_path = template
    self._water_temp = water_temp
    self._constants_path = (base_path / 'heat_pumps/constants.yaml')
    # needed to compute max demand.
    self._demand_path = (base_path / 'heat_pumps/demand.txt') if demand_path is None else demand_path
    self._city = city
    self._input_data = None
    self._base_path = base_path
    self._output_path = output_path

  def _run_insel(self, user_input: Dict, capacity_coeff: List, filename: str) -> Union[Dict, None]:
    """
    Runs insel and write the necessary files
    :param user_input: a dictionary containing the user
    values necessary to run insel
    :param capacity_coeff: a list containing capacity coefficients
    :param filename: the name of the insel file to be created
    :return:
    """
    self._input_data = user_input
    self._update_input_data_with_coff(capacity_coeff)
    # update input data with constants
    self._update_input_data_with_constants()
    # update input data with input and output files for insel
    self._update_input_data_with_files()
    insel_file_handler = None
    insel_template_handler = None
    try:
      # run insel
      insel_template_handler = open(self._template_path, "r")
      insel_template_content = insel_template_handler.read()
      insel_template = Template(insel_template_content).substitute(self._input_data)
      # create the insel file and write the template with substituted values into it
      insel_file = (self._base_path / 'heat_pumps' / filename)
      insel_file_handler = open(insel_file, "w")
      insel_file_handler.write(insel_template)
      # Now run insel
      self._delete_existing_output_files()
      os.system('insel {}'.format(insel_file))
      # Writer headers to csv output files generated by insel
      self._write_insel_output_headers()
      # User output
      return self._get_user_out_put()
    except IOError as err:
      print("I/O exception: {}".format(err))
    finally:
      insel_file_handler.close()
      insel_template_handler.close()

  def _write_insel_output_headers(self):
    """
    Write headers to the various csv file generated by insel
    :return:
    """
    header = [
      'Year', ' Month', ' Day', 'Hour', 'Minute', 'HP Heat Output (kW)', 'Heating Demand (kW)', 'HP output flow rate',
      'Building Required Flow Rate', 'TES Charging Rate (kg/s)', 'Water Flow Rate After Splitter',
      'water temperature after splitter', 'TES Discharging Rate (kg/s)', 'TES discharge temperature',
      'Mixer Outlet Flow Rate (kg/s)', 'Mixer outlet temperature', 'Auxiliary heater fuel flow rate',
      'Auxiliary heater energy input (kW)', 'Building Inlet Flow Rate (kg/s)', 'Building inlet temperature',
      'Building return temperature', 'TES Return Flow Rate (kg/s)', 'TES return temperature',
      'TES Bypass Line Flow Rate (kg/s)', 'TES bypass line temperature', 'Flow Rate from TES to mixer 2 (kg/s)',
      'Temperature from Tes to mixer', 'HP Inlet Flow Rate (kg/s)', 'HP Inlet temperature', 'TES Node 1 Temperature',
      'TES Node 2 Temperature', 'TES Node 3 Temperature', 'TES Node 4 Temperature', 'TES Energy Content (J)',
      'HP Electricity Consumption (kW)', 'HP COP', 'Ambient Temperature', 'HP Operational Cost (CAD)',
      'Auxiliary Heater Operational Cost (CAD)', 'Operational CO2 Emissions of HP (g)',
      'Operational CO2 Emissions of Auxiliary Heater (g)']
    if 'series' in str(self._template_path):
      header = [
        'Year', ' Month', ' Day', 'Hour', 'Minute', 'HP Heat Output (kW)',
        'HP Electricity Consumption (kW)', 'HP COP', 'TES Charging Rate (kg/s)',
        'TES Discharging Rate (kg/s)', 'TES Node 1 Temperature', 'TES Node 2 Temperature',
        'TES Node 3 Temperature', 'TES Node 4 Temperature', 'TES Energy Content (J)',
        'TES Energy Content (kWh)', 'TES Energy Content Variation (kWh)',
        'Auxiliary Heater Fuel Flow Rate (kg/s)', 'Auxiliary Heater Energy Input (kW)',
        'HP Operational Cost (CAD)', 'Auxiliary Heater Operational Cost (CAD)',
        'Operational CO2 Emissions of HP (g)',
        'Operational CO2 Emissions of Auxiliary Heater (g)',
        'Return Temperature', 'Demand (kW)']
    header_data = {
      self._input_data['fileOut1']: header,
      self._input_data['fileOut2']: ['Day', 'Operational Daily Emissions from Heat Pumps (g)',
                                     'Operational Daily Emissions from Auxiliary Heater (g)'],
      self._input_data['fileOut3']: ['Month', 'Monthly Operational Costs of Heat Pumps (CAD)',
                                     'Monthly Operational Costs of Auxiliary Heater (CAD)'],
      self._input_data['fileOut4']: ['Month', 'Monthly Fuel Consumption of Auxiliary Heater (m3)'],
      self._input_data['fileOut5']: ['Month', 'Operational Monthly Emissions from Heat Pumps (g)',
                                     'Operational Monthly Emissions from Auxiliary Heater (g)'],
      self._input_data['fileOut6']: ['Day', 'Daily HP Electricity Demand (kWh)'],
      self._input_data['fileOut7']: ['Day', 'Daily Operational Costs of Heat Pumps (CAD)',
                                     'Daily Operational Costs of Auxiliary Heater (CAD)'],
      self._input_data['fileOut8']: ['Month', 'Monthly HP Electricity Demand (kWh)'],
      self._input_data['fileOut9']: ['Day', 'Daily Fuel Consumption of Auxiliary Heater (m3)'],
      self._input_data['fileOut10']: ['Year', 'Month', 'Day', 'Hour', 'HP Electricity Demand (kWh)']
    }

    for file_path, header in header_data.items():
      file_path = file_path.strip("'")
      df = pd.read_csv(file_path, header=None, sep='\s+')
      # ignore ambient temperature for air source series run
      if df.shape[1] > 25 and 'series' in str(self._template_path):
        df.drop(columns=df.columns[-1],
                axis=1,
                inplace=True)
      df.to_csv(file_path, header=header)

  def _update_input_data_with_files(self):
    """
    Updates input data for insel with some files that will
    be written to after insel runs. Also specifies and input file
    which is the Heating Demand (demand.txt) file
    :return:
    """
    self._input_data["HeatingDemand"] = f"'{str(self._demand_path)}'"
    self._input_data["fileOut1"] = f"'{str((self._base_path / 'heat_pumps/technical_performance.csv'))}'"
    self._input_data["fileOut2"] = f"'{str((self._base_path / 'heat_pumps/system_daily_emissions.csv'))}'"
    self._input_data["fileOut3"] = f"'{str((self._base_path / 'heat_pumps/monthly_operational_costs.csv'))}'"
    self._input_data["fileOut4"] = f"'{str((self._base_path / 'heat_pumps/monthly_fossil_fuel_consumptions.csv'))}'"
    self._input_data["fileOut5"] = f"'{str((self._base_path / 'heat_pumps/system_monthly_emissions.csv'))}'"
    self._input_data["fileOut6"] = f"'{str((self._base_path / 'heat_pumps/daily_hp_electricity_demand.csv'))}'"
    self._input_data["fileOut7"] = f"'{str((self._base_path / 'heat_pumps/daily_operational_costs.csv'))}'"
    self._input_data["fileOut8"] = f"'{str((self._base_path / 'heat_pumps/monthly_hp_electricity_demand.csv'))}'"
    self._input_data["fileOut9"] = f"'{str((self._base_path / 'heat_pumps/daily_fossil_fuel_consumption.csv'))}'"
    self._input_data["fileOut10"] = f"'{str((self._base_path / 'heat_pumps/hp_hourly_electricity_demand.csv'))}'"
    # include water temperature for water to water heat pump
    if self._water_temp is not None:
      self._input_data['WaterTemperature'] = f"'{str(self._water_temp)}'"

  def _delete_existing_output_files(self):
    """
    Remove existing out files generated by insel before
    running insel
    :return:
    """
    for key, file_path in self._input_data.items():
      if 'fileOut' in key:
        file_path = file_path.strip("'")
        try:
          os.remove(file_path)
        except OSError:
          pass

  def _compute_max_demand(self):
    """
    Retrieves the maximum demand value from
    the demands text file
    :return: float
    """
    max_demand = -1
    with open(self._demand_path) as file_handler:
      for demand in file_handler.readlines():
        if float(demand) > max_demand:
          max_demand = float(demand)
    return max_demand

  def _update_input_data_with_constants(self):
    with open(self._constants_path) as file:
      constants_dict = yaml.load(file, Loader=yaml.FullLoader)
      for key, value in constants_dict.items():
        if key in ['LowestPossibleLoadFlow', 'HighestPossibleLoadFlow'] and self._water_temp is None:
          continue
        self._input_data[key] = value
    # compute water to water HP specific values
    if 55 <= self._input_data['HPSupTemp'] <= 60:
      self._input_data["HPDisactivationTemperature"] = self._input_data["HPSupTemp"] - 5
      self._input_data["HPReactivationTemperature"] = self._input_data["HPSupTemp"] - 18
    elif 50 <= self._input_data["HPSupTemp"] < 55:
      self._input_data["HPDisactivationTemperature"] = self._input_data["HPSupTemp"] - 5
      self._input_data["HPReactivationTemperature"] = self._input_data["HPSupTemp"] - 13
    elif 45 <= self._input_data["HPSupTemp"] < 50:
      self._input_data["HPDisactivationTemperature"] = self._input_data["HPSupTemp"] - 3
      self._input_data["HPReactivationTemperature"] = self._input_data["HPSupTemp"] - 8
    elif 35 <= self._input_data["HPSupTemp"] < 40:
      self._input_data["HPDisactivationTemperature"] = self._input_data["HPSupTemp"] - 2
      self._input_data["HPReactivationTemperature"] = self._input_data["HPSupTemp"] - 4

    # compute maximum demand. TODO: This should come from catalog in the future
    max_demand = self._compute_max_demand()
    # compute TESCapacity
    self._input_data["TESCapacity"] = self._input_data["HoursOfStorageAtMaxDemand"] * (max_demand * 3.6) / (
        (self._input_data["Cp"] / 1000) * self._input_data["TemperatureDifference"])

  def _update_input_data_with_coff(self, a_coeff: List):
    """
    Updates the user data with coefficients derived from imports
    :param a_coeff: insel a coefficient values
    Meaning of a is in the models for air source heat pump
    and water to water source heat pump
    :return:
    """

    self._input_data["a1"] = a_coeff[0]
    self._input_data["a2"] = a_coeff[1]
    self._input_data["a3"] = a_coeff[2]
    self._input_data["a4"] = a_coeff[3]
    self._input_data["a5"] = a_coeff[4]
    self._input_data["a6"] = a_coeff[5]

    # additional coefficients for water to water source
    if self._water_temp is not None:
      self._input_data["a7"] = a_coeff[6]
      self._input_data["a8"] = a_coeff[7]
      self._input_data["a9"] = a_coeff[8]
      self._input_data["a10"] = a_coeff[9]
      self._input_data["a11"] = a_coeff[10]

  def _get_user_out_put(self) -> Union[Dict, None]:
    """
    Extracts monthly electricity demand and fossil fuel consumption
    from output files generated by insel
    :return: Dict for json output
    """
    demand_data = 'fileOut8'
    fossil_data = 'fileOut4'
    fossil_index = 2
    demand_index = 2

    if self._output_path is None:
      demand_data = 'fileOut10'
      fossil_data = 'fileOut9'
      demand_index = 5

    electricity_df = pd.read_csv(self._input_data[demand_data].strip("'")).iloc[:, demand_index]
    fossil_df = pd.read_csv(self._input_data[fossil_data].strip("'")).iloc[:, fossil_index]

    if self._output_path is None:
      return {
        'hourly_electricity_demand': electricity_df.values.tolist(),
        'daily_fossil_consumption': fossil_df.values.tolist()
      }

    data = [electricity_df, fossil_df]
    df = pd.concat(data, axis=1)
    df = pd.concat([df, df.agg(['sum'])])
    s = pd.Series(["Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sept", "Oct", "Nov", "Dec", "Total"])
    df = df.set_index([s])
    df.to_csv(self._output_path)