135 lines
4.9 KiB
Python
135 lines
4.9 KiB
Python
"""Pydantic schemas for Measurement operations."""
|
|
from src.backend.models.api.tipi import UtcDateTime, UtcNaiveInput
|
|
from typing import Optional
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
|
|
class MeasurementCreate(BaseModel):
|
|
"""Schema for creating a single measurement."""
|
|
subtask_id: int
|
|
version_id: int
|
|
# Finiteness and the DECIMAL(12,6) scale bound are enforced in
|
|
# measurement_service.save_measurement, not with a Field constraint here:
|
|
# a NaN/Infinity that fails a pydantic Field constraint gets echoed back
|
|
# in the validation error's "input", and FastAPI's default error response
|
|
# then crashes with a 500 trying to JSON-encode that same NaN/Infinity
|
|
# (json.dumps(allow_nan=False) in Starlette's JSONResponse). A plain
|
|
# HTTPException raised after parsing sidesteps that.
|
|
value: float
|
|
lot_number: Optional[str] = Field(None, max_length=100)
|
|
serial_number: Optional[str] = Field(None, max_length=100)
|
|
# "camera" is deliberately absent from this pattern, even though the ORM
|
|
# enum and the database allow it. Do not add it here: a camera
|
|
# measurement must be produced by vision_service.execute_task, never by a
|
|
# client posting JSON straight to this endpoint - there is no image, no
|
|
# graph and no engine_version behind a value that arrived this way.
|
|
input_method: str = Field("manual", pattern="^(usb_caliper|manual)$")
|
|
input_duration_ms: Optional[int] = Field(None, ge=0)
|
|
# The production this belongs to, when one is open at the station.
|
|
production_run_id: Optional[int] = Field(None, gt=0)
|
|
|
|
|
|
class MeasurementBatchCreate(BaseModel):
|
|
"""Schema for creating multiple measurements at once."""
|
|
measurements: list[MeasurementCreate]
|
|
|
|
|
|
class MeasurementResponse(BaseModel):
|
|
"""Schema for measurement response."""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
subtask_id: int
|
|
version_id: int
|
|
measured_by: int
|
|
value: Optional[float] = None # #118: null = quota non prodotta dalla visione (NaN)
|
|
pass_fail: str
|
|
deviation: Optional[float] = None
|
|
lot_number: Optional[str] = None
|
|
serial_number: Optional[str] = None
|
|
input_method: str
|
|
input_duration_ms: Optional[int] = None
|
|
production_run_id: Optional[int] = None
|
|
# Who let this value stand, when it was out of tolerance. Null on a value in
|
|
# tolerance - and on one that is not and is still waiting for an answer.
|
|
supervisor_id: Optional[int] = None
|
|
authorised_at: Optional[UtcDateTime] = None
|
|
measured_at: UtcDateTime
|
|
synced_to_csv: bool
|
|
|
|
|
|
class MeasurementListResponse(BaseModel):
|
|
"""Schema for paginated measurement list."""
|
|
items: list[MeasurementResponse]
|
|
total: int
|
|
page: int
|
|
per_page: int
|
|
pages: int
|
|
|
|
|
|
class MeasurementAuthorisation(BaseModel):
|
|
"""Credentials of the supervisor allowing an out-of-tolerance value to stand."""
|
|
|
|
# #59: o username+password, oppure supervisor_user_id+supervisor_pin
|
|
supervisor_username: Optional[str] = Field(None, min_length=1)
|
|
supervisor_password: Optional[str] = Field(None, min_length=1)
|
|
supervisor_user_id: Optional[int] = Field(None, gt=0)
|
|
supervisor_pin: Optional[str] = Field(None, pattern=r"^[0-9]{4}$")
|
|
|
|
@model_validator(mode="after")
|
|
def _one_credential_pair(self):
|
|
by_password = self.supervisor_username and self.supervisor_password
|
|
by_pin = self.supervisor_user_id and self.supervisor_pin
|
|
if not (by_password or by_pin):
|
|
raise ValueError(
|
|
"supervisor credentials required: username+password or user_id+pin"
|
|
)
|
|
return self
|
|
|
|
|
|
class TaskProgressResponse(BaseModel):
|
|
"""How far one measurement task has got: quotes taken out of quotes expected."""
|
|
|
|
task_id: int
|
|
quotes: int
|
|
measured: int
|
|
# empty (declared as measurement, no quotes yet) | none | partial | complete
|
|
state: str
|
|
|
|
|
|
class TaskProgressListResponse(BaseModel):
|
|
"""Progress of every measurement task of a version."""
|
|
|
|
tasks: list[TaskProgressResponse]
|
|
|
|
|
|
class CycleMeasurementResponse(BaseModel):
|
|
"""One quote already measured in the cycle under way."""
|
|
|
|
subtask_id: int
|
|
value: Optional[float] = None # #118: null = NaN
|
|
pass_fail: str
|
|
measured_at: UtcDateTime
|
|
|
|
|
|
class CurrentCycleResponse(BaseModel):
|
|
"""The quotes to restore on the screen after a reload; empty if the task is closed."""
|
|
|
|
measurements: list[CycleMeasurementResponse]
|
|
|
|
|
|
class MeasurementQuery(BaseModel):
|
|
"""Schema for measurement query filters."""
|
|
recipe_id: Optional[int] = None
|
|
version_id: Optional[int] = None
|
|
subtask_id: Optional[int] = None
|
|
measured_by: Optional[int] = None
|
|
lot_number: Optional[str] = None
|
|
serial_number: Optional[str] = None
|
|
date_from: UtcNaiveInput = None
|
|
date_to: UtcNaiveInput = None
|
|
pass_fail: Optional[str] = Field(None, pattern="^(pass|warning|fail)$")
|
|
page: int = Field(1, ge=1)
|
|
per_page: int = Field(50, ge=1, le=500)
|