Functions
list	normalize (list paths, Path workspace, str run_id)
	Replace unstable path components with logical tokens.

dict	run_plan (Path workspace, Path case_dir, str monitor, list extra, str post="post.yml")
	Invoke the dry-run planner for one scenario and return its raw plan.

Path	init_case (Path workspace)
	Initialize a shipped case inside a scenario workspace.

str	flat_post_recipe (Path case_dir)
	Write a post recipe requesting a non-canonical output directory.

dict	map_to_identities (list artifacts, dict contract)
	Map each normalized artifact token onto a declared logical identity.

dict	build_snapshot ()
	Build the normalized topology snapshot across scenarios.

int	main ()
	Write or verify the artifact topology snapshot.

Variables
	REPO_ROOT = Path(__file__).resolve().parents[2]

str	CONTRACT_PATH = REPO_ROOT / "tests" / "tooling" / "artifact_topology.json"

str	SNAPSHOT_PATH = REPO_ROOT / "docs" / "generated" / "artifact_topology_snapshot.json"

str	PICURV = REPO_ROOT / "picurv_cli" / "picurv"

Detailed Description

Extract a normalized run-artifact topology snapshot from the CLI's own planner.

Function Documentation

◆ normalize()

list extract_artifact_topology.normalize	(	list	paths,
		Path	workspace,
		str	run_id
	)

Replace unstable path components with logical tokens.

The workspace root, the generated run id, and any embedded timestamp differ on every invocation. Fingerprinting the raw plan would therefore report drift on every run; fingerprinting the normalized shape reports drift only when the topology actually changes.

Parameters

[in]	paths	Planned artifact paths.
[in]	workspace	Temporary workspace root used for the plan.
[in]	run_id	Generated run identifier.

Returns: Sorted, normalized artifact tokens.

Definition at line 23 of file extract_artifact_topology.py.

def normalize(paths: list, workspace: Path, run_id: str) -> list:
    """!
    @brief Replace unstable path components with logical tokens.
 
    The workspace root, the generated run id, and any embedded timestamp differ on
    every invocation. Fingerprinting the raw plan would therefore report drift on
    every run; fingerprinting the normalized shape reports drift only when the
    topology actually changes.
    @param[in] paths Planned artifact paths.
    @param[in] workspace Temporary workspace root used for the plan.
    @param[in] run_id Generated run identifier.
    @return Sorted, normalized artifact tokens.
    """
    normalized = set()
    for raw in paths:
        text = str(raw)
        text = text.replace(str(workspace), "<workspace>")
        text = text.replace(run_id, "<run_id>")
        text = re.sub(r"\d{8}-\d{6}", "<timestamp>", text)
        normalized.add(text)
    return sorted(normalized)
 
 

Here is the caller graph for this function:

◆ run_plan()

dict extract_artifact_topology.run_plan	(	Path	workspace,
		Path	case_dir,
		str	monitor,
		list	extra,
		str	post = `"post.yml"`
	)

Invoke the dry-run planner for one scenario and return its raw plan.

Parameters

[in]	workspace	Scenario workspace.
[in]	case_dir	Directory holding the initialized case.
[in]	monitor	Monitor file name inside the case directory.
[in]	extra	Additional CLI arguments for the scenario.
[in]	post	Post recipe file name inside the case directory.

Returns: Parsed plan mapping.

Exceptions

RuntimeError when the planner fails.

Definition at line 46 of file extract_artifact_topology.py.

             post: str = "post.yml") -> dict:
    """!
    @brief Invoke the dry-run planner for one scenario and return its raw plan.
    @param[in] workspace Scenario workspace.
    @param[in] case_dir Directory holding the initialized case.
    @param[in] monitor Monitor file name inside the case directory.
    @param[in] extra Additional CLI arguments for the scenario.
    @param[in] post Post recipe file name inside the case directory.
    @return Parsed plan mapping.
    @throws RuntimeError when the planner fails.
    """
    result = subprocess.run(
        [
            sys.executable, str(PICURV), "run", "--dry-run", "--format", "json",
            "--case", str(case_dir / "config" / "case.yml"),
            "--solver", str(case_dir / "config" / "solver.yml"),
            "--monitor", str(case_dir / "config" / monitor),
            "--post", str(case_dir / "config" / post),
            *extra,
        ],
        cwd=workspace, capture_output=True, text=True, check=False, timeout=180,
    )
    if result.returncode != 0:
        raise RuntimeError(f"dry-run failed:\n{result.stdout}\n{result.stderr}")
    return json.loads(result.stdout)
 
 

Here is the caller graph for this function:

◆ init_case()

Path extract_artifact_topology.init_case ( Path workspace )

Initialize a shipped case inside a scenario workspace.

Parameters

[in] workspace Scenario workspace.

Returns: Path to the created case directory.

Exceptions

RuntimeError when initialization fails.

Definition at line 74 of file extract_artifact_topology.py.

