From 95d643d6d8a0801bcd43ce35665f85bd229cde3d Mon Sep 17 00:00:00 2001 From: Adriano Date: Sun, 11 Oct 2026 17:37:08 +0000 Subject: [PATCH] Worker di visione per stazione: modello, API, migrazione 019, execute_task (#45) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01NABedeMX43pyXZbDNUgA5B --- .../versions/019_station_vision_worker.py | 34 +++++++++++++++++++ src/backend/models/api/station.py | 23 ++++++++++++- src/backend/models/orm/station.py | 2 ++ src/backend/models/orm/vision.py | 2 ++ src/backend/services/station_service.py | 1 + src/backend/services/vision_service.py | 26 +++++++++++--- 6 files changed, 82 insertions(+), 6 deletions(-) create mode 100644 src/backend/migrations/versions/019_station_vision_worker.py diff --git a/src/backend/migrations/versions/019_station_vision_worker.py b/src/backend/migrations/versions/019_station_vision_worker.py new file mode 100644 index 0000000..cb73d12 --- /dev/null +++ b/src/backend/migrations/versions/019_station_vision_worker.py @@ -0,0 +1,34 @@ +"""stations.vision_worker_url e vision_results.worker_url (#45) + +stations.vision_worker_url: worker di visione della stazione (NULL = centrale). +vision_results.worker_url: worker effettivamente chiamato per quel risultato. + +Revision ID: 019_station_vision_worker +Revises: 018_task_loop_results +Create Date: 2026-10-11 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = '019_station_vision_worker' +down_revision: Union[str, None] = '018_task_loop_results' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # batch_alter_table come nelle migrazioni vicine: serve a SQLite + with op.batch_alter_table('stations') as batch: + batch.add_column(sa.Column('vision_worker_url', sa.String(500), nullable=True)) + with op.batch_alter_table('vision_results') as batch: + batch.add_column(sa.Column('worker_url', sa.String(500), nullable=True)) + + +def downgrade() -> None: + with op.batch_alter_table('vision_results') as batch: + batch.drop_column('worker_url') + with op.batch_alter_table('stations') as batch: + batch.drop_column('vision_worker_url') diff --git a/src/backend/models/api/station.py b/src/backend/models/api/station.py index 9e15bf4..6fb10bc 100644 --- a/src/backend/models/api/station.py +++ b/src/backend/models/api/station.py @@ -2,7 +2,19 @@ from src.backend.models.api.tipi import UtcDateTime from typing import Optional -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +def _normalize_worker_url(value: Optional[str]) -> Optional[str]: + """Vuoto/spazi -> None (worker centrale); altrimenti http(s) senza '/' finale (#45).""" + if value is None: + return None + value = value.strip() + if not value: + return None + if not value.lower().startswith(("http://", "https://")): + raise ValueError("vision_worker_url must start with http:// or https://") + return value.rstrip("/") class StationCreate(BaseModel): @@ -11,6 +23,10 @@ class StationCreate(BaseModel): location: Optional[str] = Field(default=None, max_length=255) notes: Optional[str] = None active: bool = True + # worker di visione della stazione (#45); vuoto = centrale + vision_worker_url: Optional[str] = Field(default=None, max_length=500) + + _check_worker_url = field_validator("vision_worker_url")(_normalize_worker_url) class StationUpdate(BaseModel): @@ -18,6 +34,10 @@ class StationUpdate(BaseModel): location: Optional[str] = Field(default=None, max_length=255) notes: Optional[str] = None active: Optional[bool] = None + # worker di visione della stazione (#45); "" -> None riporta al centrale + vision_worker_url: Optional[str] = Field(default=None, max_length=500) + + _check_worker_url = field_validator("vision_worker_url")(_normalize_worker_url) class StationResponse(BaseModel): @@ -28,6 +48,7 @@ class StationResponse(BaseModel): location: Optional[str] notes: Optional[str] active: bool + vision_worker_url: Optional[str] = None # #45 created_by: int created_at: UtcDateTime diff --git a/src/backend/models/orm/station.py b/src/backend/models/orm/station.py index ca72e0c..973bcaa 100644 --- a/src/backend/models/orm/station.py +++ b/src/backend/models/orm/station.py @@ -25,6 +25,8 @@ class Station(Base): location: Mapped[Optional[str]] = mapped_column(String(255), nullable=True) notes: Mapped[Optional[str]] = mapped_column(Text, nullable=True) active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, index=True) + # URL del worker di visione della stazione (#45); NULL = worker centrale + vision_worker_url: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) created_by: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False) created_at: Mapped[datetime] = mapped_column( DateTime, nullable=False, server_default=func.now() diff --git a/src/backend/models/orm/vision.py b/src/backend/models/orm/vision.py index f51bae1..020e039 100644 --- a/src/backend/models/orm/vision.py +++ b/src/backend/models/orm/vision.py @@ -26,6 +26,8 @@ class VisionResult(Base): Integer, ForeignKey("stations.id"), nullable=True ) device_code: Mapped[Optional[str]] = mapped_column(String(100), nullable=True) + # URL del worker effettivamente chiamato (#45), copiato: la stazione puo' cambiarlo dopo + worker_url: Mapped[Optional[str]] = mapped_column(String(500), nullable=True) # Copied, not referenced: a calibration redone six months from now must not # rewrite what a measurement already taken means. Same for the graph. calibration_snapshot: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) diff --git a/src/backend/services/station_service.py b/src/backend/services/station_service.py index 18f5f19..eb35dfa 100644 --- a/src/backend/services/station_service.py +++ b/src/backend/services/station_service.py @@ -31,6 +31,7 @@ async def create_station( location=data.location, notes=data.notes, active=data.active, + vision_worker_url=data.vision_worker_url, # #45 created_by=creator.id, ) db.add(station) diff --git a/src/backend/services/vision_service.py b/src/backend/services/vision_service.py index 400472f..7af7b2f 100644 --- a/src/backend/services/vision_service.py +++ b/src/backend/services/vision_service.py @@ -19,6 +19,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from src.backend.config import settings from src.backend.models.orm.measurement import Measurement from src.backend.models.orm.recipe import RecipeVersion +from src.backend.models.orm.station import Station from src.backend.models.orm.task import RecipeSubtask, RecipeTask from src.backend.models.orm.vision import VisionReferenceImage, VisionResult from src.backend.services import measurement_service @@ -53,6 +54,7 @@ async def _call_worker( *, recipe_dir: str | None = None, calibration: dict | None = None, + worker_url: str | None = None, ) -> dict: """The one place that talks to the worker; the tests replace it. @@ -70,6 +72,8 @@ async def _call_worker( # recipe_dir (relativo, #46) e calibration sono opzionali: se assenti non # si mandano, il worker usa i suoi default. data = {"graph": json.dumps(graph)} + # worker della stazione (#45) o, se manca, il centrale; mai un ripiego dall'uno all'altro + base = (worker_url or settings.vision_worker_url).rstrip("/") if recipe_dir: data["recipe_dir"] = recipe_dir if calibration: @@ -77,19 +81,19 @@ async def _call_worker( try: async with httpx.AsyncClient(timeout=120.0) as client: response = await client.post( - f"{settings.vision_worker_url}/run", + f"{base}/run", files={"image": ("frame.png", image_bytes, "image/png")}, data=data, ) except httpx.TimeoutException as exc: raise HTTPException( status_code=status.HTTP_504_GATEWAY_TIMEOUT, - detail=f"the vision worker did not answer in time: {exc}", + detail=f"the vision worker did not answer in time: {exc} ({base})", # #45 ) from exc except (httpx.TransportError, httpx.HTTPError) as exc: raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, - detail=f"the vision worker is unreachable: {exc}", + detail=f"the vision worker is unreachable: {exc} ({base})", # #45 ) from exc # The worker already gave this fault a name - forward its status and @@ -352,12 +356,20 @@ async def execute_task( detail="this recipe is not active", ) + run = None if production_run_id is not None: # 404 if the run does not exist, 422 if it belongs to another recipe, # 409 if it is closed or paused - same rule a manual reading follows. - await measurement_service.validate_run_for_version( + run = await measurement_service.validate_run_for_version( db, production_run_id, version_id, ) + # Worker della stazione della sessione di produzione (#45); senza run o senza URL: centrale + station_url = None + if run is not None: + station = (await db.execute( + select(Station).where(Station.id == run.station_id) + )).scalar_one_or_none() + station_url = (station.vision_worker_url or None) if station is not None else None if not task.vision_json: raise HTTPException( @@ -411,6 +423,7 @@ async def execute_task( # passa `calibration`, la ricetta porta la sua (#46). report = await _call_worker( image_bytes, task.vision_json, recipe_dir=task.vision_recipe_dir, + worker_url=station_url, # #45 ) outputs = report["outputs"] @@ -439,7 +452,10 @@ async def execute_task( image_path=image_path, overlay_path=overlay_path, engine_version=report["engine_version"], - executed_on="server", + # #45: dove e' girato davvero, e quale worker (copiato come lo snapshot) + executed_on="station" if station_url else "server", + station_id=run.station_id if run is not None else None, + worker_url=(station_url or settings.vision_worker_url).rstrip("/"), graph_snapshot=task.vision_json, duration_ms=report.get("duration_ms"), )