create_profile_from_scenario.py

import sys, pathlib
sys.path.append(str(pathlib.Path(__file__).parent.resolve().parent.joinpath('src')))

import datetime
import pytz
import numpy as np
import pandas as pd
from pathlib import Path
import csv
import yaml

from bems.helpers.data import read_block_from_simx_file

VENTILATION_POWER_WEEKDAY = 98.15  # in kW, only on weekdays between 6am and 8pm
VENTILATION_POWER = 27.37  # in W, only if not weekday between 6am and 8pm

def get_scenario_start_date(scenario_path: Path) -> datetime.datetime:
    with open(scenario_path.joinpath('Info/scenario_definition.yml'), 'r') as file:
        definition = yaml.safe_load(file)
        t_start = pytz.timezone('Europe/Berlin').localize(definition['general_description']['t_start'])

    return t_start


def retrieve_full_load_profile(scenario_path: Path) -> pd.DataFrame:
    df = None
    for file in scenario_path.joinpath('ideal_prediction_data/loads').glob('*.txt'):
        data = read_block_from_simx_file(file, 'pel')
        df_file = pd.DataFrame(data=data, columns=['time', 'P_dem']).set_index('time').sort_index()

        if df is None:
            df = df_file
        else:
            df['P_dem'] += df_file['P_dem']

    # convert W to kW
    df['P_dem'] /= 1000

    return df


def retrieve_server_load_profiles(scenario_path: Path) -> pd.DataFrame:
    file_server_o4 = next(scenario_path.joinpath('ideal_prediction_data/loads').glob('Server*.txt'))
    file_server_cis = next(scenario_path.joinpath('ideal_prediction_data/loads').glob('ClusterCIS*.txt'))

    data_server_o4 = read_block_from_simx_file(file_server_o4, 'pel')
    data_server_cis = read_block_from_simx_file(file_server_cis, 'pel')
    df_server_o4 = pd.DataFrame(data=data_server_o4, columns=['time', 'P_server_o4']).set_index('time').sort_index()
    df_server_cis = pd.DataFrame(data=data_server_cis, columns=['time', 'P_server_cis']).set_index('time').sort_index()

    df = pd.concat((df_server_o4, df_server_cis), axis=1)

    # convert W to kW
    df /= 1000

    return df


def retrieve_temperature_profile(scenario_path: Path) -> pd.DataFrame:
    file = next(scenario_path.joinpath('ideal_prediction_data/weather').glob('*.txt'))
    data = read_block_from_simx_file(file, 'TAmbientTable')
    df = pd.DataFrame(data=data, columns=['time', 'theta_air']).set_index('time').sort_index()

    # shift time-index to start at 0
    df.index -= df.index[0]

    return df


def retrieve_pv_profile(scenario_path: Path) -> pd.DataFrame:
    file = next(scenario_path.joinpath('ideal_prediction_data/PV').glob('*.txt'))
    data = read_block_from_simx_file(file, 'Pel_pred_PV')
    df = pd.DataFrame(data=data, columns=['time', 'P_ren']).set_index('time').sort_index()

    # convert W to kW
    df['P_ren'] /= 1000

    return df


def adjust_load_profile(df: pd.DataFrame, ventilation_times: tuple[float]):
    df_copy = df.copy()
    condition = (df.index.weekday > 4) | (df.index.hour < ventilation_times[0]) | (df.index.hour > ventilation_times[1])
    df_copy[condition] += VENTILATION_POWER
    df_copy[~condition] += VENTILATION_POWER_WEEKDAY

    return df_copy


def generate_bems_profile(
    scenario_path: Path,
    output_file: Path = None,
    ventilation_times: tuple[float] = None
) -> pd.DataFrame:
    t0 = get_scenario_start_date(scenario_path)

    df_load = retrieve_full_load_profile(scenario_path)
    df_pv = retrieve_pv_profile(scenario_path)
    df_temp = retrieve_temperature_profile(scenario_path)
    df_server = retrieve_server_load_profiles(scenario_path)

    df = pd.concat(
        (df_load, df_pv, df_temp, df_server), axis=1
    )
    df.reset_index(inplace=True)

    timedelta = pd.to_timedelta(df["time"], unit="h")

    # add time in seconds based on unix timestamp of t0
    df["time"] = (pd.to_timedelta(t0.timestamp(), unit="s") + timedelta).dt.total_seconds()

    # convert hours into time-aware datetime based on t0 and make index
    df["datetime"] = pd.to_datetime(t0, utc=False) + timedelta
    df.set_index("datetime", inplace=True)

    if ventilation_times is None:
        ventilation_times = (6, 20)

    # add ventilation adjustments, based on datetime conditions, flip sign
    df["P_dem"] = adjust_load_profile(df["P_dem"], ventilation_times=ventilation_times)
    df["P_dem"] *= -1

    # pv may have missing entries in PV, which cause issues with updated CSV source behavior
    df["P_ren"] = df["P_ren"].fillna(0)

    # set timestamp time as index
    df.reset_index(inplace=True)
    df.set_index("time", inplace=True)

    if output_file:
        df.round(4).to_csv(output_file, sep=',', )

    return df


if __name__ == "__main__":
    import sys

    if len(sys.argv) > 5 or len(sys.argv) < 3:
        print("script requires two arguments: 1.) the path to the scenario 2.) output csv file")
        print("a third argument is optional for specifiy ventilation start and end time in utc, specified as tuple x,y without a space")
        exit()

    path = sys.argv[1]
    output = sys.argv[2]
    if len(sys.argv) > 3:
        try:
            ventilation_times = sys.argv[3].split(",")
            ventilation_times = (float(ventilation_times[0]), float(ventilation_times[1]))
        except:
            print(f"argument 3 wrongly formatted: {sys.argv[3]}, must be x,y with x and y being int or float")
            exit()
    else:
        ventilation_times = None

    scenario_path = Path(path)
    if not scenario_path.exists():
        print(f"Scenario doesn't exist at given path {scenario_path.absolute()}")
        exit()

    generate_bems_profile(scenario_path, output_file=Path(output), ventilation_times=ventilation_times)