space_app.py

"""The Space entry point: the Animap service, plus a page a person can use.

Hugging Face runs this file, published to the Space root as `space_app.py`.

**Not `app.py`.** The service's own package is `app/` and sits at that same
root; a module of the same name beside it is a collision Python resolves in
favour of the package, so an entry point called `app.py` cannot be imported by
name. The first staging was called that, and `import app` returned the package. It exists because **ZeroGPU is Gradio-SDK only** —
the Docker Space this replaced ran the identical Azure image and could not be
given a GPU at any price, which is the whole reason CountGD has never been
measured end to end.

## What is served, and at which path

    /            the Gradio page — upload a photograph, pick a capability
    /health      the service's own, unauthenticated
    /capabilities the published contract
    /jobs        bearer token, exactly as on Azure

`app.main:app` is mounted whole rather than reimplemented, so the endpoints a
caller already scripts against keep working and the quality gate, the counting
guard and the observation vocabulary are the same code Azure runs. Gradio is
mounted **into** FastAPI rather than the other way round, because the API is the
product and the page is a demonstration of it.

## `@spaces.GPU`, and the honest state of it

ZeroGPU allocates a GPU for the duration of a decorated call and releases it
after. `_run_capability` is decorated, so any capability that reaches for CUDA
gets one.

**Nothing reaches for CUDA today, and that is worth saying plainly rather than
implying a speed-up nobody will see.** The two runnable artefacts here are
YOLOX-m and DINOv3, both ONNX, both executed by `onnxruntime` on CPU. What this
file buys is the *ability* to be given a GPU, which is the prerequisite for the
one capability that needs one: CountGD gets MAE **14.84** on broiler houses
against the deployed detector's **156.80**, and it is a PyTorch model that has
never been runnable anywhere in this project. The decorator is here so that
landing CountGD is a model change and not another SDK migration.

The decorator is documented as effect-free off ZeroGPU, so the same file runs
locally and on a CPU Space.

## What this Space still is not

**Not the production media path.** Azure reads captures from `animapmedia`
through the Container App's managed identity. A Space has no managed identity,
so this serves `ANIMAP_MEDIA_PROVIDER=local` against two public-domain frames
and no farm data reaches it. That was true of the Docker Space and it is true
here; changing SDK changes nothing about it.

**Not a licence-gate weakening.** The Docker build failed if an AGPL runtime
arrived. A Gradio Space has no Dockerfile to fail, so the gate that matters is
the runtime one that was always there: `providers.discover()` refuses to serve a
capability whose artefact fingerprints as a copyleft runtime, and `/health`
publishes `artefact_licenses` so a deployment in breach is visible from outside.
`scripts/install_models.py --check` is run below at start-up for the same
reason — verify, never fetch.
"""

from __future__ import annotations

import os
import subprocess
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent

# **The demo captures, and where they now live.**
#
# The Docker Space copied them to `/media` in a `Dockerfile.append` stanza. A
# Gradio Space has no Dockerfile, so `publish.py` stages them into
# `space/fixtures/` in the repo and this points the media store at that. Without
# it every job fails with "No media file for … under /home/user/app/media",
# which is the service correctly refusing to invent a photograph.
os.environ.setdefault("ANIMAP_MEDIA_PROVIDER", "local")
os.environ["ANIMAP_MEDIA_ROOT"] = str(HERE / "space" / "fixtures")

# **Before the service is imported.** `providers.discover()` reads the artefacts
# at import time, and a checksum that disagrees with its card should stop the
# Space rather than be discovered by a farm's job. On Docker this was a build
# step; a Gradio Space has no build step, so it is the first thing that runs.
_check = subprocess.run(
    [sys.executable, "scripts/install_models.py", "--check"],
    cwd=HERE, capture_output=True, text=True,
)
print(_check.stdout or "", flush=True)
if _check.returncode != 0:
    print(_check.stderr, file=sys.stderr, flush=True)
    raise SystemExit(
        "An artefact does not match its model card. Nothing is served: a "
        "capability answering from an artefact nobody verified is the one "
        "thing this service must not do."
    )

import gradio as gr  # noqa: E402
import uvicorn  # noqa: E402

from app.capabilities import REGISTRY  # noqa: E402
from app.main import RUNNERS, app as service, media, provider  # noqa: E402

try:
    import spaces
except ImportError:  # pragma: no cover - only present on a ZeroGPU Space
    class _Spaces:
        """A no-op stand-in, so this file runs unchanged off ZeroGPU.

        Handles both spellings — bare `@spaces.GPU` and called
        `@spaces.GPU(duration=60)` — because the real decorator does and a
        stand-in that only handled one would break the local run it exists for.
        """

        @staticmethod
        def GPU(*args, **kwargs):
            if args and callable(args[0]):
                return args[0]

            def decorate(fn):
                return fn

            return decorate

    spaces = _Spaces()


#: How long one call may hold a GPU.
#:
#: Sixty is ZeroGPU's default and comfortably over the measured worst case: a
#: dense frame runs three detection grids in a few seconds. It is deliberately
#: not raised "to be safe" — a shorter declared duration improves queue priority
#: for everybody, and a capability that genuinely needs longer should say so
#: when it lands.
GPU_SECONDS = 60


