app.py

"""
Agent 1 - Change Detection (Gradio version)
------------------------------------------------
Docker SDK requires a paid HF plan - this is the free-tier-compatible
rebuild using Gradio instead. Gradio Spaces still work perfectly as an API
for the gateway to call later: every Gradio app automatically exposes an
API endpoint (visible via the "Use via API" link at the bottom of the
deployed Space, and callable from Python with the `gradio_client` package)
in addition to the web UI - nothing about the gateway integration plan
actually changes, just how this Space is built.

Gradio's File component with type="filepath" hands the underlying function
a local temp file path directly - exactly what run_agent1() already
expects, so no manual file-saving code is needed the way the FastAPI
version required.

Run locally to test: python app.py
"""

import base64
import os

import gradio as gr

from change_detection import run_agent1, ITEM_TYPE_ANGLES


def _encode_image_to_data_url(image_path):
    """
    Reads an image file from this Space's own local disk and returns it as
    a base64 Data URL, so the calling backend can use it directly with no
    further processing - the crop_path field alone is useless to any
    caller outside this Space, since it only points to a file on THIS
    server's own filesystem. Format: "data:image/jpeg;base64,<data>",
    which most HTTP clients / image libraries can consume directly.
    """
    try:
        with open(image_path, "rb") as f:
            image_bytes = f.read()
        encoded = base64.b64encode(image_bytes).decode("utf-8")
        ext = os.path.splitext(image_path)[1].lower()
        mime_type = "image/png" if ext == ".png" else "image/jpeg"
        return f"data:{mime_type};base64,{encoded}"
    except Exception:
        return None

# FIX: this Space was created on Hugging Face's ZeroGPU hardware tier, which
# refuses to start unless at least one function is decorated with
# @spaces.GPU - even though Agent 1 doesn't actually need a GPU at all
# (classical CV, runs fine on CPU). Downgrading the Space to CPU Basic
# turned out to require a paid HF plan once a Space is already created as
# ZeroGPU, so this decorator is the practical fix instead: it satisfies
# ZeroGPU's startup check without changing what the function actually does.
# `spaces` is an HF-Space-specific package (not installed locally), so this
# falls back to a no-op decorator for local testing with `python app.py`.
try:
    import spaces
    gpu_decorator = spaces.GPU
except ImportError:
    def gpu_decorator(func):
        return func


@gpu_decorator
def process(item_id, item_type, baseline_files, return_files, complaint_text):
    """
    Core logic, kept as a plain function so it's testable independently of
    Gradio's UI wiring.

    FIX: no longer takes a manually-typed angles JSON string - the required
    angles are already fully determined by item_type via ITEM_TYPE_ANGLES,
    so asking the user to also type them was both redundant and fragile
    (found via real testing: users naturally type things like "lid_exterior,
    screen_keyboard" without brackets/quotes, which isn't valid JSON and
    just produces a confusing error). Angles are now derived automatically
    and shown to the user via show_required_angles() below, instead of
    being something they have to get exactly right by hand.
    """
    if not item_type or item_type not in ITEM_TYPE_ANGLES:
        return {"error": f"Unknown item_type '{item_type}'. Known types: {list(ITEM_TYPE_ANGLES.keys())}"}

    angle_list = ITEM_TYPE_ANGLES[item_type]

    baseline_files = baseline_files or []
    return_files = return_files or []

    if len(baseline_files) != len(angle_list) or len(return_files) != len(angle_list):
        return {"error": f"item_type '{item_type}' requires {len(angle_list)} angle(s), in this exact order: "
                          f"{angle_list}. Got {len(baseline_files)} baseline file(s) and {len(return_files)} "
                          f"return file(s) - please upload exactly one baseline and one return photo per "
                          f"angle, in the order shown above the upload boxes."}

    baseline_photos = [{"angle": angle, "path": f.name if hasattr(f, "name") else f}
                        for angle, f in zip(angle_list, baseline_files)]
    return_photos = [{"angle": angle, "path": f.name if hasattr(f, "name") else f}
                      for angle, f in zip(angle_list, return_files)]

    try:
        result = run_agent1(
            item_id=item_id or "unnamed_item", item_type=item_type,
            baseline_photos=baseline_photos, return_photos=return_photos,
            complaint_text=complaint_text or None, crop_output_dir="agent1_api_crops",
        )

        # FIX (real integration blocker): crop_path only points to a file on
        # THIS Space's own server - the calling backend has no way to fetch
        # it, since it's not a public URL. Add a base64 Data URL for every
        # region's crop image directly into the response instead, so the
        # backend can use the image with zero extra processing.
        for angle_result in result.get("angles", []):
            for region in angle_result.get("regions", []):
                crop_path = region.get("crop_path")
                if crop_path and os.path.isfile(crop_path):
                    region["crop_base64"] = _encode_image_to_data_url(crop_path)

        return result
    except Exception as e:
        return {"error": f"Agent 1 processing failed: {e}"}


def show_required_angles(item_type):
    """Updates a read-only label the moment the item type is picked, so the
    user knows exactly what order to upload photos in - replaces having to
    type that information themselves."""
    if not item_type or item_type not in ITEM_TYPE_ANGLES:
        return "Pick an item type to see which angles are required."
    angles = ITEM_TYPE_ANGLES[item_type]
    return (f"This item type requires {len(angles)} angle(s), in this exact order: "
            f"**{' -> '.join(angles)}**. Upload one baseline and one return photo per angle below, "
            f"in that order.")


with gr.Blocks(title="Agent 1 - Change Detection") as demo:
    gr.Markdown("# Agent 1 - Change Detection\n"
                "Detects whether a returned item's condition changed since pickup. "
                "No API key required - classical computer vision only.")

    with gr.Row():
        item_id_input = gr.Textbox(label="Item ID", value="test_item")
        # allow_custom_value=True: without this, Gradio's own Dropdown validation
        # rejects any unrecognized value with an unhandled crash before our own
        # process() validation ever runs (found via a real crash on Agent 4 with
        # the same pattern - fixed there, applying the same fix everywhere else).
        item_type_input = gr.Dropdown(label="Item Type", choices=list(ITEM_TYPE_ANGLES.keys()), allow_custom_value=True)

    angles_display = gr.Markdown("Pick an item type to see which angles are required.")
    item_type_input.change(fn=show_required_angles, inputs=item_type_input, outputs=angles_display)

    with gr.Row():
        baseline_input = gr.File(label="Baseline (pickup) photos, in the order shown above", file_count="multiple", type="filepath")
        return_input = gr.File(label="Return photos, in the same order", file_count="multiple", type="filepath")

    complaint_input = gr.Textbox(label="Complaint text (optional)")

    with gr.Row():
        submit_btn = gr.Button("Run Detection", variant="primary")
        clear_btn = gr.ClearButton(
            value="Clear All (start a fresh test)",
            components=[item_id_input, item_type_input, baseline_input, return_input, complaint_input, angles_display],
        )
    output = gr.JSON(label="Result")
    clear_btn.add(output)  # also reset the previous result, not just the inputs

    # FIX: found via real testing - Gradio's file upload components can retain a
    # previous selection across runs in the same browser session (a hard page
    # refresh fixed it once, but shouldn't be the expected workflow). This
    # button explicitly resets every input AND the previous output in one
    # click, so starting a genuinely fresh test doesn't depend on remembering
    # to reload the page.

    submit_btn.click(
        fn=process,
        inputs=[item_id_input, item_type_input, baseline_input, return_input, complaint_input],
        outputs=output,
        api_name="detect",  # this is what the gateway will call via gradio_client
    )

if __name__ == "__main__":
    demo.launch()