Batch CLI

cli

metrics-petri CLI — batch pipeline without GUI.

Command example

metrics-petri INPUT_DIR --output OUT.zip --threshold 0.5 metrics-petri INPUT_DIR --metadata image_metadata.csv --model path/to/model.pt

IMAGE_EXTS module-attribute

IMAGE_EXTS = {
    ".jpg",
    ".jpeg",
    ".png",
    ".tif",
    ".tiff",
    ".bmp",
    ".webp",
    ".gif",
    ".heic",
    ".heif",
    ".dng",
    ".cr2",
    ".nef",
    ".arw",
    ".raf",
    ".orf",
    ".rw2",
}

DEFAULT_DISH_SIZE_MM module-attribute

DEFAULT_DISH_SIZE_MM = 90.0

run_batch

run_batch(
    input_dir,
    output_zip,
    threshold=0.5,
    metadata_csv=None,
    dish_size_mm=DEFAULT_DISH_SIZE_MM,
    seed=0,
)

Source code in metrics_petri/pipelinesam/cli.py

def run_batch(
    input_dir: Path,
    output_zip: Path,
    threshold: float = 0.5,
    metadata_csv: Path | None = None,
    dish_size_mm: float = DEFAULT_DISH_SIZE_MM,
    seed: int | None = 0,
) -> None:
    _set_seed(seed)

    if metadata_csv and metadata_csv.exists():
        meta_df = _load_metadata(metadata_csv)
        tasks = _build_metadata_tasks(input_dir, meta_df)
        if meta_df.shape[0] and not tasks:
            raise FileNotFoundError(
                f"No metadata image_path entries in {metadata_csv} matched files "
                f"under {input_dir}. Update the metadata file, pass a matching "
                "--metadata file, or remove/rename the metadata file to scan images directly."
            )
    else:
        paths = _find_images(input_dir)
        # Auto-detect image dates; anchor experiment_date to earliest found date
        raw_dates = {p: _detect_image_date(p) for p in paths}
        known = [d for d in raw_dates.values() if d]
        auto_exp_date = min(known) if known else ""
        tasks = [
            (
                p,
                {
                    "image_path": str(p.relative_to(input_dir)),
                    "image_date": raw_dates[p],
                    "experiment_date": auto_exp_date,
                },
            )
            for p in paths
        ]

    if not tasks:
        raise FileNotFoundError(f"No images found in {input_dir}")

    print(f"Processing {len(tasks)} image(s) …", flush=True)

    results: list[dict] = []
    all_overlays: dict[str, Image.Image] = {}

    for i, (img_path, meta) in enumerate(tasks, 1):
        print(f"  [{i}/{len(tasks)}] {img_path.name}", flush=True)
        try:
            row, overlays = _process_image(
                img_path, meta, threshold, dish_size_mm=dish_size_mm
            )
            results.append(row)
            all_overlays.update(overlays)
        except Exception as exc:
            print(f"    WARNING: {exc}", file=sys.stderr, flush=True)
            results.append({**meta, "error": str(exc)})

    def _dc_to_num(s: object, fallback: int = 0) -> int:
        if isinstance(s, str) and s.startswith("d") and s[1:].isdigit():
            return int(s[1:])
        return fallback

    df = pd.DataFrame(results)

    # days_since_start: prefer numeric value from day_code column (set by the user
    # in the GUI or metadata CSV); fall back to (image_date - experiment_date) arithmetic.
    if "day_code" in df.columns:
        dc_nums = df["day_code"].apply(_dc_to_num)
        df["days_since_start"] = dc_nums
    elif "image_date" in df.columns and "experiment_date" in df.columns:
        _id = pd.to_datetime(df["image_date"], errors="coerce")
        _ed = pd.to_datetime(df["experiment_date"], errors="coerce")
        df["days_since_start"] = (_id - _ed).dt.days.fillna(0).astype(int) + 1

    # Growth-rate calculations using actual calendar-day intervals between images
    if "image_date" in df.columns and "experiment_date" in df.columns:
        df["image_date"] = pd.to_datetime(df["image_date"], errors="coerce")
        df["experiment_date"] = pd.to_datetime(df["experiment_date"], errors="coerce")
        # Sort by day_code numeric value so rows are in chronological order
        if "days_since_start" in df.columns:
            sort_cols = (
                ["experiment_name", "days_since_start"]
                if "experiment_name" in df.columns
                else ["days_since_start"]
            )
        else:
            sort_cols = (
                ["experiment_name", "image_date"]
                if "experiment_name" in df.columns
                else ["image_date"]
            )
        df = df.sort_values(sort_cols)
        df["rgr_per_day"] = float("nan")
        df["relative_growth_per_day"] = float("nan")
        grp_col = "experiment_name" if "experiment_name" in df.columns else None
        groups = df.groupby(grp_col) if grp_col else [(None, df)]
        for _, g in groups:
            g = g.sort_values("days_since_start" if "days_since_start" in g.columns else "image_date")
            for j in range(1, len(g)):
                # Use actual calendar days for rate math, not just day-code difference
                try:
                    dd = int((g.iloc[j]["image_date"] - g.iloc[j - 1]["image_date"]).days)
                except Exception:
                    dd = int(g.iloc[j].get("days_since_start", j) - g.iloc[j - 1].get("days_since_start", j - 1))
                a1 = g.iloc[j - 1].get("area_mm2") or 0
                a2 = g.iloc[j].get("area_mm2") or 0
                if dd > 0 and a1 > 0 and a2 > 0:
                    df.loc[g.index[j], "rgr_per_day"] = (
                        math.log(float(a2)) - math.log(float(a1))
                    ) / dd
                    df.loc[g.index[j], "relative_growth_per_day"] = (
                        float(a2) - float(a1)
                    ) / dd

    # Write zip
    tmp = Path(tempfile.mkdtemp())
    csv_p = tmp / "analysis_full.csv"
    json_p = tmp / "analysis_full.json"
    df.to_csv(csv_p, index=False)
    df.to_json(json_p, orient="records", indent=2, date_format="iso")

    overlays_dir = tmp / "overlays"
    overlays_dir.mkdir()
    for name, img_pil in all_overlays.items():
        img_pil.save(overlays_dir / name, quality=92)

    charts_dir = tmp / "charts"
    _write_charts(df, charts_dir)

    from .pipeline import DEVICE
    provenance_p = tmp / "provenance.json"
    provenance_p.write_text(
        json.dumps(
            build_provenance(
                interface="metrics-petri-cli",
                threshold=threshold,
                dish_size_mm=dish_size_mm,
                device=DEVICE,
                seed=seed,
            ),
            indent=2,
        )
        + "\n",
        encoding="utf-8",
    )

    with zipfile.ZipFile(output_zip, "w", zipfile.ZIP_DEFLATED) as zf:
        zf.write(csv_p, "analysis_full.csv")
        zf.write(json_p, "analysis_full.json")
        zf.write(provenance_p, "provenance.json")
        for f in sorted(overlays_dir.glob("*.jpg")):
            zf.write(f, f"overlays/{f.name}")
        for f in sorted(charts_dir.glob("*.png")):
            zf.write(f, f"charts/{f.name}")

    ok = sum(1 for r in results if "error" not in r)
    print(f"\n✓  {ok}/{len(results)} images analysed")
    print(f"✓  Output: {output_zip}")

build_parser

build_parser()

Build the batch CLI argument parser.

Source code in metrics_petri/pipelinesam/cli.py

def build_parser() -> argparse.ArgumentParser:
    """Build the batch CLI argument parser."""
    parser = argparse.ArgumentParser(
        prog="metrics-petri",
        description="metrics-petri — petri dish colony analysis (no GUI)",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument(
        "input",
        nargs="?",
        default=None,
        help="Folder containing images (default: current directory)",
    )
    parser.add_argument("--output", "-o", default=None, help="Output zip file path")
    parser.add_argument(
        "--threshold",
        "-t",
        type=float,
        default=0.5,
        help="Mask confidence threshold (0–1)",
    )
    parser.add_argument(
        "--metadata",
        "-m",
        default=None,
        help="Path to image_metadata.csv or image_metadata.json",
    )
    parser.add_argument(
        "--model",
        default=None,
        help="Path to SmallUNet checkpoint (.pt)",
    )
    parser.add_argument(
        "--dish-size-mm",
        type=_positive_float,
        default=DEFAULT_DISH_SIZE_MM,
        help="Outside diameter of the Petri dish in millimetres",
    )
    parser.add_argument(
        "--seed",
        type=_non_negative_int,
        default=0,
        help="PyTorch random seed for defensive reproducibility",
    )
    return parser

main

main()

Source code in metrics_petri/pipelinesam/cli.py

def main() -> None:
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "doctor":
        _run_doctor()
        return
    args = build_parser().parse_args()

    if args.model:
        os.environ["UNET_MODEL"] = args.model

    input_dir = Path(args.input).expanduser().resolve() if args.input else Path.cwd()
    if not input_dir.is_dir():
        print(f"error: '{input_dir}' is not a directory", file=sys.stderr)
        sys.exit(1)

    output_zip = (
        Path(args.output).expanduser().resolve() if args.output else input_dir / "analysis.zip"
    )
    if output_zip.suffix.lower() != ".zip":
        output_zip = output_zip.with_suffix(".zip")

    if args.metadata:
        metadata_csv = Path(args.metadata).expanduser().resolve()
    elif (input_dir / "image_metadata.json").exists():
        metadata_csv = input_dir / "image_metadata.json"
    else:
        metadata_csv = input_dir / "image_metadata.csv"

    try:
        run_batch(
            input_dir,
            output_zip,
            threshold=args.threshold,
            metadata_csv=metadata_csv,
            dish_size_mm=args.dish_size_mm,
            seed=args.seed,
        )
    except FileNotFoundError as exc:
        print(f"error: {exc}", file=sys.stderr)
        sys.exit(1)