simplified_radiosity_algorithm/simplified_radiosity_algorithm.py

"""
SimplifiedRadiosityAlgorithm manage the interaction with the third party software CitySim_SRA
SPDX - License - Identifier: LGPL - 3.0 - or -later
Copyright © 2020 Project Author Pilar Monsalvete Alvarez de Uribarri pilar_monsalvete@concordia.ca
"""

import os
import shutil
from pathlib import Path
import pandas as pd
import subprocess
from subprocess import SubprocessError, TimeoutExpired, CalledProcessError

from exports.exports_factory import ExportsFactory
from imports.weather_factory import WeatherFactory
from helper.helper import Helper as mv
import helpers.constants as cte


class SimplifiedRadiosityAlgorithm:
  """
  SimplifiedRadiosityAlgorithm factory class
  """
  # todo: define executable as configurable parameter
 # _executable = 'citysim_sra'       # for linux
  _executable = 'shortwave_integer'    # for windows

  def __init__(self, city, sra_working_path, weather_file_name):
    self._city = city
    self._sra_working_path = sra_working_path
    self._weather_file_name = weather_file_name
    self._sra_in_file_name = self._city.name + '_sra.xml'
    self._sra_out_file_name = self._city.name + '_sra_SW.out'

    self._tmp_path = (sra_working_path / 'tmp').resolve()
    Path(self._tmp_path).mkdir(parents=True, exist_ok=True)
    # Ensure tmp file is empty
    for child in self._tmp_path.glob('*'):
      if child.is_file():
        child.unlink()

    self._cache_path = (sra_working_path / 'cache').resolve()
    Path(self._cache_path).mkdir(parents=True, exist_ok=True)

    self._results = None
    self._radiation = []

  def call_sra(self, key, keep_files=False):
    """
    creates required input files and calls the software
    """
    self._create_cli_file(key)
    ExportsFactory('sra', self._city, self._tmp_path).export()
    try:
      completed = subprocess.run([self._executable, str(Path(self._tmp_path / self._sra_in_file_name).resolve())])
    except (SubprocessError, TimeoutExpired, CalledProcessError) as error:
      raise Exception(error)
    file = (self._tmp_path / self._sra_out_file_name).resolve()
    new_path = (self._cache_path / self._sra_out_file_name).resolve()
    try:
     shutil.move(str(file), str(new_path))
    except Exception:
      raise Exception('No SRA output file found')
    if not keep_files:
      os.remove(Path(self._tmp_path / f'{self._city.name}_sra.xml').resolve())
    return completed

  @property
  def results(self):
    if self._results is None:
      try:
        path = (self._cache_path / self._sra_out_file_name).resolve()
        self._results = pd.read_csv(path, sep='\s+', header=0)
      except Exception:
        raise Exception('No SRA output file found')
    return self._results

  @property
  def radiation(self) -> []:
    if len(self._radiation) == 0:
      id_building = ''
      header_building = []
      for column in self.results.columns.values:
        if id_building != column.split(':')[1]:
          id_building = column.split(':')[1]
          if len(header_building) > 0:
            self._radiation.append(pd.concat([mv().month_hour, self.results[header_building]],
                                             axis=1))
          header_building = [column]
        else:
          header_building.append(column)
      self._radiation.append(pd.concat([mv().month_hour, self.results[header_building]], axis=1))
    return self._radiation

  def set_irradiance_surfaces(self, city, mode=0, building_name=None):
    """
    saves in building surfaces the correspondent irradiance at different time-scales depending on the mode
    if building is None, it saves all buildings' surfaces in file, if building is specified, it saves only that
    specific building values
    mode = 0, set only monthly values
    mode = 1, set only hourly values
    mode = 2, set both
    :parameter city: city
    :parameter mode: str (time-scale definition)
    :parameter building_name: str
    :return: none
    """
    for radiation in self.radiation:
      city_object_name = radiation.columns.values.tolist()[1].split(':')[1]
      if building_name is not None:
        if city_object_name != building_name:
          # todo: in case there is a specific building name defined, then assign values only to that specific building
          continue

      building = city.city_object(city_object_name)
      for column in radiation.columns.values:
        if column == cte.MONTH:
          continue
        header_id = column
        surface_id = header_id.split(':')[2]
        surface = building.surface_by_id(surface_id)
        new_value = pd.DataFrame(radiation[[header_id]].to_numpy(), columns=['sra'])
        if mode == 0 or mode == 2:
          # todo: this is wrong?? It must be used like this in MEB or what??
          month_new_value = mv().get_mean_values(new_value)
          if cte.MONTH not in surface.global_irradiance:
            surface.global_irradiance[cte.MONTH] = month_new_value
          else:
            pd.concat([surface.global_irradiance[cte.MONTH], month_new_value], axis=1)
        if mode == 1 or mode == 2:
          if cte.HOUR not in surface.global_irradiance:
            surface.global_irradiance[cte.HOUR] = new_value
          else:
            pd.concat([surface.global_irradiance[cte.HOUR], new_value], axis=1)

  def _create_cli_file(self, key):
    file = self._city.climate_file
    print(file)
    days_in_month = [31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    WeatherFactory(key, self._city, file_name=self._weather_file_name).enrich()
    content = self._city.name + '\n'
    content += str(self._city.latitude) + ',' + str(self._city.longitude) + ',0.0,' + str(self._city.time_zone) + '\n'
    content += '\ndm  m h G_Dh  G_Bn\n'
    total_days = 0
    for month in range(1, 13):
      if month > 1:
        total_days += days_in_month[month - 2]
      for day in range(1, days_in_month[month-1]+1):
        for hour in range(1, 25):
          if month == 1:
            i = 24 * (day-1) + hour - 1
          else:
            i = (total_days+day-1)*24 + hour - 1
          representative_building = self._city.buildings[0]
          content += str(day) + '  ' + str(month) + '  ' + str(hour) + '  ' \
              + str(representative_building.global_horizontal[cte.HOUR].epw[i]) + '  ' \
              + str(representative_building.beam[cte.HOUR].epw[i]) + '\n'
    with open(file, "w") as file:
      file.write(content)
    return