""" 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 """ if os.name == 'nt': _executable = 'shortwave_integer' # for windows elif os.name == 'posix': _executable = 'citysim_sra' # for linux / mac 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, selected_buildings=None): """ creates required input files and calls the software """ self._create_cli_file(key) ExportsFactory('sra', self._city, self._tmp_path, selected_buildings).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