run_simstadt_workflow.py

r"""Tries to find simstadt, find a workflowstep, and run it with the specified
CityGMLs.

It could be useful in order to run the same workflows with many different CityGMLs, or with slightly different
parameters.

User config part will probably need to be updated.
Corresponding workflows can be defined and saved in the SimStadt GUI.


Example output with the current configuration:

    # SimStadt found in C:\Users\eric.duminil\Desktop\SimStadt_0.10.0-SNAPSHOT_master_20230601_607b426

    # Preparing 'Gruenbuehl_PhotovoltaicPotential&Financial' workflow (a.step):
    * Adding Gruenbuehl_LOD2_ALKIS_1010.gml
    * Adding Gruenbuehl_LOD2_ALKIS_1010_2buildings.gml

    # Launching Gruenbuehl_PhotovoltaicPotential&Financial:
    Workflow finished succesfuly!

    # Listing written files :
    * 'Gruenbuehl.proj\a.step\a.step\a.step\a.step\a.step\hourly_GHI_DHI_Ta.prn'
    * 'Gruenbuehl.proj\a.step\a.step\a.step\a.step\a.step\a.step\a.step\a.step\Gruenbuehl_LOD2_ALKIS_1010_2buildings_pvgis_SARAH_2005_2016_Hay_pv_potential.csv'
    * 'Gruenbuehl.proj\a.step\a.step\a.step\a.step\a.step\a.step\a.step\a.step\Gruenbuehl_LOD2_ALKIS_1010_pvgis_SARAH_2005_2016_Hay_pv_potential.csv'


    # Preparing 'Gruenbuehl_HeatDemand' workflow (b.step):
    * Adding Gruenbuehl_LOD2_ALKIS_1010.gml
    * Adding Gruenbuehl_LOD2_ALKIS_1010_2buildings.gml

    # Launching Gruenbuehl_HeatDemand:
    Workflow finished succesfuly!

    # Listing written files :
    * 'Gruenbuehl.proj\b.step\a.step\a.step\a.step\a.step\hourly_GHI_DHI_Ta.prn'
    * 'Gruenbuehl.proj\b.step\a.step\a.step\a.step\a.step\a.step\a.step\Gruenbuehl_LOD2_ALKIS_1010_2buildings_DIN18599_HEATING.csv'
    * 'Gruenbuehl.proj\b.step\a.step\a.step\a.step\a.step\a.step\a.step\Gruenbuehl_LOD2_ALKIS_1010_2buildings_DIN18599_HEATING.log'
    * 'Gruenbuehl.proj\b.step\a.step\a.step\a.step\a.step\a.step\a.step\Gruenbuehl_LOD2_ALKIS_1010_DIN18599_HEATING.csv'
    * 'Gruenbuehl.proj\b.step\a.step\a.step\a.step\a.step\a.step\a.step\Gruenbuehl_LOD2_ALKIS_1010_DIN18599_HEATING.log'
"""

from pathlib import Path
import logging
import platform
import subprocess
import os
import sys
from xml.etree import ElementTree as et

logging.basicConfig(format='%(message)s')
# NOTE: Can set to logging.DEBUG for more info
logging.getLogger().setLevel(logging.INFO)

SCRIPT_DIR = Path(__file__).parent

# TODO: Possibility to change param in params.xml
# TODO: Possibility to change attribute in physics or usage library
# TODO: Possibility to change attribute in CityGML
# TODO: Parse CSV output

##############################################
# User config

SIMSTADT_GLOB = 'Desktop/SimStadt_0.*/'

REPO_PATH = Path.home() / 'git' / 'simstadt' / 'TestRepository'
PROJECT_NAME = 'Gruenbuehl'
WORKFLOW_FOLDERS = ['a.step', 'b.step']
CITYGMLS = ['Gruenbuehl_LOD2_ALKIS_1010.gml', 'Gruenbuehl_LOD2_ALKIS_1010_2buildings.gml']


##############################################

PARAMS = 'params.xml'


