dag#

RecipeDAG: the recipe dependency graph behind orchestration.

Wraps get_recipe_dependencies and get_output_path so the recipes stay the single source of truth: adding an enrichment recipe to a curation pipeline automatically adds its jobs and its image-cache dependency. Consumed by workflow/Snakefile; imports nothing from snakemake, so the library works without it.

Classes#

`StageNode`	One orchestrated job: a (stage, recipe, admin unit) triple.
`RecipeDAG`	Dependency graph of every job needed to build a terminal recipe.

Functions#

rule_name(→ str)

Snakemake rule name of one job (used by the Snakefile and --forcerun).

Module Contents#

class openplaces.flow.dag.StageNode#: One orchestrated job: a (stage, recipe, admin unit) triple.

openplaces.flow.dag.rule_name(node: StageNode) → str#: Snakemake rule name of one job (used by the Snakefile and –forcerun).

class openplaces.flow.dag.RecipeDAG(target_recipe_id: str, admin_ids: list[str] | None = None)#

Dependency graph of every job needed to build a terminal recipe.

Parameters:

target_recipe_id (str) – Terminal recipe (e.g. ‘US_footprint-cheer-2026’).
admin_ids (list of str, optional) – Admin units in scope. Coarser IDs are used as-is; each node’s admin unit is truncated to its recipe’s save level. None builds a DAG of admin-independent nodes only (auto-discovered references stay unresolved).

nodes() → list[StageNode]#: Every job in the DAG (target included), deduplicated.

stage_nodes(stage: str) → list[StageNode]#: The jobs of one pipeline stage.

output_path(stage: str, recipe_id: str, admin_id=None) → pathlib.Path#: The primary output parquet of one job.

extra_outputs(stage: str, recipe_id: str, admin_id=None) → list[pathlib.Path]#

Secondary declared outputs of one job.

Harmonize steps with save_link and ingest-level entity_links entries both persist an n:m link sidecar at the canonical get_entity_link_path location.

input_paths(stage: str, recipe_id: str, admin_id=None) → list[pathlib.Path]#: The input files of one job: upstream outputs plus link sidecars.

retention(stage: str, recipe_id: str, admin_id=None) → str#: The retention class of one job’s output (drives temp()/protected()).

bucket(recipe_id: str) → str#: The output bucket of a recipe (‘share’ outputs get protected()).

target_paths() → list[pathlib.Path]#: The terminal outputs the workflow must produce (rule all inputs).

plan() → pandas.DataFrame#

Preview which jobs would run and why (library-side, stat-only).

The fast overview for interactive review; the authoritative scheduling decision is Snakemake’s dry run. Reasons, in priority order: ‘output missing’, ‘inputs newer than output’ (mirrors the mtime rerun trigger), ‘upstream scheduled (…)’ (propagated along the DAG edges), or ‘’ for an up-to-date job. One stat per file, no data reads.

Returns:: One row per job: stage, recipe_id, admin_id, output, exists, size_mb, will_run, reason.
Return type:: pd.DataFrame

to_mermaid(collapse_admin: bool | None = None) → str#

Mermaid flowchart source of the job DAG, styled by stage.

Render with IPython.display.Markdown in a mermaid-capable frontend (VS Code notebooks), or paste into mermaid.live.

Parameters:: collapse_admin (bool, optional) – Collapse per-admin jobs into one node per recipe (labeled with the admin-unit count). None (default) auto-collapses when the full graph exceeds 30 nodes.