def init_case(workspace: Path) -> Path:
    """!
    @brief Initialize a shipped case inside a scenario workspace.
    @param[in] workspace Scenario workspace.
    @return Path to the created case directory.
    @throws RuntimeError when initialization fails.
    """
    case_dir = workspace / "case"
    result = subprocess.run(
        [sys.executable, str(PICURV), "init", "flat_channel", "--dest", str(case_dir)],
        cwd=workspace, capture_output=True, text=True, check=False, timeout=120,
    )
    if result.returncode != 0:
        raise RuntimeError(f"picurv init failed:\n{result.stdout}\n{result.stderr}")
    return case_dir
 
 

Here is the caller graph for this function:

◆ flat_post_recipe()

str extract_artifact_topology.flat_post_recipe ( Path case_dir )

Write a post recipe requesting a non-canonical output directory.

Runtime path injection must replace this request with the stable output/visualization/<recipe_id> home.

Parameters

[in] case_dir Case directory to write into.

Returns: File name of the variant recipe.

Definition at line 91 of file extract_artifact_topology.py.

def flat_post_recipe(case_dir: Path) -> str:
    """!
    @brief Write a post recipe requesting a non-canonical output directory.
 
    @details Runtime path injection must replace this request with the stable
             `output/visualization/<recipe_id>` home.
    @param[in] case_dir Case directory to write into.
    @return File name of the variant recipe.
    """
    source = (case_dir / "config" / "post.yml").read_text(encoding="utf-8")
    source = re.sub(r'^(\s*)output_directory:\s*.*$', r'\1output_directory: "viz"',
                    source, flags=re.M)
    (case_dir / "config" / "Flat_Viz_Analysis.yml").write_text(source, encoding="utf-8")
    return "Flat_Viz_Analysis.yml"
 
 

Here is the caller graph for this function:

◆ map_to_identities()

dict extract_artifact_topology.map_to_identities	(	list	artifacts,
		dict	contract
	)

Map each normalized artifact token onto a declared logical identity.

An artifact the contract does not name is reported rather than ignored: an unmapped path is a documented-layout gap, which is exactly what a topology contract exists to surface.

Parameters

[in]	artifacts	Normalized artifact tokens.
[in]	contract	Parsed topology contract.

Returns: Mapping of identity id to matched tokens, plus an unmapped list.

Definition at line 107 of file extract_artifact_topology.py.

def map_to_identities(artifacts: list, contract: dict) -> dict:
    """!
    @brief Map each normalized artifact token onto a declared logical identity.
 
    An artifact the contract does not name is reported rather than ignored: an
    unmapped path is a documented-layout gap, which is exactly what a topology
    contract exists to surface.
    @param[in] artifacts Normalized artifact tokens.
    @param[in] contract Parsed topology contract.
    @return Mapping of identity id to matched tokens, plus an `unmapped` list.
    """
    logical_roots = {
        "<run.root>": r"<workspace>/runs/<run_id>",
        "<run.config>": r"<workspace>/runs/<run_id>/config",
        "<run.post_recipes>": r"<workspace>/runs/<run_id>/config/post\-recipes",
        "<run.inputs>": r"<workspace>/runs/<run_id>/inputs",
        "<run.runtime_logs>": r"<workspace>/runs/<run_id>/logs",
        "<run.scheduler>": r"<workspace>/runs/<run_id>/scheduler",
        "<run.solver_output>": r"<workspace>/runs/<run_id>/output",
        "<run.analysis>": r"<workspace>/runs/<run_id>/output/analysis",
        "<run.visualization>": r"<workspace>/runs/<run_id>/output/visualization",
    }
    rules = []
    for record in contract["artifacts"]:
        rule = record["path_rule"]
        pattern = re.escape(rule)
        pattern = pattern.replace(re.escape("<workspace>"), r"<workspace>")
        for token, resolved in logical_roots.items():
            pattern = pattern.replace(re.escape(token), resolved)
        pattern = pattern.replace(re.escape("<role>"), r"[^/]+")
        pattern = pattern.replace(re.escape("<ext>"), r"[^/]+")
        pattern = pattern.replace(re.escape("<name>"), r"[^/]+")
        pattern = pattern.replace(re.escape("<recipe>"), r"[^/]+")
        pattern = pattern.replace(re.escape("<run_id>"), r"<run_id>")
        pattern = pattern.replace(re.escape("<n>"), r"[^/]+")
        rules.append((record["id"], re.compile("^" + pattern + "$")))
 
    # Specific rules must win over wildcard ones. A configurable directory such as
    # `<run.root>/<configured log dir>` matches any single segment, so first-match
    # order let it swallow manifest.json, output, and scheduler.
    def specificity(item) -> tuple:
        """!
        @brief Rank a rule so literal path rules are tried before wildcard ones.
        @param[in] item Tuple of identity id and compiled pattern.
        @return Sort key placing more specific rules first.
        """
        pattern = item[1].pattern
        return (pattern.count("[^/]+"), -len(pattern))
 
    ordered = sorted(rules, key=specificity)
    mapped: dict = {rid: [] for rid, _ in rules}
    unmapped = []
    for token in artifacts:
        for rid, pattern in ordered:
            if pattern.match(token):
                mapped[rid].append(token)
                break
        else:
            unmapped.append(token)
    return {"mapped": {k: sorted(v) for k, v in mapped.items() if v}, "unmapped": sorted(unmapped)}
 
 