# TODO: write tests
def find_simstadt(simstadt_glob: str) -> Path:
    try:
        found = next(Path.home().glob(simstadt_glob))
        logging.info("# SimStadt found in %s\n", found)
        return found
    except StopIteration:
        sys.exit(f"Sorry, no SimStadt installation could be found in {simstadt_glob}")


def check_paths(repo_path: Path, workflow_path: Path):
    if not repo_path.exists():
        sys.exit(f"Sorry, no Repository could be found in {repo_path}")

    if not workflow_path.exists():
        sys.exit(f"Sorry, no workflow could be found in {workflow_path}")

    if not (workflow_path / PARAMS).exists():
        sys.exit(f"Sorry, no workflow is defined in {workflow_path / PARAMS}")

    proj_path = workflow_path.parent

    for gml in proj_path.glob('*.gml'):
        logging.debug("  * Available citygml : %s", gml.name)


def prepare_workflow(workflow_path, citygmls):
    if isinstance(citygmls, str):
        citygmls = [citygmls]

    tree = et.parse(workflow_path / PARAMS)
    root = tree.getroot()

    name = root.find(".//void[@property='name']/string").text
    logging.info("# Preparing '%s' workflow (%s):", name, workflow_path.name)

    citygml_node = root.find(".//void[@property='cityGmlFileNames']")

    if citygml_node:
        logging.debug('  Removing old citygmls')
        children = list(citygml_node)

        for add_node in children:
            citygml_node.remove(add_node)
    else:
        logging.debug('  Add new XML node')
        workflow_node = root.find(".//object[@class='eu.simstadt.workflows.CityGmlWorkflow']")
        citygml_node = et.SubElement(workflow_node, 'void')
        citygml_node.set('property', 'cityGmlFileNames')

    for citygml in citygmls:
        logging.info("  * Adding %s", citygml)
        add_node = et.SubElement(citygml_node, 'void')
        add_node.set('method', 'add')
        gml = et.SubElement(add_node, 'string')
        gml.text = citygml

    tree.write(workflow_path / PARAMS)

    return name


def get_all_files(folder):
    files = [f for f in Path(folder).glob('**/*') if f.is_file() and f.name != PARAMS]
    return {f: f.stat().st_mtime for f in files}


def simstadt_script():
    if platform.system().lower() == 'windows':
        return 'SimStadt.bat'
    else:
        return './SimStadt.sh'


def run_simstadt(simstadt_path, workflow_path, name):
    os.chdir(simstadt_path)
    try:
        logging.info("\n# Launching %s:", name)
        result = subprocess.run(' '.join([simstadt_script(), str(workflow_path)]),
                                text=True,
                                capture_output=True,
                                check=True
                                )
        logging.debug(result.stderr)
        logging.info("  Workflow finished succesfuly!\n")
    except subprocess.CalledProcessError:
        # NOTE: Current SimStadt actually never raises an error.
        logging.warning("  Workflow failed!\n")
        logging.info(result.stderr)


def compare_written_files(repo_path, before, after):
    logging.info("# Listing written files :")
    for result_file, modification_time in after.items():
        before_time = before.get(result_file, 0)
        if modification_time > before_time:
            logging.info("  * '%s'", result_file.relative_to(repo_path))
    logging.info("\n")


def run_workflow(simstadt_path: Path, workflow_path: Path, citygmls: list[str]):
    repo_path = workflow_path.parent.parent

    check_paths(repo_path, workflow_path)

    name = prepare_workflow(workflow_path, citygmls)

    before = get_all_files(workflow_path)
    run_simstadt(simstadt_path, workflow_path, name)
    after = get_all_files(workflow_path)

    compare_written_files(repo_path, before, after)


def main(glob: str, repo: Path, project_name: str, workflows: list[str], citygmls: list[str]):
    simstadt_path = find_simstadt(glob)
    project_path = repo / f'{project_name}.proj'

    for workflow_folder in workflows:
        run_workflow(simstadt_path, project_path / workflow_folder, citygmls)


if __name__ == '__main__':
    main(SIMSTADT_GLOB, REPO_PATH, PROJECT_NAME, WORKFLOW_FOLDERS, CITYGMLS)