"""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