app/training/schema.py

"""What a labelled sample has to contain.

One JSON object per line, one file per capability. The models here are strict on
purpose: a dataset is easy to collect badly and expensive to re-collect, and
almost every failure is a field somebody left out in the field and nobody
noticed until training.

The distinction that matters most is `is_ground_truth`. A record can be
perfectly well formed and still not be evidence — a girth-tape reading is an
estimate produced by the same morphometric relationship the model is trying to
learn, so training a weight model on tape readings teaches it to reproduce a
formula rather than to predict a weight. Those records are allowed in, and
excluded from the evaluation set.
"""

from __future__ import annotations

from datetime import date
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

View = Literal["side", "rear", "front", "top", "oblique"]

#: Scale readings that are measurements. Everything else is an estimate.
TRUE_SCALES = frozenset({"weighbridge", "platform_scale"})

#: A weight and a photograph taken further apart than this are not the same
#: animal state. Cattle gain and lose several kilograms of gut fill in a day.
MAX_WEIGHING_GAP_HOURS = 24.0

#: Body condition is scored in half points. A dataset with 3.7 in it was scored
#: by somebody using a different scale.
BCS_STEPS = tuple(round(1.0 + 0.5 * i, 1) for i in range(9))


class FrameRef(BaseModel):
    """One photograph, and enough geometry to know what it is a photograph of."""

    model_config = ConfigDict(frozen=True)

    path: str
    view: View
    #: Without these, two frames of the same animal at different distances are
    #: indistinguishable to a model that has to infer size.
    camera_height_cm: float | None = Field(default=None, gt=0)
    subject_distance_m: float | None = Field(default=None, gt=0)


class ScaleReference(BaseModel):
    """The object in frame that makes a pixel mean a centimetre.

    A photograph carries no scale. Without a reference of known length in the
    same plane as the animal, heart girth in pixels is not convertible to heart
    girth in centimetres, and every weight the model produces is a guess about
    how far away the camera was.
    """

    model_config = ConfigDict(frozen=True)

    kind: Literal[
        "marker_board", "chest_band", "calibration_rod", "known_gate_width",
        "depth_sensor",
    ]
    length_cm: float = Field(gt=0)
    #: The reference has to be in the frames the measurement is taken from, not
    #: merely somewhere in the capture set.
    visible_in_views: list[View] = Field(min_length=1)


class WeightSample(BaseModel):
    """One animal, photographed and weighed.

    Weight estimation from a photograph is regression on morphometrics — heart
    girth and body length, recovered from keypoints or a segmentation mask, then
    converted to a scale. Every field here exists because one of those three
    steps needs it.
    """

    model_config = ConfigDict(frozen=True)

    sample_id: str
    farm_id: str
    animal_id: str
    captured_at: date

    frames: list[FrameRef] = Field(min_length=2)
    scale_reference: ScaleReference

    weight_kg: float = Field(gt=20, lt=1200)
    scale_type: Literal[
        "weighbridge", "platform_scale", "girth_tape", "visual_estimate",
    ]
    hours_between_capture_and_weighing: float = Field(ge=0)

    breed: str
    sex: Literal["male", "female"]
    age_months: int | None = Field(default=None, ge=0, le=360)

    #: Optional, and worth collecting. A tape measurement of the same animal
    #: lets the pipeline be evaluated in two halves — did the keypoints recover
    #: the girth, and did the regression convert it — instead of only end to end.
    heart_girth_cm: float | None = Field(default=None, gt=0)
    body_length_cm: float | None = Field(default=None, gt=0)

    @model_validator(mode="after")
    def _needs_a_side_and_a_rear_view(self) -> WeightSample:
        views = {f.view for f in self.frames}
        missing = {"side", "rear"} - views
        if missing:
            raise ValueError(
                f"{self.sample_id}: missing {sorted(missing)} view(s). Side gives "
                f"body length, rear gives width; neither alone gives volume."
            )
        return self

    @model_validator(mode="after")
    def _scale_reference_must_be_in_a_measured_view(self) -> WeightSample:
        if not set(self.scale_reference.visible_in_views) & {"side", "rear"}:
            raise ValueError(
                f"{self.sample_id}: the scale reference is not visible in the side "
                f"or rear view, so it cannot scale the measurement."
            )
        return self

    @property
    def is_ground_truth(self) -> bool:
        return (
            self.scale_type in TRUE_SCALES
            and self.hours_between_capture_and_weighing <= MAX_WEIGHING_GAP_HOURS
        )

    @property
    def weight_band(self) -> str:
        """100 kg bands. Coverage per band is what stops a model that only
        works on the middle of the range from passing on an overall average."""
        if self.weight_kg < 100:
            return "<100"
        if self.weight_kg >= 500:
            return ">=500"
        lower = int(self.weight_kg // 100) * 100
        return f"{lower}-{lower + 100}"


class BcsScore(BaseModel):
    """One person's score, and what qualifies them to give it."""

    model_config = ConfigDict(frozen=True)

    scorer_id: str
    score: float
    credential: Literal["veterinarian", "trained_technician", "farmer"]

    @field_validator("score")
    @classmethod
    def _half_points_only(cls, value: float) -> float:
        if round(value, 1) not in BCS_STEPS:
            raise ValueError(
                f"{value} is not a body condition score. The scale is 1 to 5 in "
                f"half points."
            )
        return round(value, 1)


class BcsSample(BaseModel):
    """One animal, scored by more than one person.

    Body condition is ordinal regression, and its ground truth is a human
    judgement with real disagreement in it. A single scorer's opinion is a
    label with unknown error, so the schema requires at least two and the
    validator measures how far apart they were.
    """

    model_config = ConfigDict(frozen=True)

    sample_id: str
    farm_id: str
    animal_id: str
    captured_at: date

    frames: list[FrameRef] = Field(min_length=1)
    scores: list[BcsScore] = Field(min_length=2)

    breed: str
    sex: Literal["male", "female"]
    age_months: int | None = Field(default=None, ge=0, le=360)

    @model_validator(mode="after")
    def _needs_a_rear_view(self) -> BcsSample:
        if "rear" not in {f.view for f in self.frames}:
            raise ValueError(
                f"{self.sample_id}: no rear view. Tailhead and pin bones are where "
                f"body condition is read; a side view alone confuses condition "
                f"with frame size."
            )
        return self

    @model_validator(mode="after")
    def _scorers_must_be_distinct(self) -> BcsSample:
        ids = [s.scorer_id for s in self.scores]
        if len(set(ids)) != len(ids):
            raise ValueError(f"{self.sample_id}: the same scorer appears twice.")
        return self

    @property
    def consensus(self) -> float:
        """The median score, breaking an even split downwards.

        Two scorers who say 3.0 and 3.5 average to 3.25, which is not a body
        condition score. Rounding has to go one way, and down is the safe
        direction: calling an animal leaner than it is prompts somebody to look
        at it, and calling it fatter than it is hides a thin one.
        """
        ordered = sorted(s.score for s in self.scores)
        middle = len(ordered) // 2
        if len(ordered) % 2:
            return ordered[middle]
        return ordered[middle - 1]

    @property
    def spread(self) -> float:
        scores = [s.score for s in self.scores]
        return max(scores) - min(scores)

    @property
    def is_ground_truth(self) -> bool:
        """Two qualified scorers who agree within one point.

        A wider spread is not a bad sample — it is a sample nobody can grade,
        and using it as evaluation truth measures the scorers, not the model.
        """
        qualified = [s for s in self.scores if s.credential != "farmer"]
        return len(qualified) >= 2 and self.spread <= 1.0