Here is the caller graph for this function:

◆ build_snapshot()

dict extract_artifact_topology.build_snapshot ( )

Build the normalized topology snapshot across scenarios.

Returns: Snapshot mapping.

Definition at line 169 of file extract_artifact_topology.py.

def build_snapshot() -> dict:
    """!
    @brief Build the normalized topology snapshot across scenarios.
    @return Snapshot mapping.
    """
    contract = json.loads(CONTRACT_PATH.read_text(encoding="utf-8"))
    scenarios = []
    workspace = Path(tempfile.mkdtemp(prefix="picurv-topology-"))
    try:
        case_dir = init_case(workspace)
        flat_post = flat_post_recipe(case_dir)
        cases = [
            ("fresh_local_solve_and_post", "monitor.yml",
             ["--solve", "--post-process"], "post.yml"),
            ("fresh_local_solve_only", "monitor.yml", ["--solve"],
             "post.yml"),
            # A non-canonical request proves the runtime path is still fixed.
            ("post_output_request_is_canonicalized", "monitor.yml",
             ["--solve", "--post-process"], flat_post),
        ]
        for name, monitor, extra, post in cases:
            plan = run_plan(workspace, case_dir, monitor, extra, post)
            artifacts = normalize(plan.get("artifacts", []), case_dir, plan["run_id_preview"])
            scenarios.append({
                "scenario": name,
                "launch_mode": plan.get("launch_mode"),
                "artifacts": artifacts,
                "identities": map_to_identities(artifacts, contract),
            })
    finally:
        shutil.rmtree(workspace, ignore_errors=True)
    return {
        "default_layout": contract["default_layout"],
        "isolation_enforced": False,
        "logical_artifacts": [a["id"] for a in contract["artifacts"]],
        "scenarios": scenarios,
    }
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

◆ main()

int extract_artifact_topology.main ( )

Write or verify the artifact topology snapshot.

Returns: Process status code.

Definition at line 208 of file extract_artifact_topology.py.

def main() -> int:
    """!
    @brief Write or verify the artifact topology snapshot.
    @return Process status code.
    """
    parser = argparse.ArgumentParser(description="Extract the run artifact topology snapshot.")
    parser.add_argument("--check", action="store_true", help="Fail if the snapshot is stale.")
    args = parser.parse_args()
 
    try:
        snapshot = build_snapshot()
    except (RuntimeError, subprocess.TimeoutExpired, json.JSONDecodeError) as error:
        print(f"Artifact topology extraction failed: {error}", file=sys.stderr)
        return 1
 
    content = json.dumps(snapshot, indent=2, sort_keys=True) + "\n"
    if args.check:
        if not SNAPSHOT_PATH.is_file() or SNAPSHOT_PATH.read_text(encoding="utf-8") != content:
            print(
                "Run artifact topology has changed. The planned artifact set no longer matches "
                "the recorded snapshot.\n"
                "  Review the pages that document run layout, then refresh with:\n"
                "    make docs-topology",
                file=sys.stderr,
            )
            return 1
        print(f"Artifact topology snapshot is current ({len(snapshot['scenarios'])} scenario(s)).")
        return 0
 
    SNAPSHOT_PATH.parent.mkdir(parents=True, exist_ok=True)
    SNAPSHOT_PATH.write_text(content, encoding="utf-8")
    total = sum(len(s["artifacts"]) for s in snapshot["scenarios"])
    print(f"Wrote artifact topology snapshot: {len(snapshot['scenarios'])} scenario(s), {total} artifacts.")
    return 0
 
 

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

◆ REPO_ROOT

extract_artifact_topology.REPO_ROOT = Path(__file__).resolve().parents[2]

Definition at line 17 of file extract_artifact_topology.py.

◆ CONTRACT_PATH

str extract_artifact_topology.CONTRACT_PATH = REPO_ROOT / "tests" / "tooling" / "artifact_topology.json"

Definition at line 18 of file extract_artifact_topology.py.

◆ SNAPSHOT_PATH

str extract_artifact_topology.SNAPSHOT_PATH = REPO_ROOT / "docs" / "generated" / "artifact_topology_snapshot.json"

Definition at line 19 of file extract_artifact_topology.py.

◆ PICURV

str extract_artifact_topology.PICURV = REPO_ROOT / "picurv_cli" / "picurv"

Definition at line 20 of file extract_artifact_topology.py.

Functions

Variables

Detailed Description

Function Documentation

◆ normalize()

◆ run_plan()

◆ init_case()

◆ flat_post_recipe()

◆ map_to_identities()

◆ build_snapshot()

◆ main()

Variable Documentation

◆ REPO_ROOT

◆ CONTRACT_PATH

◆ SNAPSHOT_PATH

◆ PICURV