247 lines
9.0 KiB
Python
247 lines
9.0 KiB
Python
"""Pydantic schemas for RecipeTask and RecipeSubtask operations."""
|
|
import re
|
|
from typing import Any, Literal, Optional
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
|
|
|
from src.backend.models.api.station import _normalize_http_url
|
|
from src.backend.models.orm.task import DEFAULT_LOOP_TASK_TYPES, TASK_TYPES
|
|
|
|
# Kept in step with the ORM enum so the two cannot drift apart.
|
|
TaskType = Literal[TASK_TYPES]
|
|
|
|
|
|
def validate_tolerance_order(
|
|
ltl: Optional[float],
|
|
lwl: Optional[float],
|
|
nominal: Optional[float],
|
|
uwl: Optional[float],
|
|
utl: Optional[float],
|
|
) -> None:
|
|
"""ltl <= lwl <= nominal <= uwl <= utl, checked only across the values set.
|
|
|
|
A one-sided tolerance - only utl, say - is not an error: what is wrong is
|
|
limits that invert the sequence they do share, like utl landing below ltl.
|
|
Shared by SubtaskCreate's own validator and by the router's partial-update
|
|
check (SubtaskUpdate has no fixed set of fields to compare).
|
|
"""
|
|
ordered = [
|
|
("ltl", ltl), ("lwl", lwl), ("nominal", nominal), ("uwl", uwl), ("utl", utl),
|
|
]
|
|
present = [(name, value) for name, value in ordered if value is not None]
|
|
for (prev_name, prev), (name, value) in zip(present, present[1:]):
|
|
if prev > value:
|
|
raise ValueError(
|
|
f"Tolerance limits out of order: {prev_name}={prev} > {name}={value}"
|
|
)
|
|
|
|
|
|
# #55: campi extra di un task 'note'
|
|
MAX_NOTE_FIELDS = 20
|
|
_FIELD_KEY_RE = re.compile(r"^[a-z0-9_]{1,40}$")
|
|
NOTE_FIELD_TYPES = ("text", "number")
|
|
|
|
|
|
def normalize_note_fields(fields: Optional[list]) -> Optional[list[dict]]:
|
|
"""Valida e normalizza `fields` di un task 'note' (#55); lista vuota -> None.
|
|
|
|
Solleva ValueError con un messaggio leggibile (il router lo trasforma in 422).
|
|
"""
|
|
if not fields:
|
|
return None
|
|
if len(fields) > MAX_NOTE_FIELDS:
|
|
raise ValueError(f"Al massimo {MAX_NOTE_FIELDS} campi per task")
|
|
out: list[dict] = []
|
|
seen: set[str] = set()
|
|
for i, raw in enumerate(fields, start=1):
|
|
if not isinstance(raw, dict):
|
|
raise ValueError(f"Campo {i}: formato non valido")
|
|
key = raw.get("key")
|
|
if not isinstance(key, str) or not _FIELD_KEY_RE.match(key):
|
|
raise ValueError(
|
|
f"Campo {i}: chiave non valida (solo a-z, 0-9, _, da 1 a 40 caratteri)"
|
|
)
|
|
if key in seen:
|
|
raise ValueError(f"Campo {i}: chiave '{key}' duplicata")
|
|
seen.add(key)
|
|
label = raw.get("label")
|
|
label = label.strip() if isinstance(label, str) else ""
|
|
if not 1 <= len(label) <= 100:
|
|
raise ValueError(f"Campo '{key}': etichetta da 1 a 100 caratteri")
|
|
ftype = raw.get("type")
|
|
if ftype not in NOTE_FIELD_TYPES:
|
|
raise ValueError(f"Campo '{key}': tipo deve essere text o number")
|
|
default = raw.get("default")
|
|
if default is not None:
|
|
if not isinstance(default, str):
|
|
raise ValueError(f"Campo '{key}': il valore predefinito deve essere testo")
|
|
if default == "":
|
|
default = None
|
|
if default is not None and ftype == "number":
|
|
try:
|
|
float(default.replace(",", "."))
|
|
except ValueError:
|
|
raise ValueError(f"Campo '{key}': il valore predefinito non e' un numero")
|
|
attachment = raw.get("attachment", False)
|
|
if not isinstance(attachment, bool):
|
|
raise ValueError(f"Campo '{key}': attachment deve essere true o false")
|
|
out.append({
|
|
"key": key, "label": label, "type": ftype,
|
|
"default": default, "attachment": attachment,
|
|
})
|
|
return out
|
|
|
|
|
|
def check_task_loop_options(task_type: str, at_close: bool, fields: Optional[list]) -> None:
|
|
"""at_close e fields sono ammessi solo per i task 'note' (#53 #55)."""
|
|
if task_type != "note":
|
|
if at_close:
|
|
raise ValueError("'Alla chiusura' e' ammesso solo per i task di tipo note")
|
|
if fields:
|
|
raise ValueError("I campi extra sono ammessi solo per i task di tipo note")
|
|
|
|
|
|
def default_in_loop(task_type: str) -> bool:
|
|
"""in_loop quando non dichiarato: misure e check si', gli altri no (#53)."""
|
|
return task_type in DEFAULT_LOOP_TASK_TYPES
|
|
|
|
|
|
class SubtaskCreate(BaseModel):
|
|
"""Schema for creating a subtask."""
|
|
marker_number: int = Field(..., ge=1)
|
|
description: str = Field(..., min_length=1, max_length=500)
|
|
measurement_type: Optional[str] = Field(None, max_length=100)
|
|
nominal: Optional[float] = None
|
|
utl: Optional[float] = None
|
|
uwl: Optional[float] = None
|
|
lwl: Optional[float] = None
|
|
ltl: Optional[float] = None
|
|
unit: str = Field("mm", max_length=20)
|
|
image_path: Optional[str] = Field(None, max_length=500)
|
|
vision_output: Optional[str] = None
|
|
|
|
@model_validator(mode="after")
|
|
def _check_tolerance_order(self) -> "SubtaskCreate":
|
|
validate_tolerance_order(self.ltl, self.lwl, self.nominal, self.uwl, self.utl)
|
|
return self
|
|
|
|
|
|
class SubtaskUpdate(BaseModel):
|
|
"""Schema for updating a subtask."""
|
|
marker_number: Optional[int] = Field(None, ge=1)
|
|
description: Optional[str] = Field(None, min_length=1, max_length=500)
|
|
measurement_type: Optional[str] = Field(None, max_length=100)
|
|
nominal: Optional[float] = None
|
|
utl: Optional[float] = None
|
|
uwl: Optional[float] = None
|
|
lwl: Optional[float] = None
|
|
ltl: Optional[float] = None
|
|
unit: Optional[str] = Field(None, max_length=20)
|
|
image_path: Optional[str] = Field(None, max_length=500)
|
|
vision_output: Optional[str] = None
|
|
|
|
|
|
class SubtaskResponse(BaseModel):
|
|
"""Schema for subtask response."""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
task_id: int
|
|
marker_number: int
|
|
description: str
|
|
measurement_type: Optional[str] = None
|
|
nominal: Optional[float] = None
|
|
utl: Optional[float] = None
|
|
uwl: Optional[float] = None
|
|
lwl: Optional[float] = None
|
|
ltl: Optional[float] = None
|
|
unit: str
|
|
image_path: Optional[str] = None
|
|
vision_output: Optional[str] = None
|
|
|
|
|
|
class AcquisitionConfig(BaseModel):
|
|
"""#97: camera (servizio + sorgente) e modo di analisi scelti nel task."""
|
|
camera_url: str = Field(..., max_length=500)
|
|
camera_source: str = Field(..., min_length=1, max_length=200)
|
|
modo: Literal["impulsiva", "in_linea"] = "impulsiva"
|
|
|
|
@field_validator("camera_url")
|
|
@classmethod
|
|
def _check_camera_url(cls, value: str) -> str:
|
|
# stessa normalizzazione delle stazioni; qui il vuoto non e' ammesso
|
|
normalized = _normalize_http_url(value, "camera_url")
|
|
if normalized is None:
|
|
raise ValueError("camera_url must not be empty")
|
|
return normalized
|
|
|
|
@field_validator("camera_source")
|
|
@classmethod
|
|
def _check_camera_source(cls, value: str) -> str:
|
|
value = value.strip()
|
|
if not value:
|
|
raise ValueError("camera_source must not be empty")
|
|
return value
|
|
|
|
|
|
class TaskCreate(BaseModel):
|
|
"""Schema for creating a task."""
|
|
title: str = Field(..., min_length=1, max_length=255)
|
|
task_type: TaskType = "note"
|
|
directive: Optional[str] = None
|
|
description: Optional[str] = None
|
|
file_path: Optional[str] = Field(None, max_length=500)
|
|
file_type: Optional[str] = Field(None, pattern="^(image|pdf)$")
|
|
annotations_json: Optional[dict[str, Any]] = None
|
|
vision_json: Optional[dict] = None
|
|
acquisition_json: Optional[AcquisitionConfig] = None # #97
|
|
# #53: assente = default dal tipo (misure e check nel ciclo)
|
|
in_loop: Optional[bool] = None
|
|
at_close: bool = False
|
|
fields: Optional[list[dict[str, Any]]] = None
|
|
subtasks: list[SubtaskCreate] = []
|
|
|
|
|
|
class TaskUpdate(BaseModel):
|
|
"""Schema for updating a task."""
|
|
title: Optional[str] = Field(None, min_length=1, max_length=255)
|
|
task_type: Optional[TaskType] = None
|
|
directive: Optional[str] = None
|
|
description: Optional[str] = None
|
|
file_path: Optional[str] = Field(None, max_length=500)
|
|
file_type: Optional[str] = Field(None, pattern="^(image|pdf)$")
|
|
annotations_json: Optional[dict[str, Any]] = None
|
|
vision_json: Optional[dict] = None
|
|
acquisition_json: Optional[AcquisitionConfig] = None # #97
|
|
in_loop: Optional[bool] = None
|
|
at_close: Optional[bool] = None
|
|
fields: Optional[list[dict[str, Any]]] = None
|
|
|
|
|
|
class TaskResponse(BaseModel):
|
|
"""Schema for task response."""
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
id: int
|
|
version_id: int
|
|
recipe_id: Optional[int] = None
|
|
order_index: int
|
|
task_type: str
|
|
title: str
|
|
directive: Optional[str] = None
|
|
description: Optional[str] = None
|
|
file_path: Optional[str] = None
|
|
file_type: Optional[str] = None
|
|
annotations_json: Optional[dict[str, Any]] = None
|
|
vision_json: Optional[dict] = None
|
|
acquisition_json: Optional[AcquisitionConfig] = None # #97
|
|
in_loop: bool = False
|
|
at_close: bool = False
|
|
fields: Optional[list[dict[str, Any]]] = None
|
|
subtasks: list[SubtaskResponse] = []
|
|
|
|
|
|
class TaskReorderRequest(BaseModel):
|
|
"""Schema for reordering tasks."""
|
|
task_ids: list[int] = Field(..., min_length=1)
|