Proof of concept: run Simstadt from Python

python_scripts/run_simstadt_from_python/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 = SCRIPT_DIR.parent.parent / 'simstadt' / 'TestRepository'
+PROJECT_NAME = 'Gruenbuehl'
+WORKFLOW_FOLDERS = ['a.step', 'b.step']
+CITYGMLS = ['Gruenbuehl_LOD2_ALKIS_1010.gml', 'Gruenbuehl_LOD2_ALKIS_1010_2buildings.gml']
+
+
+##############################################
+
+PROJECT_PATH = REPO_PATH / f'{PROJECT_NAME}.proj'
+PARAMS = 'params.xml'
+
+
+# TODO: write tests
+def find_simstadt(simstadt_glob):
+    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, workflow_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):
+    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, workflow_folder, citygmls):
+    workflow_path = PROJECT_PATH / workflow_folder
+
+    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, workflows, citygmls):
+    simstadt_path = find_simstadt(glob)
+
+    for workflow_folder in workflows:
+        run_workflow(simstadt_path, workflow_folder, citygmls)
+
+
+if __name__ == '__main__':
+    main(SIMSTADT_GLOB, WORKFLOW_FOLDERS, CITYGMLS)