Commit e2afa436 authored by Eric Duminil's avatar Eric Duminil
Browse files

Updated logic, hopefully more flexible

parent 0614a5b7
...@@ -14,6 +14,7 @@ import platform ...@@ -14,6 +14,7 @@ import platform
import re import re
import shutil import shutil
import subprocess import subprocess
import tempfile
from pathlib import Path from pathlib import Path
from xml.etree import ElementTree as et from xml.etree import ElementTree as et
...@@ -24,6 +25,7 @@ from .utils import chdir, random_id ...@@ -24,6 +25,7 @@ from .utils import chdir, random_id
PARAMS = "params.xml" PARAMS = "params.xml"
SIMSTADT2_GLOB = "SimStadt2_0.*/" SIMSTADT2_GLOB = "SimStadt2_0.*/"
SIMSTADT_TEMP_REPO = Path(tempfile.gettempdir()) / "simstadt_repo"
load_dotenv() load_dotenv()
...@@ -218,6 +220,37 @@ def run_workflow(workflow_path: Path, citygmls: list[str]) -> list[Path]: ...@@ -218,6 +220,37 @@ def run_workflow(workflow_path: Path, citygmls: list[str]) -> list[Path]:
return _compare_written_files(repo_path, before, after) return _compare_written_files(repo_path, before, after)
def _is_in_proj_folder(path: Path) -> bool:
"""Return True if path is inside a .proj directory."""
return any(p.suffix == ".proj" for p in path.parents)
def _resolve_project(citygml_path: Path, project_path: Path | None) -> tuple[Path, str]:
"""Return (project_path, citygml_filename) based on CityGML location.
- CityGML inside a .proj folder: use parent as project, reference by name.
- project_path given: copy CityGML there if not already present.
- Neither: create a temporary repository under /tmp (not auto-cleaned).
"""
if _is_in_proj_folder(citygml_path):
return citygml_path.parent, citygml_path.name
if project_path is not None:
dest = project_path / citygml_path.name
if not dest.exists():
shutil.copy(citygml_path, dest)
log.info("Copied %s to %s", citygml_path.name, project_path)
return project_path, citygml_path.name
tmp_proj = SIMSTADT_TEMP_REPO / (citygml_path.stem + ".proj")
tmp_proj.mkdir(parents=True, exist_ok=True)
dest = tmp_proj / citygml_path.name
if not dest.exists():
shutil.copy(citygml_path, dest)
log.info("Using temporary repository at %s", SIMSTADT_TEMP_REPO)
return tmp_proj, citygml_path.name
def run_workflow_with_citygml( def run_workflow_with_citygml(
template: str | Path, template: str | Path,
citygml_path: Path, citygml_path: Path,
...@@ -234,10 +267,13 @@ def run_workflow_with_citygml( ...@@ -234,10 +267,13 @@ def run_workflow_with_citygml(
replaces: Optional dict of XML-fragment regex replacements for params.xml. replaces: Optional dict of XML-fragment regex replacements for params.xml.
description: Human-readable label for the result object. description: Human-readable label for the result object.
destination: Output workflow folder name; defaults to a timestamped random id. destination: Output workflow folder name; defaults to a timestamped random id.
project_path: Directory where the workflow folder is created. Defaults to project_path: Directory where the workflow folder is created.
citygml_path.parent. When set to a different directory, citygml_path is - If CityGML is already inside a .proj folder, project_path is ignored.
passed as an absolute path so SimStadt can locate it. - If given, CityGML is copied there.
- If omitted, a temporary repository is created under /tmp.
""" """
citygml_path = Path(citygml_path)
if isinstance(template, str): if isinstance(template, str):
template_path = get_template_path() / template template_path = get_template_path() / template
else: else:
...@@ -247,21 +283,9 @@ def run_workflow_with_citygml( ...@@ -247,21 +283,9 @@ def run_workflow_with_citygml(
if destination is None: if destination is None:
destination = random_id() + "_" + template_name.rsplit("_", 1)[-1] destination = random_id() + "_" + template_name.rsplit("_", 1)[-1]
if project_path is None: resolved_project_path, citygml_filename = _resolve_project(citygml_path, project_path)
project_path = citygml_path.parent workflow_path = copy_workflow_from_template(template_path, resolved_project_path, destination, replaces)
citygml_ref = citygml_path.name output_files = run_workflow(workflow_path, [citygml_filename])
else:
repo_path = project_path.parent
try:
citygml_ref = str(citygml_path.relative_to(repo_path))
except ValueError as e:
raise ValueError(
f"CityGML file {citygml_path} is not under the repository root {repo_path}. "
"Place the GML file inside the repository directory or omit project_path."
) from e
workflow_path = copy_workflow_from_template(template_path, project_path, destination, replaces)
output_files = run_workflow(workflow_path, [citygml_ref])
if description is None: if description is None:
description = f"{template_name} for {citygml_path.name}" description = f"{template_name} for {citygml_path.name}"
......
import shutil
from pathlib import Path from pathlib import Path
from xml.etree import ElementTree as et
import pytest import pytest
...@@ -16,6 +18,46 @@ def test_project_path() -> Path: ...@@ -16,6 +18,46 @@ def test_project_path() -> Path:
return TEST_PROJECT_PATH return TEST_PROJECT_PATH
# ---------------------------------------------------------------------------
# SimStadt spoof (replaces subprocess call with fixture file copying)
# ---------------------------------------------------------------------------
def _find_fixture_flow(provider: str) -> Path | None:
"""Find a .flow directory in test data that matches the given provider and has output files."""
for flow_dir in TEST_REPOSITORY.rglob("*.flow"):
params_xml = flow_dir / "params.xml"
if not params_xml.exists():
continue
root = et.parse(params_xml).getroot()
elem = root.find(".//void[@property='workflowProvider']/object")
if elem is not None and elem.get("class") == provider:
non_params = [f for f in flow_dir.rglob("*") if f.is_file() and f.name != "params.xml"]
if non_params:
return flow_dir
return None
@pytest.fixture
def spoof_simstadt(monkeypatch):
"""Patch run_simstadt to copy fixture output files instead of running SimStadt."""
def _spoof(workflow_path: Path, name: str) -> str:
root = et.parse(workflow_path / "params.xml").getroot()
elem = root.find(".//void[@property='workflowProvider']/object")
provider = elem.get("class") if elem is not None else None
fixture_flow = _find_fixture_flow(provider)
if fixture_flow is None:
raise ValueError(f"No fixture found for provider: {provider}")
for src in fixture_flow.rglob("*"):
if src.is_file() and src.name != "params.xml":
dst = workflow_path / src.relative_to(fixture_flow)
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
return f"Spoofed SimStadt for {name}"
monkeypatch.setattr("simstadt.runner.run_simstadt", _spoof)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Live fixtures (require SimStadt to be installed) # Live fixtures (require SimStadt to be installed)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
......
"""Tests for run_workflow_with_citygml path resolution logic."""
import shutil
from pathlib import Path
import pytest
from simstadt.runner import run_workflow_with_citygml
TEST_DATA = Path(__file__).parent / "data"
CITYGML = TEST_DATA / "TestRepo" / "MiniBuchwald.gml"
TEMPLATE = TEST_DATA / "Templates" / "01_HeatDemand"
def test_citygml_in_proj_folder(spoof_simstadt, tmp_path):
"""CityGML already inside a .proj folder — used as-is, nothing copied."""
proj = tmp_path / "Test.proj"
proj.mkdir()
gml = proj / CITYGML.name
shutil.copy(CITYGML, gml)
results = run_workflow_with_citygml(TEMPLATE, gml)
assert results is not None
assert CITYGML.name in results.citygml
assert results.workflow_path.parent == proj
def test_citygml_with_explicit_project_path(spoof_simstadt, tmp_path):
"""CityGML outside .proj with project_path — CityGML is copied to project_path."""
proj = tmp_path / "Test.proj"
proj.mkdir()
results = run_workflow_with_citygml(TEMPLATE, CITYGML, project_path=proj)
assert (proj / CITYGML.name).exists()
assert results is not None
assert results.workflow_path.parent == proj
def test_citygml_with_explicit_project_path_no_double_copy(spoof_simstadt, tmp_path):
"""CityGML already present in project_path — not copied again."""
proj = tmp_path / "Test.proj"
proj.mkdir()
dest = proj / CITYGML.name
shutil.copy(CITYGML, dest)
mtime_before = dest.stat().st_mtime
run_workflow_with_citygml(TEMPLATE, CITYGML, project_path=proj)
assert dest.stat().st_mtime == mtime_before
def test_citygml_creates_temp_repo(spoof_simstadt):
"""CityGML outside .proj with no project_path — temporary repository created."""
results = run_workflow_with_citygml(TEMPLATE, CITYGML)
assert results is not None
assert "simstadt_repo" in str(results.workflow_path)
assert results.workflow_path.suffix == ".flow"
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment