scripts/run_assessment.py

"""End-to-end CLI for the ergo-agentic pipeline.

Run a full assessment on one or more images. Pass image URLs (or local paths)
as arguments, or omit args to run on the sample images from the bundled
report-sample.json.

Usage:
    uv run python scripts/run_assessment.py
    uv run python scripts/run_assessment.py https://example.com/img1.jpg https://example.com/img2.jpg

Output: pretty-printed OutcomeMatrix JSON.
"""

from __future__ import annotations

import asyncio
import json
import sys
from pathlib import Path

from dotenv import load_dotenv

from ergo_agentic.graph import build_graph
from ergo_agentic.state import ImageInput

REPORT_SAMPLE_FILE = (
    Path(__file__).resolve().parents[1] / "docs" / "datasources" / "report-sample.json"
)


def _sample_image_urls() -> list[str]:
    with REPORT_SAMPLE_FILE.open() as f:
        return json.load(f).get("uploadedImages", [])


def _build_image_inputs(urls: list[str]) -> list[ImageInput]:
    return [
        {"image_id": f"img_{i}", "url": url, "label": None}
        for i, url in enumerate(urls, start=1)
    ]


def _print_run_header(image_count: int) -> None:
    from ergo_agentic.models import DEFAULT_MODEL_CONFIG as cfg

    print(f"Running pipeline on {image_count} image(s)...")
    print("Models:")
    print(f"  image_analyzer: {cfg.image_analyzer}")
    print(f"  vision_passes:  {list(cfg.vision_passes)}")
    print(f"  review_agent:   {cfg.review_agent}")
    print()


def _print_scene(matrix) -> None:
    sc = matrix.scene_config
    print("Scene:")
    print(f"  work_location: {sc.work_location.value if sc.work_location else 'unknown'}")
    print(f"  screen_count:  {sc.screen_count} ({', '.join(sc.screen_types) or 'none'})")
    print(f"  has_standing_desk: {sc.has_standing_desk}")
    print(f"  person_detected:   {sc.person_detected}")
    print()


def _print_accessories(matrix) -> None:
    if not matrix.detected_accessories:
        return
    print("Detected accessories:")
    for acc in matrix.detected_accessories:
        desc = f" — {acc.description}" if acc.description else ""
        print(f"  - {acc.type}{desc}")
    print()


def _print_assessments(matrix) -> None:
    print(f"Assessments ({len(matrix.assessments)}):")
    for a in matrix.assessments:
        flag = "✓" if a.is_good_habit else "⚠"
        print(f"  {flag} [{a.parameter_text}] {a.review_decision.value}")
        for o in a.final_outcomes:
            print(f"      → {o}")
        if a.evidence_summary:
            print(f"      evidence: {a.evidence_summary}")
        print(f"      images: {a.source_images} | models: {a.source_models}")
        print()


def _print_skipped(matrix) -> None:
    if not matrix.skipped_parameters:
        return
    print(f"Skipped ({len(matrix.skipped_parameters)}):")
    for s in matrix.skipped_parameters:
        print(f"  - {s.parameter_text}: {s.reason}")
    print()


def _print_matrix(matrix) -> None:
    print("=" * 60)
    print("OUTCOME MATRIX")
    print("=" * 60)
    print()
    _print_scene(matrix)
    _print_accessories(matrix)
    _print_assessments(matrix)
    _print_skipped(matrix)
    print("=" * 60)
    print("RAW JSON")
    print("=" * 60)
    print(matrix.model_dump_json(indent=2))


def main(argv: list[str]) -> int:
    load_dotenv()

    urls = argv[1:] if len(argv) > 1 else _sample_image_urls()
    if not urls:
        print("No images to process.", file=sys.stderr)
        return 1

    images = _build_image_inputs(urls)
    _print_run_header(len(images))

    graph = build_graph()
    final_state = asyncio.run(graph.ainvoke({"images": images}))

    _print_matrix(final_state["outcome_matrix"])
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))