@spaces.GPU(duration=GPU_SECONDS)
def _run_capability(capability_key: str, image):
    """One capability against one uploaded image, inside a GPU allocation.

    The decorated boundary is here rather than deeper because ZeroGPU allocates
    per call: wrapping the whole job means one allocation for a whole answer,
    where wrapping an inner tensor op would mean many.
    """
    import datetime
    import uuid

    from app.schemas import InferenceRequest

    capability = REGISTRY[capability_key]
    runner = RUNNERS[capability_key]

    class _Store:
        def open_image(self, ref):
            return image

    request = InferenceRequest(
        capability_key=capability_key,
        farm_id=uuid.UUID(int=2),
        subject_type="animal" if capability.species == "cattle" else "flock_cycle",
        subject_id=uuid.UUID(int=1),
        media_ids=[uuid.uuid4()],
        captured_at=datetime.datetime.now(datetime.timezone.utc),
    )
    return runner.run(
        request=request, capability=capability,
        artefact=provider.artefact_for(capability), store=_Store(),
        request_id=uuid.uuid4(),
    )


def _describe(capability_key: str, image):
    """The page's handler. Returns what the service returned, and its caveats."""
    if image is None:
        return "Upload a photograph first.", {}

    capability = REGISTRY.get(capability_key)
    if capability is None or not provider.can_run(capability):
        return (
            f"**{capability_key}** cannot run here. A capability with no "
            f"verified artefact and no configured model answers `unavailable` "
            f"rather than a placeholder.",
            {},
        )

    try:
        result = _run_capability(capability_key, image)
    except Exception as exc:  # a refusal is an answer; a crash is not
        return f"The run failed: `{type(exc).__name__}: {exc}`", {}

    if result.recommended_recapture and not result.observations:
        headline = "**Nothing is claimed for this photograph.** Take it again."
    else:
        rows = [
            f"- `{o.type}` = **{o.value}**" + (f" ({o.unit})" if o.unit else "")
            for o in result.observations
        ]
        headline = "\n".join(rows) or "_No observation was produced._"

    caveats = "\n".join(f"> {w}" for w in result.warnings)
    return f"{headline}\n\n{caveats}", result.model_dump(mode="json")


_RUNNABLE = sorted(k for k in RUNNERS if provider.can_run(REGISTRY[k]))

with gr.Blocks(title="Animap inference") as page:
    gr.Markdown(
        "# Animap inference\n"
        "Livestock models that refuse to invent a result. Its most important "
        "property is what it **refuses**: a capability with no verified "
        "artefact behind it says so rather than returning a plausible number.\n\n"
        "**Read the caveats under the answer before you read the answer.** "
        "Every capability here is `experimental`, and the hosted ones have "
        "never been measured on a Nigerian herd.\n\n"
        f"Media provider: `{media.provider}` — two public-domain frames. "
        "No farm data reaches this Space."
    )
    with gr.Row():
        with gr.Column():
            choice = gr.Dropdown(
                choices=_RUNNABLE or ["nothing is runnable here"],
                value=(_RUNNABLE[0] if _RUNNABLE else None),
                label="Capability",
            )
            photo = gr.Image(type="pil", label="Photograph")
            go = gr.Button("Read it", variant="primary")
        with gr.Column():
            answer = gr.Markdown(label="What it said")
            raw = gr.JSON(label="The result, whole")

    go.click(_describe, inputs=[choice, photo], outputs=[answer, raw])

# **Launched by Gradio, with the service mounted onto Gradio's own app.**
#
# The order matters and it is not the obvious one. Mounting Gradio *into* a
# FastAPI app and serving that with `uvicorn.run` works on `cpu-basic` and dies
# on ZeroGPU: the app starts, logs "Uvicorn running on http://0.0.0.0:7860",
# and is immediately shut down. ZeroGPU supervises the Gradio server it expects
# `launch()` to create, and a uvicorn started by hand is not that server.
#
# So Gradio launches, and the whole Animap service is mounted onto the app it
# creates. `prevent_thread_lock` returns control so the mount can happen, and
# `block_thread` then holds the process open — which a Space requires, because
# a script that exits is a runtime error with nothing in the log.
page.launch(
    server_name="0.0.0.0",
    server_port=int(os.environ.get("GRADIO_SERVER_PORT", 7860)),
    prevent_thread_lock=True,
    show_api=False,
)

# `/animap/health`, `/animap/capabilities`, `/animap/jobs`.
#
# Prefixed because Gradio owns `/` once it has launched.
#
# **And moved to the front of the router, which is the part that actually
# mattered.** Starlette matches routes in order, and `launch()` has already
# registered Gradio's catch-all `GET /{path:path}` for its own client-side
# routing. A mount appended after that is unreachable for GET while still
# working for POST — which is exactly what the first two attempts did:
# `POST /animap/jobs` answered 401, and `GET /animap/health` returned the
# page's HTML. Changing the prefix from `/api` to `/animap` did nothing,
# because the prefix was never the problem.
#
# A half-working mount is worse than a broken one: the half that works reads as
# proof that the wiring is right.
page.app.mount("/animap", service)
page.app.router.routes.insert(0, page.app.router.routes.pop())

page.block_thread()