Worker di visione per stazione: modello, API, migrazione 019, execute_task (#45)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01NABedeMX43pyXZbDNUgA5B

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')

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
 

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()

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)

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)

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