stranded_assets_combination_script.py

import json
import os
import pandas as pd
import re

from typing import Any

GRID_FILENAME_REGEX = re.compile(
    r"stranded_assets_from_(?P<service_hours>\d*)_to_grid_(?P<grid_hours>\d*)_priority_(?P<grid_priority>GP|SG)_hpc_run_.*_info_file.json"
)
GRID_REGEX = re.compile(r"BW_UP_(?P<grid_hours>\d*)_(?P<grid_priority>.*)")


def parse_optimisation_data_dict(filedata: dict[str, Any]) -> dict[str, Any]:
    temp_dict = filedata["system_appraisals"]["iteration_0"]["cumulative_results"]
    temp_dict.update(
        filedata["system_appraisals"]["iteration_0"]["environmental_appraisal"]
    )
    temp_dict.update(
        filedata["system_appraisals"]["iteration_0"]["technical_appraisal"]
    )
    temp_dict.update(
        filedata["system_appraisals"]["iteration_0"]["financial_appraisal"]
    )
    temp_dict.update(filedata["system_appraisals"]["iteration_0"]["criteria"])
    temp_dict.update(filedata["system_appraisals"]["iteration_0"]["system_details"])
    temp_dict.pop("input_files")
    grid_match = GRID_REGEX.match(filedata["scenario"]["name"])
    temp_dict["prioritise_self_generation"] = (
        False if grid_match.group("grid_priority") == "GP" else True
    )
    temp_dict["grid_hours"] = (grid_hours:=float(grid_match.group("grid_hours")))
    if grid_hours <= 9:
        import pdb

        pdb.set_trace()
    return temp_dict


def parse_simulation_data_dict(
    filedata: dict[str, Any], filename: str
) -> dict[str, Any]:
    temp_dict = filedata["simulation_1"]["system_appraisal"]["environmental_appraisal"]
    temp_dict.update(
        filedata["simulation_1"]["system_appraisal"]["technical_appraisal"]
    )
    temp_dict.update(
        filedata["simulation_1"]["system_appraisal"]["financial_appraisal"]
    )
    temp_dict.update(filedata["simulation_1"]["system_appraisal"]["criteria"])
    grid_match = GRID_FILENAME_REGEX.match(filename)
    # Try the alternative naming scheme if this files.
    if grid_match is None:
        alternative_regex = re.compile(
            r"(?P<grid_hours>\d*)_hour_grid_only_info_file.json"
        )
        grid_match = alternative_regex.match(filename)

    try:
        temp_dict["prioritise_self_generation"] = (
            True if grid_match.group("grid_priority") == "SG" else False
        )
    except IndexError:
        # Assume grid-priority if not specified.
        temp_dict["prioritise_self_generation"] = False

    temp_dict["grid_hours"] = float(grid_match.group("grid_hours"))
    try:
        temp_dict["service_hours"] = float(grid_match.group("service_hours"))
    except IndexError:
        # Assume grid-priority if not specified.
        temp_dict["service_hours"] = temp_dict["grid_hours"]

    return temp_dict


def generate_combined_results_frame(results_dirname: str) -> pd.DataFrame:
    filename_to_data_map: dict[str, Any] = {}
    for dirname in os.listdir(
        (base_dir := os.path.join(results_dirname, "optimisation_outputs"))
    ):
        with open(
            (filename := os.path.join(base_dir, dirname, "optimisation_output_1.json")),
            "r",
            encoding="UTF-8",
        ) as f:
            file_data = json.load(f)
            filename_to_data_map[filename] = parse_optimisation_data_dict(file_data)

    filename_to_data_frame = pd.DataFrame(filename_to_data_map)
    filename_to_data_frame = filename_to_data_frame.transpose()
    return filename_to_data_frame


def generate_combined_simulations_results_frame(results_dirname: str) -> pd.DataFrame:
    filename_to_data_map = dict()
    for filename in os.listdir(
        (sub_dirname := os.path.join(results_dirname, "simulation_results"))
    ):
        # Skip directories.
        if not os.path.isfile(filepath := os.path.join(sub_dirname, filename)):
            continue

        # Parse the contents and include.
        with open(filepath, "r", encoding="UTF-8") as f:
            file_data = json.load(f)
            filename_to_data_map[filename] = parse_simulation_data_dict(
                file_data, filename
            )

    filename_to_data_frame = pd.DataFrame(filename_to_data_map)
    filename_to_data_frame = filename_to_data_frame.transpose()
    return filename_to_data_frame


def main():
    # Parse optimisation results
    for results_dirname in [
        "stranded_assets_results_3",
        "stranded_assets_results_4_unsubsidised_grid",
    ]:
        data_frame = generate_combined_results_frame(results_dirname)
        with open(f"combined_{results_dirname}.csv", "w", encoding="UTF-8") as f:
            data_frame.to_csv(f)

    # Parse the grid-only simulations
    combined_stranded_assets_results = generate_combined_simulations_results_frame(
        "stranded_assets_results_from_13_hour_grid"
    )
    with open(
        "combined_stranded_assets_results_from_13_hour_grid_simulations.csv",
        "w",
        encoding="UTF-8",
    ) as f:
        combined_stranded_assets_results.to_csv(f)

    combined_grid_results = generate_combined_simulations_results_frame(
        "stranded_assets_grid_only_results"
    )
    with open(
        "combined_stranded_assets_grid_only_results.csv", "w", encoding="UTF-8"
    ) as f:
        combined_grid_results.to_csv(f)


if __name__ == "__main__":
    main()