"""Shared CLI ↔ GUI artifact-layout constants.
Single source of truth for everything written to disk by the forward CLI,
the SLURM recompile worker, and the chunk writer — and read back by the
GUI viewer / analysis sub-app. Both subpackages should import from here
rather than re-spelling. This module replaces the scattered inline
``"master_measurements.parquet"`` / ``"progress"`` / etc. literals that
previously lived as private constants in ~10 different files.
Module layout
-------------
* **Filenames** (`MASTER_MEASUREMENTS_*`, `PIPELINE_JSON`, …) — bare strings
written/read by both producer (CLI) and consumer (GUI).
* **Directory names** (`DIR_*`) — same.
* **Templated filenames** — private ``_FOO_TEMPLATE: Final[str]``
constants paired with public ``foo_filename(...)`` render functions.
The render function's typed signature is the public API; the template
itself is private. This is the project's first deployment of the
CLAUDE.md "parameterized strings are not enumerations" pattern.
* **Path helpers** — `progress_dir(output)`, `event_log_path(output)`,
etc. Take a base ``output_dir: Path`` and return the canonical
artifact path. Replace ``output_dir / "literal"`` constructions
scattered across the CLI. Two parameter conventions are used:
- **`output_dir: Path`** — the run output root (e.g. ``./out``).
Use these from any caller that has the run root in scope:
``master_measurements_parquet_path``, ``manifest_json_path``,
``job_metadata_path``, ``pipeline_json_path``, ``task_status_path``,
``logs_dir``, ``slurm_scripts_dir``, ``processing_report_html_path``,
``measurements_by_feature_dir``,
etc.
- **`progress_dir_: Path`** — the already-resolved progress dir
(i.e. ``output_dir / "progress"``). Used for helpers that produce
paths to *internal mid-run artifacts* the SLURM sentinel + chunk
writer hand around between each other:
``analysis_full_parquet_path``, ``sentinel_resubmitted_path``,
``chunk_lock_path``, ``chunks_dir``,
``chunk_parquet_path``, ``checkpoint_lock_path``, ``recompile_dir``,
``recompile_status_dir``. The trailing underscore on the parameter
name disambiguates it from the ``progress_dir(output_dir)``
function.
When in doubt, prefer the ``output_dir``-rooted form — it composes
cleanly with ``progress_dir(output_dir)`` if you need the progress
directory separately. Helpers that take ``progress_dir_`` are noted
in their individual docstrings.
* **Reader helpers** — `read_run_manifest` and `resolve_execution_mode`
consolidate two high-frequency duplicates.
* **JSON contract keys** (`JobMetadataKey`, `DashboardManifestKey`,
`ChunkStateKey`, `ChunkManifestKey`) — namespace classes
whose class-level ``Final[str]`` attributes are the keys writers and
readers must reference instead of bare strings. Keeps the contract
on-disk format mechanically discoverable.
* **`ModulePath`** — importable module paths used in ``importlib.import_module``
dispatch by the analysis GUI's recipe loader.
* **`EnvVar`** — environment variable names read by the CLI (SLURM /
scratch).
See also
--------
:mod:`phenotypic.sdk_.constants_`
Image-data and framework-config enums (``IMAGE_MODE``, ``IMAGE_TYPES``,
``GAMMA_ENCODINGS``, ``PIPE_STATUS``). The ``IMAGE`` enum and the
experimental-tag vocabulary now live in :mod:`phenotypic.schema`.
:mod:`phenotypic.sdk_.typing_`
Literal aliases for closed value sets used at public boundaries
(``ExecutionMode``, ``ImageTypeName``, …).
"""
from __future__ import annotations
import hashlib
import json
import logging
import os
import re
import stat as stat_module
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Final, Iterable, Optional
from .typing_ import (
CheckpointType,
ExecutionMode,
ImageTypeName,
)
if TYPE_CHECKING:
from phenotypic._core._grid_image import GridImage as _GridImage
from phenotypic._core._image import Image as _Image
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# CLI artifact filenames
# ---------------------------------------------------------------------------
LEGACY_JSON_SUFFIX: Final[str] = ".json"
CONFIG_SUFFIX_PIPELINE: Final[str] = ".json.pht-pipe"
CONFIG_SUFFIX_OPERATION: Final[str] = ".json.pht-op"
CONFIG_SUFFIX_COLOR_CHECKER: Final[str] = ".json.pht-cc"
CONFIG_SUFFIX_TUNING: Final[str] = ".json.pht-tune"
CONFIG_SUFFIXES: Final[frozenset[str]] = frozenset(
{
CONFIG_SUFFIX_PIPELINE,
CONFIG_SUFFIX_OPERATION,
CONFIG_SUFFIX_COLOR_CHECKER,
CONFIG_SUFFIX_TUNING,
}
)
PIPELINE_CONFIG_SUFFIXES: Final[frozenset[str]] = frozenset(
{CONFIG_SUFFIX_PIPELINE, LEGACY_JSON_SUFFIX}
)
TUNING_CONFIG_SUFFIXES: Final[frozenset[str]] = frozenset(
{CONFIG_SUFFIX_TUNING, LEGACY_JSON_SUFFIX}
)
[docs]
def has_config_suffix(path: str | Path, suffixes: Iterable[str]) -> bool:
"""Return whether ``path`` ends with any configured suffix.
Matching is case-insensitive so callers can discover user-provided files
from case-preserving filesystems without rewriting their names.
"""
text = str(path).lower()
return any(text.endswith(suffix.lower()) for suffix in suffixes)
[docs]
def matches_any_suffix(path: str | Path, suffixes: Iterable[str]) -> bool:
"""Return whether ``path`` ends with any suffix in ``suffixes``."""
return has_config_suffix(path, suffixes)
[docs]
def ensure_typed_json_suffix(path: str | Path, suffix: str) -> Path:
"""Return ``path`` with the canonical typed JSON suffix appended.
Bare stems receive the full typed suffix. Legacy ``.json`` paths receive
only the typed tail, preserving the user-provided stem and case.
"""
target = Path(path)
text = str(target)
lowered = text.lower()
canonical_suffix = suffix.lower()
if lowered.endswith(canonical_suffix):
return target
if lowered.endswith(LEGACY_JSON_SUFFIX):
typed_tail = suffix.removeprefix(LEGACY_JSON_SUFFIX)
return Path(f"{text}{typed_tail}")
return Path(f"{text}{suffix}")
[docs]
def bytes_fingerprint(data: bytes) -> str:
"""Return a versioned SHA-256 fingerprint for exact bytes.
Args:
data: Bytes to fingerprint.
Returns:
A ``"sha256:<hex>"`` content fingerprint.
"""
return f"sha256:{hashlib.sha256(data).hexdigest()}"
[docs]
def file_fingerprint(path: Path) -> str:
"""Return a versioned SHA-256 fingerprint for one file's contents.
Args:
path: Existing regular file to fingerprint.
Returns:
A ``"sha256:<hex>"`` content fingerprint.
"""
digest = hashlib.sha256()
with Path(path).open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return f"sha256:{digest.hexdigest()}"
[docs]
def paths_fingerprint(paths: Iterable[Path], *, root: Path | None = None) -> str:
"""Fingerprint an ordered set of named filesystem entries without writes.
Files include their exact contents, directories include their normalized
names, and missing entries are represented explicitly. Sorting by name
makes the result independent of caller enumeration order.
Args:
paths: Files to include.
root: Optional anchor used to normalize names.
Returns:
A deterministic ``"sha256:<hex>"`` fingerprint.
"""
anchor = Path(root).resolve() if root is not None else None
named_paths: list[tuple[str, Path]] = []
for raw_path in paths:
path = Path(raw_path)
resolved = path.resolve(strict=False)
if anchor is not None:
try:
name = resolved.relative_to(anchor).as_posix()
except ValueError:
name = resolved.as_posix()
else:
name = resolved.as_posix()
named_paths.append((name, path))
digest = hashlib.sha256()
for name, path in sorted(named_paths, key=lambda item: item[0]):
encoded_name = name.encode("utf-8")
digest.update(len(encoded_name).to_bytes(8, "big"))
digest.update(encoded_name)
if path.is_dir():
digest.update(b"\x02")
continue
if not path.is_file():
digest.update(b"\x00")
continue
digest.update(b"\x01")
digest.update(path.stat().st_size.to_bytes(8, "big"))
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return f"sha256:{digest.hexdigest()}"
[docs]
def source_cache_key(source: Path, fingerprint: str) -> str:
"""Return an opaque cache key bound to canonical source and content."""
identity = f"{Path(source).resolve(strict=False)}\0{fingerprint}".encode()
return hashlib.sha256(identity).hexdigest()[:32]
[docs]
def migration_backup_dir(config_path: Path) -> Path:
"""Return the dedicated sibling backup directory for a configuration."""
return Path(config_path).parent / ".migration_backups"
[docs]
def pipeline_publication_lock_path(config_path: Path) -> Path:
"""Return the shared interprocess lock path for pipeline publication.
Every production writer of a canonical output pipeline must acquire this
lock before checking a source generation or replacing the file. The lock
is intentionally about publication, not one particular migration, so CLI,
QC, Analysis, and compatibility writers serialize against each other.
Legacy ``pipeline.json`` and canonical ``pipeline.json.pht-pipe`` paths
intentionally map to the same output-level identity. This prevents a V1
reader/writer and a V2 writer from bypassing one another merely because
they selected different compatibility filenames.
"""
config = Path(config_path)
if config.name in {PIPELINE_JSON, _LEGACY_PIPELINE_JSON}:
return config.parent / ".pipeline-config.migration.lock"
return config.with_name(f".{config.name}.migration.lock")
[docs]
def migration_lock_path(config_path: Path) -> Path:
"""Return the shared pipeline publication lock used by migrations.
This compatibility alias preserves the original SDK name while ensuring
migrations coordinate with every ordinary canonical pipeline writer.
"""
return pipeline_publication_lock_path(config_path)
[docs]
def migration_backup_path(
config_path: Path,
*,
timestamp: str,
source_fingerprint: str,
) -> Path:
"""Return a timestamped, fingerprinted backup path."""
digest = source_fingerprint.removeprefix("sha256:")[:12]
return migration_backup_dir(config_path) / (
f"{Path(config_path).name}.{timestamp}.{digest}.bak"
)
[docs]
def migration_receipt_path(
config_path: Path,
*,
resulting_fingerprint: str,
) -> Path:
"""Return the durable receipt path for one migrated generation."""
digest = resulting_fingerprint.removeprefix("sha256:")[:12]
return migration_backup_dir(config_path) / (
f"{Path(config_path).name}.{digest}.migration.json"
)
[docs]
def generation_staging_path(target: Path, generation: str) -> Path:
"""Return a sibling staging path for an explicit publication generation.
Raises:
ValueError: If ``generation`` is not a safe path component.
"""
if (
not generation
or generation in {".", ".."}
or "/" in generation
or "\\" in generation
):
raise ValueError("generation must be a nonempty path component")
target = Path(target)
return target.with_name(f".{target.name}.{generation}.generation")
#: Master archive of all aggregated measurements (clean, pre-post). Written by
#: :func:`phenotypic._cli._cli_finalize_run.finalize_run` as the exact
#: concatenation of the authorized embedded measurement tables: **un-joined**,
#: carrying intrinsic identity only. Treated as the authoritative source by
#: downstream tooling; never edited in place.
#:
#: **Parquet-only since D8.** ``master_measurements.csv`` is gone, along with
#: its path helper, its reader and its entry in the aggregate proof's
#: ``required_outputs``. The un-joined master is no longer the file a human
#: opens -- :data:`MEASUREMENTS_CSV`, which carries the metadata join and the
#: post-applied frame, is -- and the master's dtypes are exactly what a CSV
#: could not preserve.
MASTER_MEASUREMENTS_PARQUET: Final[str] = "master_measurements.parquet"
#: Editable curated CSV mirror seeded by the CLI from the post-applied master
#: frame after :func:`_apply_post_to_master`. The
#: results viewer rewrites this file in place when the user removes/restores
#: colonies. Re-running the CLI overwrites it with a fresh full copy.
MEASUREMENTS_CSV: Final[str] = "measurements.csv"
#: Editable curated Parquet companion to :data:`MEASUREMENTS_CSV`. The
#: parquet is the GUI's source of truth at boot (CSV is the human-readable
#: mirror); both are written atomically together.
MEASUREMENTS_PARQUET: Final[str] = "measurements.parquet"
#: Best-effort co-located copy of the run's ``--metadata`` source CSV,
#: written into :data:`DIR_DELIVERABLES` by
#: :func:`phenotypic._cli._cli_output_manager.finalize_post_master_outputs`.
#: The post-applied mirror carries the metadata columns, while this file
#: preserves the full portable original mapping.
DELIVERABLES_METADATA_CSV: Final[str] = "metadata.csv"
#: Authoritative index for class-named analysis table artifacts.
ANALYSIS_MANIFEST_FILENAME: Final[str] = "analysis_manifest.json"
ANALYSIS_MANIFEST_SCHEMA_VERSION: Final[int] = 1
_ANALYSIS_ID_PATTERN = re.compile(r"[A-Za-z][A-Za-z0-9_.-]{0,127}\Z", re.ASCII)
def _reserved_analysis_artifact_stems() -> frozenset[str]:
"""Return CSV/Parquet stems that share the deliverables root."""
return frozenset(
Path(filename).stem.casefold()
for filename in (
MASTER_MEASUREMENTS_PARQUET,
MEASUREMENTS_CSV,
MEASUREMENTS_PARQUET,
DELIVERABLES_METADATA_CSV,
ERROR_ANALYSIS_CSV,
ERROR_ANALYSIS_PARQUET,
VERIFIED_PARQUET,
)
)
[docs]
def validate_analysis_id(analysis_id: str) -> str:
"""Validate an analysis ID used as a deliverables artifact stem.
Args:
analysis_id: Candidate stable analysis identity.
Returns:
The unchanged validated ID.
Raises:
TypeError: If ``analysis_id`` is not a string.
ValueError: If the ID is unsafe or collides with a canonical table.
"""
if not isinstance(analysis_id, str):
raise TypeError("analysis_id must be a string")
if not _ANALYSIS_ID_PATTERN.fullmatch(analysis_id):
raise ValueError(
"analysis_id must be 1-128 ASCII characters, start with a letter, "
"and contain only letters, digits, '_', '-', or '.'"
)
if analysis_id.casefold() in _reserved_analysis_artifact_stems():
raise ValueError(
f"analysis_id {analysis_id!r} collides with a canonical "
"deliverables table"
)
return analysis_id
[docs]
@dataclass(frozen=True)
class AnalysisArtifactPaths:
"""Concrete paths for one named analysis generation."""
csv: Path
parquet: Path
manifest: Path
[docs]
def analysis_manifest_path(deliverables_base: Path) -> Path:
"""Return the analysis manifest inside a deliverables directory."""
return Path(deliverables_base) / ANALYSIS_MANIFEST_FILENAME
[docs]
def named_analysis_csv_path(deliverables_base: Path, analysis_id: str) -> Path:
"""Return the named CSV artifact path for ``analysis_id``."""
return Path(deliverables_base) / f"{validate_analysis_id(analysis_id)}.csv"
[docs]
def named_analysis_parquet_path(
deliverables_base: Path, analysis_id: str
) -> Path:
"""Return the named Parquet artifact path for ``analysis_id``."""
return (
Path(deliverables_base)
/ f"{validate_analysis_id(analysis_id)}.parquet"
)
[docs]
def named_analysis_paths(
deliverables_base: Path, analysis_id: str
) -> AnalysisArtifactPaths:
"""Return all persisted paths associated with one analysis ID."""
return AnalysisArtifactPaths(
csv=named_analysis_csv_path(deliverables_base, analysis_id),
parquet=named_analysis_parquet_path(deliverables_base, analysis_id),
manifest=analysis_manifest_path(deliverables_base),
)
#: REMBI run manifest filename, flat under deliverables/ beside metadata.csv.
#: Written best-effort by
#: :func:`phenotypic.sdk_._rembi_manifest.write_rembi_manifest` in finalize;
#: folds the post-applied measurements mirror up to each REMBI module's scope.
REMBI_MANIFEST_YAML: Final[str] = "rembi.yaml"
#: Canonical pipeline-spec filename written into the output root by the
#: CLI (and rewritten by the analysis GUI on every recipe edit). Captures
#: operations, measurements, post, filters, and model — i.e. the whole
#: reproducibility surface.
PIPELINE_JSON: Final[str] = f"pipeline{CONFIG_SUFFIX_PIPELINE}"
_LEGACY_PIPELINE_JSON: Final[str] = f"pipeline{LEGACY_JSON_SUFFIX}"
#: Resume-state JSON written by ``ProcessingState.save`` and read by
#: ``ProcessingState.load`` during automatic continuation.
PROCESSING_STATE_JSON: Final[str] = "processing_state.json"
#: Human-readable run README generated by the CLI's ``READMEGenerator``
#: (output-layout + measurement-schema documentation). A user-facing
#: deliverable, so it lives in :data:`DIR_DELIVERABLES` — see
#: :func:`readme_md_path`.
README_MD: Final[str] = "README.md"
#: The resolved tuning spec echoed by ``python -m phenotypic.tune``
#: into :data:`DIR_DELIVERABLES` — the self-contained, re-runnable recipe.
TUNING_SPEC_JSON: Final[str] = f"tuning_spec{CONFIG_SUFFIX_TUNING}"
_LEGACY_TUNING_SPEC_JSON: Final[str] = f"tuning_spec{LEGACY_JSON_SUFFIX}"
#: The winning best pipeline written by the tune CLI into
#: :data:`DIR_DELIVERABLES`; reloads as a runnable ``ImagePipeline``.
BEST_PIPELINE_JSON: Final[str] = f"best_pipeline{CONFIG_SUFFIX_PIPELINE}"
_LEGACY_BEST_PIPELINE_JSON: Final[str] = f"best_pipeline{LEGACY_JSON_SUFFIX}"
#: Sidecar with the selected headline trial's params and scores, written by the
#: tune CLI into :data:`DIR_DELIVERABLES` for GUI Monitor display.
BEST_PARAMS_JSON: Final[str] = "best_params.json"
#: The RF-permutation ``param_importance.json`` report written by the tune
#: CLI into :data:`DIR_DELIVERABLES`.
PARAM_IMPORTANCE_JSON: Final[str] = "param_importance.json"
#: The tune trial journal ``trials.parquet`` written at the output-dir root
#: (powers CLI resume), not under :data:`DIR_DELIVERABLES`.
TRIALS_PARQUET: Final[str] = "trials.parquet"
#: The canonical Optuna study database ``study.db`` (SQLite WAL) written inside
#: the hidden tune cache (:data:`DIR_PHT_TUNE_CACHE`). Holds the Optuna-backed
#: store's persistent, resumable sampler state when the ``tune`` extra is used.
#: A legacy run wrote it at the output root; :func:`resolve_study_db_path` reads
#: either location (no migration).
STUDY_DB: Final[str] = "study.db"
#: Append-only Optuna JournalStorage log for a distributed tuning run.
STUDY_JOURNAL_LOG: Final[str] = "journal.log"
#: The robust-eval held-out split assignment ``split.json`` written inside
#: :data:`DIR_SPLITS`. A machine-state sidecar — it must survive a
#: fresh-master rewrite and gate resume — so it lives in the hidden tune cache
#: (:data:`DIR_PHT_TUNE_CACHE`), not under :data:`DIR_DELIVERABLES`. See
#: :func:`tune_cache_split_assignment_path`.
SPLIT_ASSIGNMENT_JSON: Final[str] = "split.json"
#: The tune-run marker ``run.json`` written into :data:`DIR_PHT_TUNE_CACHE` at
#: run START (before any deliverable exists), so the GUI shell classifier can
#: recognise a live or finished tune output even before ``deliverables/`` lands.
#: Carries the study identity + storage URL + run policy. See
#: :func:`tune_cache_run_marker_path`.
RUN_MARKER_JSON: Final[str] = "run.json"
#: The robust-eval generalization report ``generalization.json`` written into
#: :data:`DIR_DELIVERABLES` — a user-facing deliverable (the winner's held-out
#: gap verdict). See :func:`generalization_path`.
GENERALIZATION_JSON: Final[str] = "generalization.json"
#: The Pareto-front parquet written by a **multi-objective** tune run into
#: :data:`DIR_PARETO` (under ``deliverables/``). One row per non-dominated trial
#: (the same schema as :data:`TRIALS_PARQUET`, with ``objectives_json`` populated).
PARETO_FRONT_PARQUET: Final[str] = "pareto_front.parquet"
#: Per-objective best-pipeline filename template written into :data:`DIR_PARETO`:
#: ``best_<objective>`` with the pipeline config suffix. One per
#: objective axis — the pipeline with the lowest cost on that axis of the front.
#: Rendered by :func:`pareto_best_pipeline_path`; kept private (a parameterized
#: string is not an enumeration — see the code-style note on render functions).
_PARETO_BEST_PIPELINE_FILENAME_TEMPLATE: Final[str] = (
f"best_{{objective}}{CONFIG_SUFFIX_PIPELINE}"
)
#: Per-objective param-importance filename template written into :data:`DIR_PARETO`:
#: ``param_importance_<objective>.json`` (e.g. ``param_importance_s0.json``). The
#: multi-objective sibling of :data:`PARAM_IMPORTANCE_JSON` — one RF-permutation
#: importance report per objective axis. Rendered by :func:`pareto_importance_path`;
#: kept private (a parameterized string is not an enumeration).
_PARETO_IMPORTANCE_FILENAME_TEMPLATE: Final[str] = (
"param_importance_{objective}.json"
)
_WINDOWS_INVALID_FILENAME_CHARS: Final[frozenset[str]] = frozenset('<>:"/\\|?*')
_WINDOWS_RESERVED_DEVICE_NAMES: Final[frozenset[str]] = frozenset(
{
"CON",
"PRN",
"AUX",
"NUL",
"CONIN$",
"CONOUT$",
*(f"COM{suffix}" for suffix in (*range(1, 10), "¹", "²", "³")),
*(f"LPT{suffix}" for suffix in (*range(1, 10), "¹", "²", "³")),
}
)
# ---------------------------------------------------------------------------
# QC artifact filenames (live inside DIR_QC)
# ---------------------------------------------------------------------------
#: DuckDB QC analysis database filename: ``<output>/deliverables/qc/qc.duckdb``.
#: The sole QC artifact written by
#: :func:`phenotypic.sdk_._qc_recipe._runner.run_qc`: one self-describing table
#: per QC module (worst-direction metric, tri-state status, flag, member rows,
#: worst-first ``rank``) plus a ``qc_modules`` catalog describing each table.
QC_DUCKDB: Final[str] = "qc.duckdb"
#: Per-module GUI review progress (``instance_id`` -> reviewed group keys +
#: last position). Written **only** by the results-viewer QC Review tab;
#: :func:`phenotypic.sdk_._qc_recipe._runner.run_qc` never touches it. The CLI finalize
#: path clears it on every rerun (a fresh run resets review progress).
QC_REVIEW_STATE_JSON: Final[str] = "review_state.json"
# ---------------------------------------------------------------------------
# Run-time progress sidecar files (live inside DIR_PROGRESS)
# ---------------------------------------------------------------------------
#: Append-only JSONL event log of per-image processing transitions.
#: Producers: ``_cli_update_state.append_event``. Consumers: dashboard
#: generator, recompile worker. Was previously re-spelled in 10 files.
PROCESSING_EVENTS_LOG: Final[str] = "processing_events.log"
#: SLURM job-metadata sidecar written once at job submission. Keys are
#: documented in :class:`JobMetadataKey`.
JOB_METADATA_JSON: Final[str] = "job_metadata.json"
#: GUI launch-generation ownership record written before execution starts.
GUI_LAUNCH_OWNER_JSON: Final[str] = "gui_launch_owner.json"
#: Private child-process environment binding for a GUI launch generation.
GUI_RECORD_GENERATION_ENV_VAR: Final[str] = (
"PHENOTYPIC_GUI_RECORD_GENERATION"
)
#: Mutable active-generation fence for a SLURM launch. One of spec §4.1's
#: three written authorities (liveness and ownership), so the run-state
#: reader needs the name and INV-LAYER forbids it importing the CLI module
#: that used to be the name's only home.
SLURM_LIFECYCLE_JSON: Final[str] = "slurm_lifecycle.json"
#: Generation- and mode-bearing terminal publication marker.
RUN_COMPLETION_JSON: Final[str] = "run_completion.json"
#: Marker-last evidence for one internally consistent aggregate snapshot.
AGGREGATE_PUBLICATION_JSON: Final[str] = "aggregate_publication.json"
#: Append-only JSONL of per-image failures. Each row carries a
#: :data:`phenotypic.sdk_.typing_.FailureSource` tag.
FAILURES_JSONL: Final[str] = "failures.jsonl"
#: Authoritative append-only journal of terminal per-image scientific failures.
#: Unlike ``FAILURES_JSONL``, this file lives directly under ``.phenotypic/``
#: because progress artifacts are rebuildable display state.
TERMINAL_FAILURES_JSONL: Final[str] = "terminal_failures.jsonl"
#: SLURM checkpoint chunk manifest (mid-run partial-result feed).
CHUNK_MANIFEST_JSON: Final[str] = "chunk_manifest.json"
#: SLURM checkpoint chunk state (which per-image files have been chunked).
CHUNK_STATE_JSON: Final[str] = "chunk_state.json"
#: Top-level dashboard manifest read by the dashboard HTML JS shim.
#: Keys are documented in :class:`DashboardManifestKey`.
MANIFEST_JSON: Final[str] = "manifest.json"
#: Per-dataset pre-aggregated Parquet emitted by the chunk writer for the
#: GPFS-optimized aggregation path. Underscore prefix marks the file as
#: an aggregator-internal intermediate (chunk-writer scan should skip).
DATASET_AGGREGATED_PARQUET: Final[str] = "_dataset_aggregated.parquet"
# ---------------------------------------------------------------------------
# Dashboard / log files
# ---------------------------------------------------------------------------
#: Generated progress and failure dashboard.
DASHBOARD_HTML: Final[str] = "dashboard.html"
#: Per-run stdout capture written by the run console's local runner.
STDOUT_LOG: Final[str] = "stdout.log"
#: Reserved directory created inside an output root by the GUI runner.
RUN_LOG_DIRNAME: Final[str] = ".gui_log"
#: Complete allowlist of files permitted inside :data:`RUN_LOG_DIRNAME` when
#: the CLI decides whether an output is otherwise fresh.
GUI_LOG_FILENAMES: Final[frozenset[str]] = frozenset({STDOUT_LOG})
#: Top-level processing report HTML emitted by recompile mode.
PROCESSING_REPORT_HTML: Final[str] = "processing_report.html"
#: Legacy-named rolling aggregation state that the chunk writer rewrites as
#: per-image measurements arrive. It is unrelated to the retired static
#: analysis dashboard and remains in ``DIR_PROGRESS`` for resumable runs.
ANALYSIS_FULL_PARQUET: Final[str] = "analysis_full.parquet"
#: Sentinel marker file written by the SLURM sentinel after a re-submission;
#: presence prevents an infinite resubmission loop on the next checkpoint.
SENTINEL_RESUBMITTED_MARKER: Final[str] = "sentinel_resubmitted"
#: Hidden chunk-aggregation lock file (lives in ``DIR_PROGRESS``).
CHUNK_LOCK: Final[str] = ".chunk_lock"
#: Recompile task manifest JSON (one per recompile invocation; lives at
#: ``<output>/.phenotypic/progress/recompile/task_manifest.json``). Lists every per-image
#: shard the recompile worker is responsible for; consumed by
#: ``_cli_recompile_worker._main``.
RECOMPILE_TASK_MANIFEST_JSON: Final[str] = "task_manifest.json"
# ---------------------------------------------------------------------------
# CLI artifact directory names
# ---------------------------------------------------------------------------
#: ``<output>/results/`` — per-dataset subdirectories live below here.
DIR_RESULTS: Final[str] = "results"
#: ``progress`` — directory-name segment for sidecar state under
#: ``<output>/.phenotypic/progress/``. Legacy readers may resolve a root-level
#: ``<output>/progress/`` directory.
DIR_PROGRESS: Final[str] = "progress"
DIR_IMAGE_COMPLETE: Final[str] = "image_complete"
#: One record per image, replacing ``image_complete/`` and
#: ``stage3_complete/`` (spec §6.1). ``stage2_raw/`` stays a separate tree: it is
#: bulk replay data, not a record.
DIR_IMAGE_RECORDS: Final[str] = "images"
#: ``<output>/.phenotypic/legacy-v2/`` -- the pre-collapse marker trees, kept
#: after ``--mode migrate`` so ``--mode migrate --revert`` costs a rename back
#: rather than a full reprocess (CAN-12, §15.1).
#:
#: **Retained for revert; read by nothing.** It is not tracked state: no
#: verdict consults it, nothing derives from it, and nothing must be kept in
#: sync with it. It sits directly below ``.phenotypic/`` rather than below
#: ``progress/`` **on purpose** -- the schema gate's directory signals look
#: only below ``progress/``, so a retained tree here cannot make an
#: already-converted output classify ``CONVERT`` again.
DIR_LEGACY_V2: Final[str] = "legacy-v2"
#: ``<progress>/stage2_done/`` -- the consumable Stage-2 token's tree.
#:
#: **Retained, not collapsed** (U-9), which is why it is here and
#: ``stage3_complete/`` is not. The stage-3 marker's segment stayed a
#: module-private literal in :mod:`._schema_shape` on the reasoning that P3
#: deletes the tree, so promoting it would add a constant the change was about
#: to remove. That reasoning does not transfer: this tree survives the collapse
#: with its file and its atomic ``unlink`` intact, so its segment is a durable
#: layout fact with two readers -- the token's path helper and the schema gate,
#: which must keep *not* firing on it -- and belongs beside its siblings.
DIR_STAGE2_DONE: Final[str] = "stage2_done"
#: ``<output>/.phenotypic/verification_cache.json`` -- the on-disk second tier
#: of the verification cache (spec §9.1, as reversed back on by U-11).
#:
#: **A cache, and named like one.** Nothing branches on it and no verdict is
#: derived from it; it only ever licenses *skipping* a deep pass whose stat
#: tuples still match. It is **deleted** by :func:`clear_machine_state`; the
#: rule that decides why, and the test that enforces it, are stated once at
#: :data:`_PRESERVED_ON_RESTART`.
VERIFICATION_CACHE_JSON: Final[str] = "verification_cache.json"
#: ``<output>/.phenotypic/restart_epoch.json`` -- the run's restart counter
#: (spec §5.1 D4), and the **one tracked value this design adds**.
#:
#: Preserved across ``--restart``, unlike everything else under
#: ``.phenotypic/`` bar the terminal-failure journal. The rule that puts it
#: there -- and keeps the verification cache out -- is at
#: :data:`_PRESERVED_ON_RESTART`.
RESTART_EPOCH_JSON: Final[str] = "restart_epoch.json"
#: Schema version of the persisted verification cache.
#:
#: **Bump this when the deep-verification RULES change, not only when the JSON
#: shape does.** The reader has no other way to tell that a file was written
#: by a build whose notion of "verified" differed from its own: the payload
#: records what was checked, never how. A rules change shipped without a bump
#: is a build silently honouring another build's verdicts, which is the one
#: failure the in-process tier could not have.
#:
#: **2 since FU-1's exclusion advisory.** Deep verification now records two
#: more facts into each image's ``measured`` stage -- whether its record
#: authorizes a measurement table, and whether its store declares a
#: projectable column list -- and ``resolve_run_state`` names an excluded
#: store by projecting over them. A version-1 entry carries valid stat
#: tuples and neither fact, so a warm shallow pass would reuse it and emit
#: no advisory: the diagnostic would be silently switched off by a cache,
#: which is exactly what the paragraph above forbids. This is the "rules
#: changed, shape did not" case it describes, and the first bump to invoke
#: it. Guard:
#: ``test_run_state.py::test_a_cache_written_before_the_advisory_cannot_silence_it``.
VERIFICATION_CACHE_VERSION: Final[int] = 2
#: Per-image success-marker schema version. Bumped to 2 when artifact
#: descriptors gained ``kind``: a v1 marker describes the per-image ``.h5``,
#: which ``--mode migrate`` keeps by default, so a v1 marker would *validate*
#: against that file while the store it should describe went entirely
#: unverified (ledger FLOW-23).
#:
#: It lives here rather than in ``_cli_completion`` -- which re-exports it,
#: so every existing importer is unchanged -- because the run-state reader
#: must check the same number and INV-LAYER forbids that module importing the
#: writer. Two copies of a version number that gate a *completion* verdict is
#: the one duplication that can silently manufacture a false ``complete``.
SUCCESS_MARKER_VERSION: Final[int] = 2
#: Artifact descriptor kinds. ``"file"`` is the default for a descriptor
#: written before the ``kind`` tag existed; a ``"store"`` descriptor carries
#: no ``size`` and digests the store's root ``zarr.json`` instead.
ARTIFACT_KIND_FILE: Final[str] = "file"
ARTIFACT_KIND_STORE: Final[str] = "store"
#: Marker-last proof schema versions, read by the run-state resolver and
#: written by ``_cli_completion``'s two publishers.
#:
#: ``AGGREGATE_PROOF_VERSION`` moved 1 -> 2 in P4. Two independent reasons, and
#: the second is the decisive one: the proof lost its ``master_csv`` required
#: output and its ``publication_id`` (D8, U-4); and **the master's own shape
#: changed** -- it is now un-joined and carries intrinsic identity only. Every
#: pre-P4 aggregate proof is therefore stale in substance whether or not the
#: version moves, and a stale proof that still validates is indistinguishable
#: from a current one. Invalidating costs **re-aggregation, not
#: reprocessing**: the master is rebuilt from embedded tables already on disk,
#: with no image re-measured and no store rewritten.
#:
#: ``RUN_PROOF_VERSION`` is deliberately NOT bumped alongside it. The run
#: proof's new ``source_set_digest`` is read through
#: ``_run_state._source_set_binding``, which handles both shapes on purpose so
#: P1's comparison keeps being made across P4's writer bump.
AGGREGATE_PROOF_VERSION: Final[int] = 2
RUN_PROOF_VERSION: Final[int] = 2
#: Per-dataset measurements subdirectory: ``<output>/results/<ds>/measurements/``.
DIR_MEASUREMENTS: Final[str] = "measurements"
#: Per-feature spreadsheet split written by
#: :func:`phenotypic._cli._cli_output_manager.split_master_by_feature`.
DIR_MEASUREMENTS_BY_FEATURE: Final[str] = "measurements_by_feature"
#: SLURM stdout/stderr subdirectory inside the hidden machine-state cache.
DIR_LOGS: Final[str] = "logs"
#: HDF5 image-state subdirectory of a LEGACY run: ``<output>/results/<ds>/hdf/``.
#: Module-private since Phase 6. The only thing here that still needs it is
#: :func:`datasets_needing_migration`, the predicate that refuses an
#: unconverted tree; ``--mode migrate`` carries its own copy in
#: :mod:`phenotypic.sdk_._hdf_to_zarr`, which is the module allowed to know
#: the legacy layout.
_DIR_HDF: Final[str] = "hdf"
#: OME-Zarr image-state subdirectory: ``<output>/results/<ds>/zarr/``.
DIR_ZARR: Final[str] = "zarr"
#: Overlay PNG subdirectory: ``<output>/deliverables/overlays/<ds>/``.
DIR_OVERLAYS: Final[str] = "overlays"
#: Configured plot outputs: ``<output>/deliverables/plots/``.
DIR_PLOTS: Final[str] = "plots"
#: Mid-run chunk parquet subdirectory: ``<progress>/chunks/``.
DIR_CHUNKS: Final[str] = "chunks"
#: Recompile-worker shard / status subdirectory: ``<progress>/recompile/``.
DIR_RECOMPILE: Final[str] = "recompile"
#: Per-task JSON status subdirectory: ``<progress>/recompile/status/``.
DIR_RECOMPILE_STATUS: Final[str] = "status"
#: Per-shard parquet subdirectory inside the recompile dir:
#: ``<progress>/recompile/measurement_shards/``.
DIR_RECOMPILE_SHARDS: Final[str] = "measurement_shards"
#: Per-invocation aggregation shard subdirectory:
#: ``<progress>/aggregation_shards/<scheduler_epoch>/``.
#:
#: **Deliberately not ``measurement_shards``**, which is
#: :data:`DIR_RECOMPILE_SHARDS` and already names two live and *different*
#: paths: ``recompile_dir(progress) / DIR_RECOMPILE_SHARDS``, which
#: ``_cli_finalize_run._invalidate_finalization_intermediates`` removes, and
#: ``attempt_dir / DIR_RECOMPILE_SHARDS``, which the recompile worker writes
#: and reads. A third use of that string would make an existing ambiguity
#: harder to see rather than adding a new one, so the fan-out's shards get a
#: leaf name whose ``grep`` means one thing.
DIR_AGGREGATION_SHARDS: Final[str] = "aggregation_shards"
#: Path segment standing in for a null ``scheduler_epoch``.
#:
#: :func:`phenotypic.sdk_._run_state._scheduler_epoch` returns ``None``
#: whenever there is no ``slurm_lifecycle.json`` -- which is every local run --
#: so the local fan-out driver has no epoch to namespace by. The segment keeps
#: one path shape across both drivers instead of letting a ``None`` reach the
#: filesystem as the literal string ``"None"``.
#:
#: **Namespacing is not what makes the local path correct.** Consecutive local
#: runs share this segment, so the fan-out empties the shard directory when it
#: starts (user ruling, P2 close) rather than relying on a distinct key.
#:
#: **A different field in this subsystem legitimately takes this same string,
#: and they are not the same namespace.** ``execution_epoch`` is written as
#: ``"local"`` by ``_cli_gui_lifecycle`` and compared against it in
#: ``_cli_completion``. ``scheduler_epoch`` is a minted lifecycle generation
#: and is never this literal, so the two cannot collide -- but they sit close
#: enough that a reader could conclude otherwise.
LOCAL_SCHEDULER_EPOCH: Final[str] = "local"
#: Generated SLURM script subdirectory inside the hidden machine-state cache.
DIR_SLURM_SCRIPTS: Final[str] = "slurm_scripts"
#: QC artifact subdirectory: ``<output>/deliverables/qc/``. Holds
#: :data:`QC_DUCKDB` (written by ``run_qc``) and
#: :data:`QC_REVIEW_STATE_JSON` (written by the GUI Review tab). Relocated
#: under ``deliverables/`` so a bundle is self-contained;
#: :attr:`BundleLayout.qc_dir` / :func:`migrate_legacy_qc` handle the legacy
#: pre-relocation root ``<output>/qc/``.
DIR_QC: Final[str] = "qc"
#: ``<output>/deliverables/`` — all user-facing run outputs collected in one
#: folder: master + post-applied measurements, the per-feature splits, the
#: analysis frames, the generated dashboard/analysis/report HTML, the
#: :data:`README_MD`, and the canonical :data:`PIPELINE_JSON`. Distinct from
#: machine-state sidecars (``progress/``, ``processing_state.json``) and from
#: per-image artifacts (``results/``), which stay at the output root. Every
#: artifact-path helper below that previously rooted at ``<output>/`` now
#: roots at :func:`deliverables_dir`.
DIR_DELIVERABLES: Final[str] = "deliverables"
#: Per-category error-object parquet subdirectory under deliverables:
#: ``<output>/deliverables/errors/<category>.parquet``. Holds the master rows
#: for each triaged error category. **Dual-owned:** the GUI writes them live as
#: the user curates (via ``CurationLabels._save_locked``) and CLI finalize
#: re-emits them (all categories) from the durable ``qc/curation_labels.parquet``
#: via ``reemit_error_deliverables`` — so headless == live.
DIR_ERRORS: Final[str] = "errors"
#: Durable curation-labels store: ``<output>/deliverables/qc/curation_labels.parquet``.
#: The source of truth for categorized removals; the CLI re-keys but never
#: wipes it (contrast :data:`QC_REVIEW_STATE_JSON`).
CURATION_LABELS_PARQUET: Final[str] = "curation_labels.parquet"
#: Ordered custom-category registry sidecar:
#: ``<output>/deliverables/qc/custom_categories.json``.
CUSTOM_CATEGORIES_JSON: Final[str] = "custom_categories.json"
#: Ranked error-cutoff analysis deliverables (``ErrorCutoffFinder`` output;
#: columns ``[category, *RESULT_COLUMNS]``). **Dual-owned, but with differing
#: scope:** the GUI's Error tab writes the parquet/csv live for the *focused*
#: category (transient, last-viewed-in-session) on each recompute, while CLI
#: finalize (``reemit_error_deliverables``) authoritatively rewrites them across
#: *all* labeled categories from the durable labels store. The HTML is written
#: only on an explicit GUI ``Save analysis report`` or by CLI finalize — never on
#: a live recompute.
ERROR_ANALYSIS_PARQUET: Final[str] = "error_analysis.parquet"
ERROR_ANALYSIS_CSV: Final[str] = "error_analysis.csv"
ERROR_ANALYSIS_HTML: Final[str] = "error_analysis.html"
#: Filename of the GUI-written verified-good baseline archive (spec §9). It is
#: derived from ``qc/review_state.json`` (which CLI finalize RESETS), so it is
#: GUI-owned and never CLI-emitted; finalize leaves any existing file untouched.
VERIFIED_PARQUET: Final[str] = "verified.parquet"
#: ``splits/`` — the robust-eval held-out **split assignment** sidecar folder
#: (holds :data:`SPLIT_ASSIGNMENT_JSON`). Lives inside the hidden tune cache
#: (:data:`DIR_PHT_TUNE_CACHE`), **not** under :data:`DIR_DELIVERABLES`: the
#: split is machine state that must survive a fresh-master rewrite and gate
#: resume. See :func:`tune_cache_splits_dir`.
DIR_SPLITS: Final[str] = "splits"
#: ``<output>/deliverables/pareto/`` — the **multi-objective** sub-folder holding
#: the Pareto front parquet (:data:`PARETO_FRONT_PARQUET`) and the per-objective
#: best pipelines (:func:`pareto_best_pipeline_path`). Written only by a
#: multi-objective tune run; a single-objective run never creates it (the
#: back-compat lock — plan §0b). Rooted under :func:`deliverables_dir`.
DIR_PARETO: Final[str] = "pareto"
#: ``<output>/.phenotypic/`` — hidden machine-state cache root. Holds the
#: run's progress/, processing_state.json, and processing_events.log. Hidden
#: so it does not clutter the user-facing output folder and is skipped by the
#: GUI's run/dataset candidate scan. NOTE: distinct from the GUI's
#: ``.phenotypic-gui`` sandbox dir (presets/state) — different root, different
#: purpose.
DIR_PHENOTYPIC: Final[str] = ".phenotypic"
#: ``<output>/.pht-tune-cache/`` — hidden machine-state cache root for a **tune**
#: run, the sibling of :data:`DIR_PHENOTYPIC` for the forward CLI. Holds the
#: Optuna ``study.db`` (+ WAL), the held-out ``splits/split.json``, and the
#: GUI-discovery ``run.json`` marker. Hidden so it does not clutter the
#: user-facing output and is skipped by the GUI candidate scan; the tune cache
#: is kept distinct from ``.phenotypic`` so a directory that hosts both a
#: forward run and a tune run never collides their machine-state. Note that
#: ``trials.parquet`` is NOT relocated here — it stays at the output root as the
#: dual-purpose Optuna-resume + user-facing trial journal.
DIR_PHT_TUNE_CACHE: Final[str] = ".pht-tune-cache"
# ---------------------------------------------------------------------------
# Templated filenames — Final[str] template + typed render function
# ---------------------------------------------------------------------------
# Pattern: keep the raw ``"…{x}…"`` template private as ``_FOO_TEMPLATE``,
# expose a typed render function whose parameters are the public API.
# Callers must reach the template through the function — never .format() it
# directly. This is the project's first deployment of CLAUDE.md's
# "parameterized strings are not enumerations" rule.
_TASK_STATUS_FILENAME_TEMPLATE: Final[str] = "task_{task_index}.json"
[docs]
def task_status_filename(task_index: int) -> str:
"""Filename of a per-task SLURM-recompile status JSON.
Args:
task_index: Zero-based recompile task index.
Returns:
Filename relative to ``<progress>/recompile/status/``.
"""
return _TASK_STATUS_FILENAME_TEMPLATE.format(task_index=task_index)
_SHARD_PARQUET_FILENAME_TEMPLATE: Final[str] = "shard_{shard_id}.parquet"
[docs]
def shard_parquet_filename(shard_id: int) -> str:
"""Filename of a per-shard Parquet inside the recompile worker.
Args:
shard_id: Zero-based shard index.
Returns:
Filename relative to the recompile shard directory.
"""
return _SHARD_PARQUET_FILENAME_TEMPLATE.format(shard_id=shard_id)
_CHUNK_PARQUET_FILENAME_TEMPLATE: Final[str] = "chunk_{chunk_id:03d}.parquet"
[docs]
def chunk_parquet_filename(chunk_id: int) -> str:
"""Filename of a dashboard chunk Parquet (zero-padded chunk id).
Args:
chunk_id: Zero-based chunk index, formatted ``{chunk_id:03d}``.
Returns:
Filename relative to ``<progress>/chunks/``.
"""
return _CHUNK_PARQUET_FILENAME_TEMPLATE.format(chunk_id=chunk_id)
_DEFAULT_OUTPUT_DIR_NAME_TEMPLATE: Final[str] = (
"phenotypic_results_{timestamp}"
)
_DEFAULT_OUTPUT_TIMESTAMP_FORMAT: Final[str] = "%Y%m%d_%H%M%S"
[docs]
def default_output_dir_name(now: Optional[datetime] = None) -> str:
"""Default name for an auto-generated output directory.
Legacy helper for timestamped output-directory names.
Args:
now: Override clock for tests; defaults to ``datetime.now()``.
Returns:
``"phenotypic_results_YYYYMMDD_HHMMSS"``.
"""
when = now if now is not None else datetime.now()
return _DEFAULT_OUTPUT_DIR_NAME_TEMPLATE.format(
timestamp=when.strftime(_DEFAULT_OUTPUT_TIMESTAMP_FORMAT)
)
_CHECKPOINT_LOCK_FILENAME_TEMPLATE: Final[str] = ".{checkpoint_type}_lock"
[docs]
def checkpoint_lock_filename(checkpoint_type: CheckpointType) -> str:
"""Filename of the SLURM-sentinel exclusive lock for a checkpoint task.
Args:
checkpoint_type: ``"manifest"`` or ``"finalize"``. Validated by
the :data:`phenotypic.sdk_.typing_.CheckpointType` Literal
alias at type-check time.
Returns:
Filename relative to ``<progress>/`` (hidden file, leading ``.``).
"""
return _CHECKPOINT_LOCK_FILENAME_TEMPLATE.format(
checkpoint_type=checkpoint_type
)
# ---------------------------------------------------------------------------
# Path-builder helpers — replace inline ``output / "literal"`` constructions
# ---------------------------------------------------------------------------
[docs]
def phenotypic_cache_dir(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/`` — the hidden machine-state root.
Pure path expression; callers ``mkdir`` when they intend to write.
"""
return output_dir / DIR_PHENOTYPIC
[docs]
def progress_dir(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/progress/``.
Pure path expression; callers are responsible for ``mkdir`` when they
intend to write into it.
"""
return phenotypic_cache_dir(output_dir) / DIR_PROGRESS
[docs]
def restart_epoch_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/restart_epoch.json``.
Pure path expression. The readers and the writer are
:func:`phenotypic._cli._cli_identity.read_restart_epoch` and
:func:`~phenotypic._cli._cli_identity.bump_restart_epoch` -- the writer
lives in ``_cli`` because spec §5.2 keeps every publisher out of ``sdk_``,
and only the path belongs here.
"""
return phenotypic_cache_dir(output_dir) / RESTART_EPOCH_JSON
[docs]
def verification_cache_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/verification_cache.json``.
Pure path expression; the caller decides whether to write. Note that the
cache's writer deliberately does **not** ``mkdir`` this path's parent --
see :func:`phenotypic.sdk_._verification_cache.persist_states`.
"""
return phenotypic_cache_dir(output_dir) / VERIFICATION_CACHE_JSON
[docs]
def results_dir(output_dir: Path) -> Path:
"""Return ``<output>/results/``."""
return output_dir / DIR_RESULTS
[docs]
def deliverables_dir(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/`` — the user-facing-output folder.
Pure path expression; callers are responsible for ``mkdir`` when they
intend to write into it. Writers that go through
:func:`phenotypic.sdk_.atomic_write_with_writer` get the ``mkdir`` for free
(it creates ``target.parent``); direct ``write_text``/``write_bytes``
writers must ``mkdir`` explicitly.
Every artifact helper that previously rooted at ``<output>/`` (master /
measurements / per-feature split / analysis / dashboard / report /
pipeline.json / README) now composes from here, so a future relocation
is a one-line change.
"""
return output_dir / DIR_DELIVERABLES
[docs]
def plots_dir(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/plots/``."""
return deliverables_dir(output_dir) / DIR_PLOTS
[docs]
def event_log_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/processing_events.log``."""
return phenotypic_cache_dir(output_dir) / PROCESSING_EVENTS_LOG
[docs]
def terminal_failures_jsonl_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/terminal_failures.jsonl``."""
return phenotypic_cache_dir(output_dir) / TERMINAL_FAILURES_JSONL
[docs]
def processing_state_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/processing_state.json``."""
return phenotypic_cache_dir(output_dir) / PROCESSING_STATE_JSON
def _legacy_progress_dir(output_dir: Path) -> Path:
"""Pre-migration location: ``<output>/progress/``."""
return output_dir / DIR_PROGRESS
def _legacy_processing_state_path(output_dir: Path) -> Path:
"""Pre-migration location: ``<output>/processing_state.json``."""
return output_dir / PROCESSING_STATE_JSON
[docs]
def resolve_progress_dir(output_dir: Path) -> Path:
"""Return the progress dir that exists, preferring ``.phenotypic/``.
Read-only helper for resume/discovery so a pre-migration run (progress
at the output root) is still found. Falls back to the new location when
neither exists (the default for fresh writes).
"""
new = progress_dir(output_dir)
if new.exists():
return new
legacy = _legacy_progress_dir(output_dir)
if legacy.exists():
return legacy
return new
[docs]
def resolve_processing_state_path(output_dir: Path) -> Path:
"""Return the processing-state file that exists, preferring ``.phenotypic/``."""
new = processing_state_path(output_dir)
if new.exists():
return new
legacy = _legacy_processing_state_path(output_dir)
if legacy.exists():
return legacy
return new
[docs]
def resolve_manifest_json_path(output_dir: Path) -> Path:
"""Return ``<progress>/manifest.json`` resolving the progress dir for legacy runs."""
return resolve_progress_dir(output_dir) / MANIFEST_JSON
[docs]
def resolve_event_log_path(output_dir: Path) -> Path:
"""Return the event log sibling of the resolved progress dir.
The event log lives beside ``progress/`` (D14): in ``.phenotypic/`` for a
migrated/new run, at the output root for a not-yet-migrated legacy read.
Read-only helper for resume/discovery; never mutates the run dir.
"""
return resolve_progress_dir(output_dir).parent / PROCESSING_EVENTS_LOG
[docs]
def migrate_legacy_machine_state(output_dir: Path) -> bool:
"""Move a pre-migration run's machine-state into ``.phenotypic/``.
If legacy machine-state (``progress/``, ``processing_state.json``,
``processing_events.log``) is present at the output root, move each artifact
into the ``.phenotypic/`` cache so the run proceeds coherently against a
single location. A no-op when no legacy state is present or everything is
already migrated.
Robust to interruption and concurrency (the SLURM array case): each artifact
is moved only when its source still exists and its destination does not, so a
migration interrupted mid-move *completes* on the next call rather than
leaving split state; and a lost move race (a concurrent worker moved the
artifact first) is ignored rather than crashing. Keying per-artifact instead
of on ``cache.exists()`` is what makes both safe.
Returns:
``True`` if this call moved anything, else ``False``.
"""
import shutil
cache = phenotypic_cache_dir(output_dir)
legacy_progress = _legacy_progress_dir(output_dir)
legacy_state = _legacy_processing_state_path(output_dir)
legacy_events = output_dir / PROCESSING_EVENTS_LOG
if not (
legacy_progress.exists()
or legacy_state.exists()
or legacy_events.exists()
):
return False
cache.mkdir(parents=True, exist_ok=True)
moved = False
for src, dst in (
(legacy_progress, cache / DIR_PROGRESS),
(legacy_state, cache / PROCESSING_STATE_JSON),
(legacy_events, cache / PROCESSING_EVENTS_LOG),
):
if src.exists() and not dst.exists():
try:
shutil.move(str(src), str(dst))
moved = True
except (FileNotFoundError, shutil.Error):
# Lost a migration race with a concurrent worker; the winner is
# moving (or has moved) this artifact — safe to skip.
pass
return moved
[docs]
def migrate_legacy_qc(output_dir: Path) -> bool:
"""Move a pre-relocation run's ``<output>/qc/`` into ``deliverables/qc/``.
Hard cutover (MOVE, no duplication), mirroring
:func:`migrate_legacy_machine_state`. A no-op when there is no legacy ``qc/``
or when the canonical ``deliverables/qc/`` already exists (the move is
whole-directory; we never merge a half-written canonical with legacy).
Returns:
``True`` if this call moved the directory, else ``False``.
"""
import shutil
legacy = _legacy_qc_dir(output_dir)
canonical = qc_dir(output_dir)
if not legacy.is_dir() or canonical.exists():
return False
canonical.parent.mkdir(parents=True, exist_ok=True)
try:
shutil.move(str(legacy), str(canonical))
except (FileNotFoundError, shutil.Error):
# Lost a race with a concurrent migrator; safe to skip.
return False
return True
#: Children of ``.phenotypic/`` that :func:`clear_machine_state` keeps.
#:
#: **THE MEMBERSHIP RULE — read this before adding a name.** A name belongs
#: here only when carrying it across a restart is **safer than losing it**.
#:
#: * ``terminal_failures.jsonl`` qualifies: append-only history, and losing it
#: loses the record of *why* images failed.
#: * ``restart_epoch.json`` qualifies: a counter that resets on the operation
#: it fences is not a fence.
#:
#: Anything recording a **verdict reached before the fence** does **not**
#: qualify, however expensive it was to compute -- a restart exists to
#: invalidate exactly those.
#:
#: **The worked exclusion, which is why this rule is written down at all:**
#: ``verification_cache.json`` is a completion verdict, and an expensive one
#: (spec U-11 measures the deep pass it replaces at 1403 s against ~37 s).
#: That expense is precisely the argument that will be made for preserving it,
#: and it is precisely the wrong argument -- a preserved cache would carry a
#: pre-restart ``complete`` across the fence the restart exists to raise. It
#: is deleted by falling into the sweep's everything-else branch, and that
#: exclusion is enforced by
#: ``test_clear_machine_state_deletes_the_persisted_cache`` in
#: ``tests/unit/sdk_/test_verification_cache_disk.py`` -- a suite an editor of
#: this set will not have open, which is why the rule lives here rather than
#: only there.
#:
#: This set is a **module-level constant on purpose**: later phases are told
#: to grow it (P7 adds ``legacy-v2/``, the retained revert path), and a set
#: those phases must find and extend does not belong in a function body where
#: it can carry no documentation.
_PRESERVED_ON_RESTART: Final[frozenset[str]] = frozenset(
{TERMINAL_FAILURES_JSONL, RESTART_EPOCH_JSON, DIR_LEGACY_V2}
)
[docs]
def clear_machine_state(output_dir: Path) -> bool:
"""Remove **all** of a run's machine-state for a clean ``--restart``.
Deletes current state inside ``.phenotypic/`` (``progress/``,
``processing_state.json``, ``processing_events.log``, logs, and generated
SLURM scripts) and any pre-migration root-level machine-state, while
preserving :data:`_PRESERVED_ON_RESTART` -- the append-only
``terminal_failures.jsonl`` journal **and** ``restart_epoch.json``, because
a counter that resets on the operation it fences is not a fence -- along
with user-facing output artifacts (``deliverables/``, ``results/``,
``qc/``, …). **Read that set's membership rule before adding to it**; it is
now the only thing standing between ``--restart`` and every artifact under
``.phenotypic/``.
This is
the difference between ``--restart``
(re-run the orchestration against clean state, keep outputs) and
``--overwrite`` (delete the whole output dir). Clearing the event log here is
what stops a restart from appending to — and rebuilding its manifest/failure
records from — the prior run's events.
Returns:
``True`` if any machine-state was removed, else ``False``.
"""
import shutil
removed = False
# Current layout: clear each child so restart can preserve the append-only
# terminal-failure journal. Explicit --overwrite removes the whole output
# directory through its separate destructive path.
cache = phenotypic_cache_dir(output_dir)
if cache.exists():
for child in cache.iterdir():
if child.name in _PRESERVED_ON_RESTART:
continue
if child.is_dir() and not child.is_symlink():
shutil.rmtree(child)
else:
child.unlink()
removed = True
if not any(cache.iterdir()):
cache.rmdir()
# Pre-migration (legacy) root-level machine-state, if a legacy run is restarted.
legacy_progress = _legacy_progress_dir(output_dir)
if legacy_progress.exists():
shutil.rmtree(legacy_progress)
removed = True
for legacy_file in (
_legacy_processing_state_path(output_dir),
output_dir / PROCESSING_EVENTS_LOG,
):
if legacy_file.exists():
legacy_file.unlink()
removed = True
return removed
[docs]
def master_measurements_parquet_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/master_measurements.parquet``."""
return deliverables_dir(output_dir) / MASTER_MEASUREMENTS_PARQUET
[docs]
def measurements_csv_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/measurements.csv`` (post-applied mirror)."""
return deliverables_dir(output_dir) / MEASUREMENTS_CSV
[docs]
def measurements_parquet_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/measurements.parquet`` (post-applied mirror)."""
return deliverables_dir(output_dir) / MEASUREMENTS_PARQUET
[docs]
def rembi_manifest_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/rembi.yaml`` — the REMBI run manifest."""
return deliverables_dir(output_dir) / REMBI_MANIFEST_YAML
[docs]
def pipeline_json_path(output_dir: Path) -> Path:
"""Return the canonical typed pipeline config path under ``deliverables/``."""
return deliverables_dir(output_dir) / PIPELINE_JSON
def _legacy_pipeline_json_path(output_dir: Path) -> Path:
"""Return the legacy plain-JSON pipeline config path under ``deliverables/``."""
return deliverables_dir(output_dir) / _LEGACY_PIPELINE_JSON
[docs]
def resolve_pipeline_config_path(output_dir: Path) -> Path:
"""Return the best existing pipeline config path for ``output_dir``.
Resolution prefers the canonical typed path, falls back to legacy
``pipeline.json`` when present, and returns the canonical path when neither
exists so writers naturally create typed config files.
"""
canonical = pipeline_json_path(output_dir)
if canonical.exists():
return canonical
legacy = _legacy_pipeline_json_path(output_dir)
if legacy.exists():
return legacy
return canonical
[docs]
def tuning_spec_path(output_dir: Path) -> Path:
"""Return the canonical typed tuning spec path under ``deliverables/``."""
return deliverables_dir(output_dir) / TUNING_SPEC_JSON
def _legacy_tuning_spec_path(output_dir: Path) -> Path:
"""Return the legacy plain-JSON tuning spec path under ``deliverables/``."""
return deliverables_dir(output_dir) / _LEGACY_TUNING_SPEC_JSON
[docs]
def resolve_tuning_spec_path(output_dir: Path) -> Path:
"""Return the best existing tuning spec path for ``output_dir``."""
canonical = tuning_spec_path(output_dir)
if canonical.exists():
return canonical
legacy = _legacy_tuning_spec_path(output_dir)
if legacy.exists():
return legacy
return canonical
[docs]
def best_pipeline_path(output_dir: Path) -> Path:
"""Return the canonical typed tuned-winner pipeline path."""
return deliverables_dir(output_dir) / BEST_PIPELINE_JSON
[docs]
def param_importance_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/param_importance.json`` (the report)."""
return deliverables_dir(output_dir) / PARAM_IMPORTANCE_JSON
[docs]
def best_params_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/best_params.json`` (winner params sidecar)."""
return deliverables_dir(output_dir) / BEST_PARAMS_JSON
[docs]
def trials_parquet_path(output_dir: Path) -> Path:
"""Return ``<output>/trials.parquet`` (the trial journal; output-dir root)."""
return Path(output_dir) / TRIALS_PARQUET
[docs]
def tune_cache_dir(output_dir: Path) -> Path:
"""Return ``<output>/.pht-tune-cache/`` — the tune run's machine-state root.
The tune-side sibling of :func:`phenotypic_cache_dir`. Pure path expression;
callers ``mkdir`` when they intend to write.
Args:
output_dir: The run output directory.
Returns:
``<output_dir>/.pht-tune-cache/``.
"""
return Path(output_dir) / DIR_PHT_TUNE_CACHE
[docs]
def tune_cache_run_marker_path(output_dir: Path) -> Path:
"""Return ``<output>/.pht-tune-cache/run.json`` — the tune-run marker.
Written at run START (before any deliverable lands) so a live or finished
tune output is GUI-discoverable. See :data:`RUN_MARKER_JSON`.
Args:
output_dir: The run output directory.
Returns:
``<output_dir>/.pht-tune-cache/run.json``.
"""
return tune_cache_dir(output_dir) / RUN_MARKER_JSON
[docs]
def tune_cache_study_db_path(output_dir: Path) -> Path:
"""Return ``<output>/.pht-tune-cache/study.db`` (the Optuna study DB).
The canonical SQLite-WAL storage for the Optuna-backed
:class:`OptunaStudyStore` when the ``tune`` extra is installed, relocated
into the hidden tune cache. A legacy run wrote it at the output root; use
:func:`resolve_study_db_path` to read either location.
Args:
output_dir: The run output directory.
Returns:
``<output_dir>/.pht-tune-cache/study.db``.
"""
return tune_cache_dir(output_dir) / STUDY_DB
[docs]
def tune_cache_journal_path(output_dir: Path) -> Path:
"""Return ``<output>/.pht-tune-cache/journal.log`` for a tune fleet."""
return tune_cache_dir(output_dir) / STUDY_JOURNAL_LOG
[docs]
def tune_cache_splits_dir(output_dir: Path) -> Path:
"""Return ``<output>/.pht-tune-cache/splits/`` — the held-out split folder.
Machine state that must survive a fresh-master rewrite and gate resume, so
it lives in the hidden tune cache, **not** under :func:`deliverables_dir`.
Pure path expression; callers ``mkdir`` when they intend to write.
Args:
output_dir: The run output directory.
Returns:
``<output_dir>/.pht-tune-cache/splits/``.
"""
return tune_cache_dir(output_dir) / DIR_SPLITS
[docs]
def tune_cache_split_assignment_path(output_dir: Path) -> Path:
"""Return ``<output>/.pht-tune-cache/splits/split.json`` — the held-out split.
The persisted calibration / held-out partition (plate names + split kind +
dataset identity + seed entropy). Read-if-exists-else-derive on resume, so
a re-run reuses the original partition regardless of the new master seed. A
legacy run wrote it under ``<output>/splits/``; use
:func:`resolve_split_assignment_path` to read either location.
Args:
output_dir: The run output directory.
Returns:
``<output_dir>/.pht-tune-cache/splits/split.json``.
"""
return tune_cache_splits_dir(output_dir) / SPLIT_ASSIGNMENT_JSON
def _legacy_study_db_path(output_dir: Path) -> Path:
"""Pre-relocation location: ``<output>/study.db``."""
return Path(output_dir) / STUDY_DB
def _legacy_split_assignment_path(output_dir: Path) -> Path:
"""Pre-relocation location: ``<output>/splits/split.json``."""
return Path(output_dir) / DIR_SPLITS / SPLIT_ASSIGNMENT_JSON
[docs]
def resolve_study_db_path(output_dir: Path) -> Path:
"""Return the study DB that exists, preferring ``.pht-tune-cache/``.
Read-only resolver: a relocated run keeps ``study.db`` under the hidden tune
cache; a legacy run kept it at the output root. Falls back to the new
location when neither exists (so a cold sampler restart from a missing
``study.db`` is harmless — no migration is performed).
Args:
output_dir: The run output directory.
Returns:
The study DB path that exists, else the new cache location.
"""
new = tune_cache_study_db_path(output_dir)
if new.exists():
return new
legacy = _legacy_study_db_path(output_dir)
if legacy.exists():
return legacy
return new
[docs]
def resolve_split_assignment_path(output_dir: Path) -> Path:
"""Return the split assignment that exists, preferring ``.pht-tune-cache/``.
Read-only resolver mirroring :func:`resolve_progress_dir`. The held-out
split is checked in the hidden tune cache FIRST, THEN at the legacy output
root — a missing split silently RE-DERIVES a fresh held-out partition on
resume (a reproducibility / held-out-leak bug), so resume MUST find a
legacy-root ``split.json``. Falls back to the new location when neither
exists (the default for a fresh derive-and-write).
Args:
output_dir: The run output directory.
Returns:
The split-assignment path that exists, else the new cache location.
"""
new = tune_cache_split_assignment_path(output_dir)
if new.exists():
return new
legacy = _legacy_split_assignment_path(output_dir)
if legacy.exists():
return legacy
return new
[docs]
def generalization_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/generalization.json`` — the held-out report.
The winner's generalization verdict (calibration vs held-out score, the gap,
and the pass/fail margin), a user-facing deliverable. The held-out pass that
writes it is Phase 4.5 part 2; this helper resolves the canonical location.
Args:
output_dir: The run output directory.
Returns:
``<output_dir>/deliverables/generalization.json``.
"""
return deliverables_dir(output_dir) / GENERALIZATION_JSON
[docs]
def pareto_dir(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/pareto/`` — the multi-objective sub-folder.
Holds a multi-objective tune run's Pareto front + per-objective best
pipelines. A single-objective run never creates it (the back-compat lock).
"""
return deliverables_dir(output_dir) / DIR_PARETO
[docs]
def pareto_front_parquet_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/pareto/pareto_front.parquet`` (the front)."""
return pareto_dir(output_dir) / PARETO_FRONT_PARQUET
def _safe_pareto_objective_component(objective: str) -> str:
"""Require one contained human-readable Pareto filename component."""
from pathlib import PurePosixPath, PureWindowsPath
import unicodedata
if (
not isinstance(objective, str)
or not objective
or objective in {".", ".."}
or any(char in _WINDOWS_INVALID_FILENAME_CHARS for char in objective)
or PurePosixPath(objective).is_absolute()
or PureWindowsPath(objective).is_absolute()
or bool(PureWindowsPath(objective).drive)
or any(unicodedata.category(char) == "Cc" for char in objective)
or objective.endswith((".", " "))
or objective.split(".", 1)[0].upper() in _WINDOWS_RESERVED_DEVICE_NAMES
):
raise ValueError(
f"Pareto objective {objective!r} is not a safe filename component"
)
return objective
[docs]
def pareto_best_pipeline_path(output_dir: Path, objective: str) -> Path:
"""Return ``deliverables/pareto/best_<objective>.json`` (a per-axis winner).
The pipeline minimizing cost on the single ``objective`` axis of the Pareto front.
``objective`` is the objective name as it appears in ``objectives_json`` (a
scorer-defined label, e.g. ``"Dice"`` or a composite child handle ``"s0"``).
Args:
output_dir: The run directory.
objective: The objective-axis name (the ``best_<objective>.json`` stem).
Returns:
The per-objective best-pipeline path under :func:`pareto_dir`.
"""
safe_objective = _safe_pareto_objective_component(objective)
path = pareto_dir(output_dir) / _PARETO_BEST_PIPELINE_FILENAME_TEMPLATE.format(
objective=safe_objective
)
if path.parent != pareto_dir(output_dir):
raise ValueError("Pareto output path escaped its containing directory")
return path
[docs]
def pareto_importance_path(output_dir: Path, objective: str) -> Path:
"""Return ``deliverables/pareto/param_importance_<objective>.json``.
The per-objective RF-permutation importance report (the multi-objective
sibling of :func:`param_importance_path`). ``objective`` is the objective
name as it appears in ``objectives_json`` (a scorer-defined label, e.g.
``"Dice"`` or a composite child handle ``"s0"``).
Args:
output_dir: The run directory.
objective: The objective-axis name (the filename's ``<objective>`` slot).
Returns:
The per-objective importance-report path under :func:`pareto_dir`.
"""
safe_objective = _safe_pareto_objective_component(objective)
path = pareto_dir(output_dir) / _PARETO_IMPORTANCE_FILENAME_TEMPLATE.format(
objective=safe_objective
)
if path.parent != pareto_dir(output_dir):
raise ValueError("Pareto output path escaped its containing directory")
return path
[docs]
def phenotypic_cache_pipeline_json_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/pipeline.json.pht-pipe`` — the process-only run's
reproducibility copy. Distinct from :func:`pipeline_json_path`, which roots
under ``deliverables/`` (process-only writes no deliverables)."""
return phenotypic_cache_dir(output_dir) / PIPELINE_JSON
[docs]
def dashboard_html_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/dashboard.html``."""
return deliverables_dir(output_dir) / DASHBOARD_HTML
[docs]
def dataset_results_dir(output_dir: Path, dataset: str) -> Path:
"""Return ``<output>/results/<dataset>/``."""
return results_dir(output_dir) / dataset
[docs]
def dataset_measurements_dir(output_dir: Path, dataset: str) -> Path:
"""Return ``<output>/results/<dataset>/measurements/``."""
return dataset_results_dir(output_dir, dataset) / DIR_MEASUREMENTS
[docs]
def dataset_zarr_dir(output_dir: Path, dataset: str) -> Path:
"""Return ``<output>/results/<dataset>/zarr/``."""
return dataset_results_dir(output_dir, dataset) / DIR_ZARR
#: The remedy named in every "this output needs migrating" message. One
#: string, so the CLI's refusal and the viewer's banner cannot drift apart.
MIGRATION_REMEDY: Final[str] = "--mode migrate"
[docs]
def datasets_needing_migration(output_dir: Path) -> list[str]:
"""Datasets holding at least one `.h5` result without a VALID store.
One predicate, so the CLI and the GUI cannot disagree about what
"needs migrating" means.
Per-IMAGE, not per-dataset: the half-migrated tree this exists to catch
has converted and unconverted images in the SAME dataset, so a
dataset-level "has .h5 and has no zarr/ dir" test misses it entirely.
That tree is the expected state after any interruption, because migration
is resumable -- and it is neither "only .h5" nor fully converted, so the
older "only .h5" guard let it through and `--mode full` silently
reprocessed every unconverted image from source.
Validity, not existence: `valid_staged_store`, not `path.exists()`. A
store written at an older `store_schema_version` is present but the
loader refuses it, so an existence test reads that tree as clean while
every image fails to open.
Args:
output_dir: Run output root.
Returns:
Dataset names needing migration, sorted. Empty for a modern tree.
"""
from phenotypic.sdk_.ngff_ import valid_staged_store
root = results_dir(Path(output_dir))
if not root.is_dir():
return []
needing: list[str] = []
for dataset_dir in sorted(path for path in root.iterdir() if path.is_dir()):
hdf_dir = dataset_dir / _DIR_HDF
if not hdf_dir.is_dir():
continue
for hdf_path in sorted(hdf_dir.glob("*.h5")):
if hdf_path.name.startswith("."):
continue
store = zarr_store_path(output_dir, dataset_dir.name, hdf_path.stem)
if not valid_staged_store(store):
needing.append(dataset_dir.name)
break
return needing
[docs]
def zarr_store_path(output_dir: Path, dataset: str, stem: str) -> Path:
"""Return ``<output>/results/<dataset>/zarr/<stem>.ome.zarr/``.
The single place ``.ome.zarr`` is joined to an image stem. Callers must
never hand-join the suffix, and must take the stem back off a store with
:func:`store_stem` rather than ``Path.stem``. Both rules are enforced by
``tests/unit/test_ome_zarr_invariants.py``
(``test_store_suffix_is_joined_in_exactly_one_place`` and
``test_path_stem_is_never_taken_of_a_store_directory``).
Args:
output_dir: Run output root.
dataset: Dataset name.
stem: Image filename without extension.
Returns:
The per-image store path. Existence is not checked.
"""
from phenotypic.sdk_.ngff_ import STORE_SUFFIX
return dataset_zarr_dir(output_dir, dataset) / f"{stem}{STORE_SUFFIX}"
[docs]
def store_stem(store_path: Path) -> str:
"""Return the image stem of an ``*.ome.zarr`` or ``*.zarr`` directory.
``Path.stem`` is WRONG here — it strips one suffix and leaves ``img.ome``,
which is a plausible-looking wrong name rather than an error: it propagates
into parquet filenames and completion markers, and
``zarr_store_path(out, ds, "img.ome")`` then resolves to a store that does
not exist, so every image reprocesses forever.
Args:
store_path: A ``<stem>.ome.zarr`` or ``<stem>.zarr`` directory.
Returns:
The bare stem, e.g. ``"img"`` for ``img.ome.zarr``.
Raises:
ValueError: If *store_path* does not end in ``.zarr``. It raises
rather than falling back to ``.stem``, because a silent fallback is
exactly the failure being prevented.
"""
from phenotypic.sdk_.ngff_ import STORE_SUFFIX
name = Path(store_path).name
suffix = STORE_SUFFIX if name.endswith(STORE_SUFFIX) else ".zarr"
if not name.endswith(suffix) or name == suffix:
raise ValueError(f"not an OME-Zarr store directory: {store_path}")
return name[: -len(suffix)]
[docs]
def is_zarr_store_name(path: Path | str) -> bool:
"""Return whether a path name uses a supported Zarr store suffix.
``.ome.zarr`` remains the canonical PhenoTypic output suffix. Generic
``.zarr`` names are accepted as inputs so validity can be decided by the
NGFF reader at the open/render boundary.
"""
name = Path(path).name
return name.endswith(".zarr") and name != ".zarr"
[docs]
def source_image_stem(path: Path) -> str:
"""Return the canonical artifact stem for a source image path.
OME-Zarr source images use a double suffix, so their identity strips the
complete ``.ome.zarr`` suffix. Every other source keeps the standard
:attr:`pathlib.Path.stem` contract, including ordinary multi-dot files.
Args:
path: Source image file or OME-Zarr store path.
Returns:
The canonical source-image stem.
"""
source = Path(path)
if is_zarr_store_name(source):
return store_stem(source)
return source.stem
[docs]
def source_image_suffix(path: Path) -> str:
"""Return the canonical suffix for a source image path.
Args:
path: Source image file or OME-Zarr store path.
Returns:
``.ome.zarr`` for a store source, otherwise the standard final suffix.
"""
from phenotypic.sdk_.ngff_ import STORE_SUFFIX
source = Path(path)
if source.name.endswith(STORE_SUFFIX):
return STORE_SUFFIX
return ".zarr" if is_zarr_store_name(source) else source.suffix
[docs]
def store_revision_identity(path: Path) -> str:
"""Return a stable revision identity for one OME-Zarr store.
PhenoTypic-published immutable generations use the explicit root-last
publication token and touch only ``zarr.json``. Generic third-party stores
have no publication invariant, so the conservative fallback hashes framed
relative paths, member types, sizes, and nanosecond mtimes twice to reject
an unstable snapshot. It intentionally does not read chunk contents: CLI
work and completion use a separate content-digest contract.
Args:
path: Existing ``*.ome.zarr`` directory.
Returns:
A versioned SHA-256 metadata identity.
Raises:
OSError: If the store is unstable or contains a symlink or another
non-regular member.
ValueError: If ``path`` is not named as an OME-Zarr store.
"""
from phenotypic.sdk_.ngff_ import STORE_ROOT_JSON
store = Path(path)
if not is_zarr_store_name(store):
raise ValueError(f"not an OME-Zarr store directory: {store}")
published = store_publication_token(store)
if published is not None:
return published
first = _store_revision_snapshot(store, root_json=STORE_ROOT_JSON)
second = _store_revision_snapshot(store, root_json=STORE_ROOT_JSON)
if first != second:
raise OSError("OME-Zarr store changed during revision inspection")
members, promoted_root_token = second
digest = hashlib.sha256(b"phenotypic-store-revision\x00v1\x00")
for relative_path, member_type, size, mtime_ns in members:
encoded_path = relative_path.encode("utf-8")
digest.update(len(encoded_path).to_bytes(8, "big"))
digest.update(encoded_path)
digest.update(member_type)
digest.update(size.to_bytes(8, "big", signed=False))
digest.update(mtime_ns.to_bytes(8, "big", signed=True))
digest.update(b"\x01" if promoted_root_token is not None else b"\x00")
if promoted_root_token is not None:
size, mtime_ns = promoted_root_token
digest.update(size.to_bytes(8, "big", signed=False))
digest.update(mtime_ns.to_bytes(8, "big", signed=True))
return f"sha256-stat-tree-v1:{digest.hexdigest()}"
[docs]
def store_publication_token(
store: Path,
*,
root_dir_fd: int | None = None,
) -> str | None:
"""Return the root-last token for a PhenoTypic-published store.
PhenoTypic promotes an immutable store by replacing ``zarr.json`` last.
Its root bytes and file identity therefore identify the complete
generation without touching every chunk on GPFS. Inode and ctime close
the gap where a byte-identical replacement preserves the old mtime. A
generic third-party store has no such publication contract and returns
``None`` so the caller uses the conservative recursive snapshot fallback.
Args:
store: Published store path. Used for ordinary path-based inspection.
root_dir_fd: Optional held descriptor for the store root. When given,
``zarr.json`` is opened relative to that identity with
``O_NOFOLLOW`` so a route can keep validation and serving bound to
one directory generation.
Returns:
The publication token, or ``None`` when the protocol is not declared.
"""
from phenotypic.sdk_.ngff_ import STORE_ROOT_JSON
root = Path(store) / STORE_ROOT_JSON
try:
if root_dir_fd is None:
before = root.lstat()
if not stat_module.S_ISREG(before.st_mode):
return None
raw = root.read_bytes()
after = root.lstat()
else:
flags = os.O_RDONLY | os.O_NONBLOCK | os.O_NOFOLLOW
root_fd = os.open(STORE_ROOT_JSON, flags, dir_fd=root_dir_fd)
try:
before = os.fstat(root_fd)
if (
not stat_module.S_ISREG(before.st_mode)
or before.st_nlink != 1
):
return None
chunks: list[bytes] = []
while chunk := os.read(root_fd, 1024 * 1024):
chunks.append(chunk)
raw = b"".join(chunks)
after = os.fstat(root_fd)
finally:
os.close(root_fd)
except OSError:
return None
before_identity = (
before.st_size,
before.st_mtime_ns,
before.st_ctime_ns,
before.st_ino,
)
after_identity = (
after.st_size,
after.st_mtime_ns,
after.st_ctime_ns,
after.st_ino,
)
if before_identity != after_identity or len(raw) != after.st_size:
raise OSError("OME-Zarr root changed during revision inspection")
try:
payload = json.loads(raw)
except (UnicodeDecodeError, json.JSONDecodeError):
return None
attributes = payload.get("attributes") if isinstance(payload, dict) else None
phenotypic = attributes.get("phenotypic") if isinstance(attributes, dict) else None
if not isinstance(phenotypic, dict):
return None
from phenotypic.sdk_.ngff_ import (
PhenotypicAttr,
ROOT_LAST_PUBLICATION_PROTOCOL,
)
if (
phenotypic.get(PhenotypicAttr.PUBLICATION_PROTOCOL)
!= ROOT_LAST_PUBLICATION_PROTOCOL
):
return None
digest = hashlib.sha256(b"phenotypic-root-publication\x00v2\x00")
digest.update(len(raw).to_bytes(8, "big"))
digest.update(raw)
digest.update(after.st_mtime_ns.to_bytes(8, "big", signed=True))
digest.update(after.st_ctime_ns.to_bytes(8, "big", signed=True))
digest.update(after.st_ino.to_bytes(8, "big", signed=False))
return f"sha256-root-publish-v2:{digest.hexdigest()}"
def _store_revision_snapshot(
store: Path,
*,
root_json: str,
) -> tuple[tuple[tuple[str, bytes, int, int], ...], tuple[int, int] | None]:
"""Capture one symlink-free stat snapshot for ``store``."""
try:
root_stat = store.lstat()
except OSError as exc:
raise OSError("OME-Zarr store cannot be inspected") from exc
if stat_module.S_ISLNK(root_stat.st_mode):
raise OSError("OME-Zarr store cannot be a symlink")
if not stat_module.S_ISDIR(root_stat.st_mode):
raise OSError("OME-Zarr store must be a directory")
members: list[tuple[str, bytes, int, int]] = []
directories = [store]
while directories:
directory = directories.pop()
try:
entries = sorted(directory.iterdir(), key=lambda item: item.name)
except OSError as exc:
raise OSError("OME-Zarr store member cannot be inspected") from exc
child_directories: list[Path] = []
for entry in entries:
try:
entry_stat = entry.lstat()
except OSError as exc:
raise OSError(
"OME-Zarr store member cannot be inspected"
) from exc
relative = entry.relative_to(store).as_posix()
mode = entry_stat.st_mode
if stat_module.S_ISLNK(mode):
raise OSError(
f"OME-Zarr store contains symlink member: {relative}"
)
if stat_module.S_ISDIR(mode):
member_type = b"d"
child_directories.append(entry)
elif stat_module.S_ISREG(mode):
member_type = b"f"
else:
raise OSError(
f"OME-Zarr store contains non-regular member: {relative}"
)
members.append(
(
relative,
member_type,
entry_stat.st_size,
entry_stat.st_mtime_ns,
)
)
directories.extend(reversed(child_directories))
members.sort(key=lambda item: item[0])
root_record = next(
(
(size, mtime_ns)
for relative, member_type, size, mtime_ns in members
if relative == root_json and member_type == b"f"
),
None,
)
return tuple(members), root_record
[docs]
def overlays_dir(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/overlays/`` — the overlay package root."""
return deliverables_dir(output_dir) / DIR_OVERLAYS
[docs]
def dataset_overlays_dir(output_dir: Path, dataset: str) -> Path:
"""Return ``<output>/deliverables/overlays/<dataset>/``."""
return overlays_dir(output_dir) / dataset
[docs]
def measurements_by_feature_dir(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/measurements_by_feature/``."""
return deliverables_dir(output_dir) / DIR_MEASUREMENTS_BY_FEATURE
[docs]
def logs_dir(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/logs/``."""
return phenotypic_cache_dir(output_dir) / DIR_LOGS
[docs]
def chunks_dir(progress_dir_: Path) -> Path:
"""Return ``<progress>/chunks/`` for the dashboard chunk parquets."""
return progress_dir_ / DIR_CHUNKS
[docs]
def recompile_dir(progress_dir_: Path) -> Path:
"""Return ``<progress>/recompile/``."""
return progress_dir_ / DIR_RECOMPILE
[docs]
def recompile_status_dir(progress_dir_: Path) -> Path:
"""Return ``<progress>/recompile/status/``."""
return recompile_dir(progress_dir_) / DIR_RECOMPILE_STATUS
[docs]
def aggregation_shard_dir(
output_dir: Path, scheduler_epoch: str | None
) -> Path:
"""Return ``<progress>/aggregation_shards/<scheduler_epoch>/``.
Spec §7.5: the fan-out's measurement shards are per-invocation scratch, so
a prior run's shards can never be merged into this run's master. Recompile
already namespaces its shards this way under
``recompile/attempts/<attempt_id>/``; this generalises the pattern to the
forward path.
**The namespace is not the correctness argument, and must not be read as
one.** ``_scheduler_epoch`` returns ``None`` for every local run, so
consecutive local invocations share :data:`LOCAL_SCHEDULER_EPOCH` and would
collide. The fan-out therefore *empties* this directory when it starts, on
both drivers, at the same logical point -- which is strictly stronger than
namespacing, since namespacing also leaves every prior run's shards on disk
accumulating forever. The epoch stays in the path because it costs nothing
and keeps one path shape across the two drivers.
Pure path expression; callers ``mkdir`` when they intend to write.
Args:
output_dir: Run output root.
scheduler_epoch: The active SLURM lifecycle generation, or ``None``
for a local run.
Returns:
The shard directory for this invocation.
"""
return (
progress_dir(output_dir)
/ DIR_AGGREGATION_SHARDS
/ (scheduler_epoch or LOCAL_SCHEDULER_EPOCH)
)
[docs]
def task_status_path(output_dir: Path, task_index: int) -> Path:
"""Return ``<progress>/recompile/status/task_<idx>.json``."""
return recompile_status_dir(
progress_dir(output_dir)
) / task_status_filename(task_index)
[docs]
def chunk_parquet_path(progress_dir_: Path, chunk_id: int) -> Path:
"""Return ``<progress>/chunks/chunk_<id:03d>.parquet``."""
return chunks_dir(progress_dir_) / chunk_parquet_filename(chunk_id)
[docs]
def checkpoint_lock_path(
progress_dir_: Path, checkpoint_type: CheckpointType
) -> Path:
"""Return ``<progress>/.{checkpoint_type}_lock``."""
return progress_dir_ / checkpoint_lock_filename(checkpoint_type)
[docs]
def gui_launch_owner_path(output_dir: Path) -> Path:
"""Return the canonical GUI launch-generation owner record path."""
return progress_dir(output_dir) / GUI_LAUNCH_OWNER_JSON
[docs]
def slurm_lifecycle_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/progress/slurm_lifecycle.json``.
The mutable active-generation fence. ``_cli_slurm_lifecycle`` owns every
*write*; this helper exists because the run-state reader must be able to
ask who is in flight without importing a writer (INV-LAYER).
"""
return progress_dir(output_dir) / SLURM_LIFECYCLE_JSON
[docs]
def run_completion_marker_path(output_dir: Path) -> Path:
"""Return the canonical generation-bearing completion marker path."""
return progress_dir(output_dir) / RUN_COMPLETION_JSON
[docs]
def aggregate_publication_marker_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/aggregate_publication.json``."""
return phenotypic_cache_dir(output_dir) / AGGREGATE_PUBLICATION_JSON
[docs]
def image_completion_marker_path(
output_dir: Path, dataset: str, image_stem: str
) -> Path:
"""Return the general marker path for one dataset image stem."""
return progress_dir(output_dir) / DIR_IMAGE_COMPLETE / dataset / (
f"{image_stem}.json"
)
[docs]
def image_record_path(output_dir: Path, dataset: str, image_stem: str) -> Path:
"""Return ``<output>/.phenotypic/progress/images/<ds>/<stem>.json``."""
return progress_dir(output_dir) / DIR_IMAGE_RECORDS / dataset / (
f"{image_stem}.json"
)
[docs]
def failures_jsonl_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/progress/failures.jsonl``."""
return progress_dir(output_dir) / FAILURES_JSONL
[docs]
def manifest_json_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/progress/manifest.json``."""
return progress_dir(output_dir) / MANIFEST_JSON
[docs]
def chunk_manifest_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/progress/chunk_manifest.json``."""
return progress_dir(output_dir) / CHUNK_MANIFEST_JSON
[docs]
def chunk_state_path(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/progress/chunk_state.json``."""
return progress_dir(output_dir) / CHUNK_STATE_JSON
[docs]
def processing_report_html_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/processing_report.html``."""
return deliverables_dir(output_dir) / PROCESSING_REPORT_HTML
[docs]
def readme_md_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/README.md``."""
return deliverables_dir(output_dir) / README_MD
[docs]
def analysis_full_parquet_path(progress_dir_: Path) -> Path:
"""Return the legacy-named rolling measurement aggregation state.
Takes a *progress_dir* (not the run output root) since this file lives
inside ``progress/``. It is an internal resumable-run artifact, not a
static-analysis sidecar or user-facing output.
"""
return progress_dir_ / ANALYSIS_FULL_PARQUET
[docs]
def sentinel_resubmitted_path(progress_dir_: Path) -> Path:
"""Return ``<progress>/sentinel_resubmitted`` marker file path."""
return progress_dir_ / SENTINEL_RESUBMITTED_MARKER
[docs]
def chunk_lock_path(progress_dir_: Path) -> Path:
"""Return ``<progress>/.chunk_lock``."""
return progress_dir_ / CHUNK_LOCK
[docs]
def slurm_scripts_dir(output_dir: Path) -> Path:
"""Return ``<output>/.phenotypic/slurm_scripts/``."""
return phenotypic_cache_dir(output_dir) / DIR_SLURM_SCRIPTS
[docs]
def qc_dir(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/qc/`` — durable QC + curation state.
Relocated under ``deliverables/`` so a deliverables bundle is
self-contained and portable. For reads that must honour the legacy root
``<output>/qc/`` of pre-relocation runs, use
:attr:`BundleLayout.qc_dir`, which resolves the same three branches and
is the one with callers. A second module-level resolver existed here and
was deleted in P6 Task 7 -- two implementations of one fallback, only one
of them reachable.
"""
return deliverables_dir(output_dir) / DIR_QC
def _legacy_qc_dir(output_dir: Path) -> Path:
"""Pre-relocation location: ``<output>/qc/``."""
return output_dir / DIR_QC
[docs]
def qc_duckdb_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/qc/qc.duckdb``."""
return qc_dir(output_dir) / QC_DUCKDB
[docs]
def qc_review_state_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/qc/review_state.json`` (GUI-owned review progress)."""
return qc_dir(output_dir) / QC_REVIEW_STATE_JSON
[docs]
def errors_dir(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/errors/`` (per-category error parquets)."""
return deliverables_dir(output_dir) / DIR_ERRORS
[docs]
def error_category_parquet_path(output_dir: Path, category: str) -> Path:
"""Return ``<output>/deliverables/errors/<category>.parquet``.
Args:
output_dir: Run output directory.
category: A bare, already-sanitized category token (e.g.
``"background_noise"``). The caller is responsible for sanitization.
"""
return errors_dir(output_dir) / f"{category}.parquet"
[docs]
def error_analysis_parquet_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/error_analysis.parquet``."""
return deliverables_dir(output_dir) / ERROR_ANALYSIS_PARQUET
[docs]
def error_analysis_csv_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/error_analysis.csv``."""
return deliverables_dir(output_dir) / ERROR_ANALYSIS_CSV
[docs]
def error_analysis_html_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/error_analysis.html``."""
return deliverables_dir(output_dir) / ERROR_ANALYSIS_HTML
[docs]
def verified_parquet_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/verified.parquet`` (GUI-written, §9)."""
return deliverables_dir(output_dir) / VERIFIED_PARQUET
[docs]
def curation_labels_parquet_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/qc/curation_labels.parquet`` (durable labels store)."""
return qc_dir(output_dir) / CURATION_LABELS_PARQUET
[docs]
def custom_categories_json_path(output_dir: Path) -> Path:
"""Return ``<output>/deliverables/qc/custom_categories.json`` (custom-category registry)."""
return qc_dir(output_dir) / CUSTOM_CATEGORIES_JSON
# ---------------------------------------------------------------------------
# Loader / reader helpers (consolidate 3 duplicated read sites each)
# ---------------------------------------------------------------------------
[docs]
def read_run_manifest(output_dir: Path) -> Optional[dict]:
"""Read the run manifest if present, resolving legacy layouts.
Reads ``<output>/.phenotypic/progress/manifest.json``, falling back to the
pre-migration ``<output>/progress/manifest.json`` for legacy runs (via
:func:`resolve_manifest_json_path`). Replaces 4 inline
``json.loads(manifest_path.read_text())`` blocks.
Args:
output_dir: Run output directory containing ``progress/``.
Returns:
Parsed manifest dict, or :data:`None` when the file is missing
or unparseable (callers can decide whether absence is fatal).
"""
path = resolve_manifest_json_path(output_dir)
if not path.exists():
return None
try:
return json.loads(path.read_text())
except (json.JSONDecodeError, OSError):
logger.warning(
"Failed to parse %s; treating as missing", path, exc_info=True
)
return None
[docs]
def resolve_execution_mode(job_meta: Optional[dict]) -> ExecutionMode:
"""Extract :data:`ExecutionMode` from job metadata, defaulting to ``"local"``.
Replaces a 5-site copy-paste of the
``job_meta.get("execution_mode", "local") if job_meta else "local"``
pattern.
**Silent coercion:** any value that isn't exactly ``"slurm"`` collapses
to ``"local"`` — including ``None`` (no metadata file), ``{}`` (no
key), garbage strings (e.g. ``"validate"``, ``""``), and the literal
``None`` value. The function never raises. Callers who need to detect
an unknown mode and warn / refuse should inspect ``job_meta`` directly
before calling this helper.
Args:
job_meta: Parsed ``.phenotypic/progress/job_metadata.json`` content, or
:data:`None` when the file is absent.
Returns:
``"local"`` or ``"slurm"``.
"""
if not job_meta:
return "local"
raw = job_meta.get(JobMetadataKey.EXECUTION_MODE, "local")
if raw == "slurm":
return "slurm"
return "local"
# ---------------------------------------------------------------------------
# Cross-file JSON contract keys
# ---------------------------------------------------------------------------
[docs]
class DashboardManifestKey:
"""Keys inside ``<output>/.phenotypic/progress/manifest.json``.
The manifest is built by :func:`_cli._dashboard._manifest_builder.build_manifest`
and consumed by both the dashboard JS and the GUI run-console's runs
registry (``_runs_registry.py``). Writers and readers must reference
these constants rather than spelling the bare string.
"""
#: **Written, never read -- and that is the correct state for it.** A
#: format version exists to be readable by something that does not exist
#: yet, so "zero readers" is the expected condition of a healthy one, not
#: evidence of death. P6 Task 7's deletion ledger listed it as dead on a
#: zero-reader count; it was removed and restored, because dropping it
#: leaves the manifest with no schema discriminator and that cannot be
#: added retroactively to trees written meanwhile. The criterion is right
#: for a function and wrong for a format version.
VERSION: Final[str] = "version"
LAST_UPDATED: Final[str] = "last_updated"
EXECUTION_MODE: Final[str] = "execution_mode"
TOTAL_IMAGES: Final[str] = "total_images"
COMPLETED: Final[str] = "completed"
FAILED: Final[str] = "failed"
STARTED: Final[str] = "started"
PENDING: Final[str] = "pending"
SUCCESS_RATE: Final[str] = "success_rate"
IS_COMPLETE: Final[str] = "is_complete"
START_TIME: Final[str] = "start_time"
#: Exact GUI generation that published this canonical local manifest.
#: Omitted from non-GUI and scheduler manifests for compatibility.
GUI_RECORD_GENERATION: Final[str] = "gui_record_generation"
PROCESSING_GENERATION: Final[str] = "processing_generation"
EVENT_DIAGNOSTICS: Final[str] = "event_diagnostics"
INPUT_PATH: Final[str] = "input_path"
DATASETS: Final[str] = "datasets"
FAILURE_CATEGORIES: Final[str] = "failure_categories"
SLURM_INFO: Final[str] = "slurm_info"
[docs]
class DashboardManifestSlurmInfoKey:
"""Keys inside the ``slurm_info`` sub-dict of the dashboard manifest.
Distinct from :class:`JobMetadataKey`, even when string values overlap —
these describe the *manifest* contract, not the job-metadata sidecar.
"""
CHUNK_SCRIPTS: Final[str] = "chunk_scripts"
TOTAL_CHUNKS: Final[str] = "total_chunks"
CHUNK_JOB_IDS: Final[str] = "chunk_job_ids"
ACTIVE_CHUNKS: Final[str] = "active_chunks"
COMPLETED_CHUNKS: Final[str] = "completed_chunks"
PENDING_CHUNKS: Final[str] = "pending_chunks"
[docs]
class ChunkStateKey:
"""Keys inside ``<output>/.phenotypic/progress/chunk_state.json``."""
CHUNKED_FILES: Final[str] = "chunked_files"
NEXT_CHUNK_ID: Final[str] = "next_chunk_id"
[docs]
class ProcessingStateKey:
"""Keys inside ``<output>/.phenotypic/processing_state.json``.
Distinct from :class:`JobMetadataKey` even where string values overlap
(e.g. ``EXECUTION_MODE``, ``INPUT_PATH``) — these describe the
`processing_state.json` contract, not the SLURM job metadata sidecar.
Some values intentionally match across the two contracts so that a
single field (like ``execution_mode``) can be migrated atomically;
the ``test_processing_state_keys_match_job_metadata_keys`` regression
test asserts the overlap.
"""
VERSION: Final[str] = "version"
PIPELINE_PATH: Final[str] = "pipeline_path"
INPUT_PATH: Final[str] = "input_path"
OUTPUT_DIR: Final[str] = "output_dir"
TIMESTAMP: Final[str] = "timestamp"
EXECUTION_MODE: Final[str] = "execution_mode"
LAST_UPDATED: Final[str] = "last_updated"
DATASETS: Final[str] = "datasets"
CONFIG: Final[str] = "config"
# Per-dataset state-dict sub-keys
COMPLETED: Final[str] = "completed"
FAILED: Final[str] = "failed"
STARTED: Final[str] = "started"
ERRORS: Final[str] = "errors"
INITIAL_IMAGES: Final[str] = "initial_images"
[docs]
class ChunkManifestKey:
"""Keys inside ``<output>/.phenotypic/progress/chunk_manifest.json``."""
CHUNKS: Final[str] = "chunks"
ROWS: Final[str] = "rows"
DATASETS: Final[str] = "datasets"
TOTAL_ROWS: Final[str] = "total_rows"
NAME: Final[str] = "name"
# ---------------------------------------------------------------------------
# Module-path constants for importlib dispatch
# ---------------------------------------------------------------------------
[docs]
class ModulePath:
"""Importable module paths used in dynamic ``importlib.import_module`` dispatch.
Spelled out here so a renamed sub-package fails at type-check time
(consumers reference ``ModulePath.POST`` — a typo there is caught
by mypy) rather than silently at runtime.
"""
POST: Final[str] = "phenotypic.post"
ANALYSIS: Final[str] = "phenotypic.analysis"
# ---------------------------------------------------------------------------
# Environment variable names
# ---------------------------------------------------------------------------
[docs]
class EnvVar:
"""Environment variable names read or set by the CLI.
SLURM injects these into batch scripts; the CLI reads them to
discover its execution context (job id, array task id, …) and
to find node-local scratch storage.
"""
SCRATCH: Final[str] = "SCRATCH"
SLURM_JOB_ID: Final[str] = "SLURM_JOB_ID"
SLURM_ARRAY_JOB_ID: Final[str] = "SLURM_ARRAY_JOB_ID"
SLURM_ARRAY_TASK_ID: Final[str] = "SLURM_ARRAY_TASK_ID"
SLURM_ARRAY_TASK_COUNT: Final[str] = "SLURM_ARRAY_TASK_COUNT"
SLURM_CPUS_PER_TASK: Final[str] = "SLURM_CPUS_PER_TASK"
SLURM_MEM_PER_NODE: Final[str] = "SLURM_MEM_PER_NODE"
# ---------------------------------------------------------------------------
# Store image-class reader
# ---------------------------------------------------------------------------
[docs]
def load_image_from_store(
store_path: Path,
*,
fallback: ImageTypeName = "Image",
) -> "_Image | _GridImage":
"""Read ``phenotypic.image_class`` from a store root and dispatch the loader.
Dispatches on ``image_class`` (``Image`` / ``GridImage``), which is the
loader-dispatch field. It is **not** ``Metadata_ImageType``, which is
user-visible schema metadata and may be ``GridSection`` on a plain
:class:`Image`.
**Bypasses the public :meth:`Image.load_zarr` guard, deliberately.** That
guard refuses a store carrying no ``image_class``, because a *user* calling
the public verb on such a store has almost certainly mistaken a
``--mode process`` export for a run bundle and wants ``imread`` instead.
This function is the internal dispatcher: its caller supplies *fallback*
and has therefore already made that determination itself
(``_cli_process_single`` passes the run's own image type; the tune CLI
passes ``"GridImage"``). Routing through ``load_zarr`` would raise before
the resolved class could ever be used, making *fallback* dead code. So the
resolved class is asked to load the store directly.
A store with no bundle *content* still fails, one layer down and by its
own error: ``_load_from_store`` subscripts the series mapping bare at
``series["gray"]`` and ``series["detect_mat"]``
(``_image_io_handler.py``), so a single-series process store raises
``KeyError: 'detect_mat'``.
Args:
store_path: Path to a ``*.ome.zarr`` directory.
fallback: Class name used when the block carries no ``image_class``.
Returns:
An :class:`Image` or :class:`GridImage` loaded from the store.
Raises:
KeyError: If the store root carries no ``phenotypic`` block, or if it
carries one but no bundle series.
ValueError: If ``store_schema_version`` is not this build's.
"""
from phenotypic import (
GridImage,
Image,
) # lazy: avoids circular import at module load
from phenotypic.sdk_.ngff_ import PhenotypicAttr, require_readable_store
# `require_readable_store`, not `read_phenotypic_attributes`: bypassing
# `load_zarr` must not also bypass the store_schema_version gate it
# applied. One read serves both the dispatch and the load.
block = require_readable_store(store_path)
class_name = block.get(PhenotypicAttr.IMAGE_CLASS, fallback)
# See ``_hdf_to_zarr._load_image_from_hdf``: the comparison is against
# the class name the writer recorded, not against ``IMAGE_TYPES.GRID`` -- that enum is the
# ``Metadata_ImageType`` vocabulary, a different field that spec 2.1 keeps
# deliberately independent of ``image_class``.
image_cls = GridImage if class_name == GridImage.__name__ else Image
return image_cls._load_from_store(store_path, block)
# ---------------------------------------------------------------------------
# BundleLayout — resolved on-disk topology value object
# ---------------------------------------------------------------------------
[docs]
@dataclass(frozen=True)
class BundleLayout:
"""Resolved on-disk topology of a run output or a standalone deliverables bundle.
Separates the *deliverables base* (the folder directly holding
``master_measurements.parquet``) from the optional *output root* (the parent
that also holds ``results/`` and ``.phenotypic/``). A standalone bundle has
``output_root is None``; deliverables-internal artefacts always resolve from
``deliverables_base`` so the bundle is portable.
Attributes:
deliverables_base: Folder containing ``master_measurements.parquet``.
output_root: Parent run directory holding ``results/`` + machine state,
or ``None`` for a standalone (deliverables-only) bundle.
"""
deliverables_base: Path
output_root: Optional[Path]
[docs]
@classmethod
def detect(cls, path: Path) -> "BundleLayout":
"""Classify ``path`` as a run output dir or a standalone deliverables bundle.
Case 1 — ``path`` directly holds ``master_measurements.parquet``: treat it
as the deliverables base. Promote ``path.parent`` to ``output_root`` ONLY
when ``path`` is literally named ``deliverables`` AND a sibling ``results/``
exists (the "pointed at the deliverables subdir of a full run" case); this
guard stops a renamed standalone bundle from adopting an unrelated sibling
``results/``.
Case 2 — ``path`` contains ``deliverables/master_measurements.parquet``:
``deliverables_base = path/deliverables`` and ``output_root = path``.
Args:
path: Either a run output directory (containing a ``deliverables/``
subdirectory) or a standalone deliverables folder (directly
containing ``master_measurements.parquet``).
Returns:
A :class:`BundleLayout` with resolved ``deliverables_base`` and
``output_root``.
Raises:
FileNotFoundError: ``path`` is neither a run output directory nor a
deliverables bundle.
"""
path = Path(path).resolve()
if (path / MASTER_MEASUREMENTS_PARQUET).is_file():
output_root: Optional[Path] = None
if (
path.name == DIR_DELIVERABLES
and (path.parent / DIR_RESULTS).is_dir()
):
output_root = path.parent
return cls(deliverables_base=path, output_root=output_root)
if (path / DIR_DELIVERABLES / MASTER_MEASUREMENTS_PARQUET).is_file():
return cls(
deliverables_base=path / DIR_DELIVERABLES, output_root=path
)
raise FileNotFoundError(
f"{path} is neither a deliverables bundle nor a run output directory "
f"containing {DIR_DELIVERABLES}/{MASTER_MEASUREMENTS_PARQUET}. Point the "
"viewer at a `python -m phenotypic` output dir or a deliverables/ folder."
)
# -- capability ---------------------------------------------------------
@property
def has_results(self) -> bool:
"""Return ``True`` when a ``results/`` directory exists under the output root."""
return (
self.output_root is not None
and (self.output_root / DIR_RESULTS).is_dir()
)
@property
def results_dir(self) -> Optional[Path]:
"""Return the ``results/`` directory, or ``None`` for a standalone bundle."""
if self.output_root is None:
return None
results = self.output_root / DIR_RESULTS
return results if results.is_dir() else None
[docs]
def store_path(self, dataset: str, stem: str) -> Optional[Path]:
"""Full-res per-image OME-Zarr store for ``(dataset, stem)``, or ``None``.
Args:
dataset: Dataset name (subdirectory under ``results/``).
stem: Image stem (filename without extension).
Returns:
Resolved store path if the **directory** exists, otherwise ``None``.
Note the ``is_dir`` check: a store is a directory, so the
``is_file`` test the removed ``hdf_path`` used (it resolved a single
per-image HDF file) would always return ``None`` here.
"""
if self.output_root is None:
return None
candidate = zarr_store_path(self.output_root, dataset, stem)
return candidate if candidate.is_dir() else None
# -- deliverables-anchored artefacts ------------------------------------
@property
def master_parquet(self) -> Path:
"""Return path to ``master_measurements.parquet`` in the deliverables base."""
return self.deliverables_base / MASTER_MEASUREMENTS_PARQUET
@property
def mirror_parquet(self) -> Path:
"""Return path to ``measurements.parquet`` (post-applied mirror)."""
return self.deliverables_base / MEASUREMENTS_PARQUET
@property
def mirror_csv(self) -> Path:
"""Return path to ``measurements.csv`` (post-applied mirror)."""
return self.deliverables_base / MEASUREMENTS_CSV
@property
def plots_dir(self) -> Path:
"""Return the resolved ``plots/`` directory inside the bundle."""
return self.deliverables_base / DIR_PLOTS
@property
def pipeline_config_path(self) -> Path:
"""Return path to ``pipeline.json`` in the deliverables base."""
return self.deliverables_base / PIPELINE_JSON
@property
def resolved_pipeline_config_path(self) -> Path:
"""Return the best existing pipeline config path inside the bundle.
Mirrors :func:`resolve_pipeline_config_path`'s precedence but anchored
on :attr:`deliverables_base` (so a standalone bundle resolves *inside
itself* without double-joining ``deliverables/``): the canonical typed
config when present, else the legacy plain ``pipeline.json`` when
present, else the canonical path (so writers naturally create typed
config files).
"""
canonical = self.pipeline_config_path
if canonical.exists():
return canonical
legacy = self.deliverables_base / _LEGACY_PIPELINE_JSON
if legacy.exists():
return legacy
return canonical
@property
def qc_dir(self) -> Path:
"""Return the QC directory, resolving legacy ``<output>/qc/`` layouts.
Prefers ``deliverables/qc/`` when it exists. Falls back to the
legacy root ``<output>/qc/`` when only that exists (pre-relocation
runs). Returns the canonical ``deliverables/qc/`` path for fresh
writes when neither is present.
"""
canonical = self.deliverables_base / DIR_QC
if canonical.exists():
return canonical
if self.output_root is not None:
legacy = self.output_root / DIR_QC
if legacy.exists():
return legacy
return canonical
@property
def qc_duckdb(self) -> Path:
"""Return path to ``qc/qc.duckdb`` (the QC analysis database)."""
return self.qc_dir / QC_DUCKDB
@property
def qc_review_state_path(self) -> Path:
"""Return path to ``deliverables/qc/review_state.json`` (GUI-owned review progress)."""
return self.qc_dir / QC_REVIEW_STATE_JSON
@property
def curation_labels_parquet(self) -> Path:
"""Return path to ``deliverables/qc/curation_labels.parquet`` (durable labels store)."""
return self.qc_dir / CURATION_LABELS_PARQUET
@property
def custom_categories_json(self) -> Path:
"""Return path to ``deliverables/qc/custom_categories.json`` (custom-category registry)."""
return self.qc_dir / CUSTOM_CATEGORIES_JSON
@property
def errors_dir(self) -> Path:
"""Return path to the ``errors/`` directory under the deliverables base."""
return self.deliverables_base / DIR_ERRORS
@property
def error_analysis_parquet(self) -> Path:
"""Return path to ``error_analysis.parquet`` in the deliverables base."""
return self.deliverables_base / ERROR_ANALYSIS_PARQUET
@property
def error_analysis_csv(self) -> Path:
"""Return path to ``error_analysis.csv`` in the deliverables base."""
return self.deliverables_base / ERROR_ANALYSIS_CSV
@property
def error_analysis_html(self) -> Path:
"""Return path to ``error_analysis.html`` in the deliverables base."""
return self.deliverables_base / ERROR_ANALYSIS_HTML
@property
def verified_parquet(self) -> Path:
"""Return path to ``verified.parquet`` (GUI-written verified-good archive)."""
return self.deliverables_base / VERIFIED_PARQUET
[docs]
def error_category_parquet(self, category: str) -> Path:
"""Return path to ``errors/<category>.parquet``.
Args:
category: Bare, already-sanitized category token
(e.g. ``"background_noise"``).
Returns:
Path to the per-category error parquet file.
"""
return self.errors_dir / f"{category}.parquet"
[docs]
def overlays_dir(self, dataset: str) -> Path:
"""Return path to ``overlays/<dataset>/``.
Args:
dataset: Dataset name.
Returns:
Directory path for overlay PNGs of the given dataset.
"""
return self.deliverables_base / DIR_OVERLAYS / dataset
[docs]
def overlay_path(self, dataset: str, stem: str) -> Path:
"""Return path to ``overlays/<dataset>/<stem>.png``.
Args:
dataset: Dataset name.
stem: Image stem (filename without extension).
Returns:
Path to the overlay PNG for the given image.
"""
return self.overlays_dir(dataset) / f"{stem}.png"
