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"), ) diff --git a/src/backend/tests/test_migrations.py b/src/backend/tests/test_migrations.py index 9ba63d4..08a91ff 100644 --- a/src/backend/tests/test_migrations.py +++ b/src/backend/tests/test_migrations.py @@ -233,3 +233,24 @@ class TestTaskLoopResultsMigration: with pytest.raises(RuntimeError, match="check"): command.downgrade(cfg, "017_user_pin") + + +class TestStationVisionWorkerMigration: + """019_station_vision_worker: stations.vision_worker_url e vision_results.worker_url, su e giu (#45).""" + + def test_019_upgrade_and_downgrade(self, sqlite_database_url): + db_path, _ = sqlite_database_url + cfg = _alembic_config() + + def cols(table): + return {c["name"] for c in _inspect(db_path).get_columns(table)} + + command.upgrade(cfg, "head") + assert "vision_worker_url" in cols("stations") + assert "worker_url" in cols("vision_results") + command.downgrade(cfg, "018_task_loop_results") + assert "vision_worker_url" not in cols("stations") + assert "worker_url" not in cols("vision_results") + command.upgrade(cfg, "head") # e di nuovo su + assert "vision_worker_url" in cols("stations") + assert "worker_url" in cols("vision_results") diff --git a/src/backend/tests/test_stations_api.py b/src/backend/tests/test_stations_api.py index 432794b..87551d2 100644 --- a/src/backend/tests/test_stations_api.py +++ b/src/backend/tests/test_stations_api.py @@ -374,3 +374,59 @@ async def test_reset_unknown_station_is_404(client: AsyncClient, admin_user): "/api/stations/999999/recipes", headers=auth_headers(admin_user), ) assert resp.status_code == 404 + + +# --- vision_worker_url (#45) --- + +async def test_station_vision_worker_url_roundtrip(client: AsyncClient, admin_user): + created = await client.post( + "/api/stations", + headers=auth_headers(admin_user), + json={"code": "ST-VW", "name": "VW", "vision_worker_url": " http://10.0.0.7:8100/ "}, + ) + assert created.status_code == 201, created.text + # spazi tolti e "/" finale tolto + assert created.json()["vision_worker_url"] == "http://10.0.0.7:8100" + + sid = created.json()["id"] + updated = await client.put( + f"/api/stations/{sid}", headers=auth_headers(admin_user), + json={"vision_worker_url": "https://w.example:9000"}, + ) + assert updated.status_code == 200 + assert updated.json()["vision_worker_url"] == "https://w.example:9000" + + # stringa vuota = torna al worker centrale + cleared = await client.put( + f"/api/stations/{sid}", headers=auth_headers(admin_user), + json={"vision_worker_url": ""}, + ) + assert cleared.status_code == 200 + assert cleared.json()["vision_worker_url"] is None + + +async def test_station_vision_worker_url_defaults_to_null(client: AsyncClient, admin_user): + created = await client.post( + "/api/stations", headers=auth_headers(admin_user), + json={"code": "ST-VN", "name": "VN"}, + ) + assert created.status_code == 201 + assert created.json()["vision_worker_url"] is None + + +async def test_station_vision_worker_url_must_be_http(client: AsyncClient, admin_user): + resp = await client.post( + "/api/stations", headers=auth_headers(admin_user), + json={"code": "ST-VF", "name": "VF", "vision_worker_url": "ftp://x"}, + ) + assert resp.status_code == 422 + + created = await client.post( + "/api/stations", headers=auth_headers(admin_user), + json={"code": "ST-VG", "name": "VG"}, + ) + resp = await client.put( + f"/api/stations/{created.json()['id']}", headers=auth_headers(admin_user), + json={"vision_worker_url": "ftp://x"}, + ) + assert resp.status_code == 422 diff --git a/src/backend/tests/test_vision_station_worker.py b/src/backend/tests/test_vision_station_worker.py new file mode 100644 index 0000000..a2fef01 --- /dev/null +++ b/src/backend/tests/test_vision_station_worker.py @@ -0,0 +1,199 @@ +"""#45: il worker di visione si sceglie per stazione. + +La stazione si legge dalla sessione di produzione (`production_run_id`): con un +URL sulla stazione `_call_worker` lo riceve e il VisionResult lo registra; +senza URL o senza run si resta sul worker centrale, senza ripiego fra i due. +""" +import io + +import httpx +import pytest +from fastapi import HTTPException +from PIL import Image + +from src.backend.config import settings +from src.backend.models.api.production import ProductionRunCreate +from src.backend.models.orm.station import Station +from src.backend.models.orm.vision import VisionResult +from src.backend.services import production_service +from src.backend.services.vision_service import _call_worker +from src.backend.tests.conftest import auth_headers, create_test_recipe +from src.backend.tests.test_vision_execute import recipe_parts +from sqlalchemy import select + +STATION_URL = "http://10.0.0.7:8100" + + +def an_image() -> bytes: + buffer = io.BytesIO() + Image.new("L", (64, 64), color=0).save(buffer, format="PNG") + return buffer.getvalue() + + +@pytest.fixture +def worker_calls(monkeypatch): + """Fake `_call_worker` che ricorda i kwargs con cui e' stato chiamato.""" + calls: list[dict] = [] + + async def fake_call(image_bytes: bytes, graph: dict, **kwargs) -> dict: + calls.append(kwargs) + return {"outputs": {"t1.diameter": 10.0}, "failures": [], + "engine_version": "e" * 40, "duration_ms": 3} + + monkeypatch.setattr("src.backend.services.vision_service._call_worker", fake_call) + return calls + + +async def _camera_task(db_session, user_id: int): + recipe = await create_test_recipe(db_session, user_id) + version, task, subtask = await recipe_parts(db_session, recipe.id) + task.task_type = "camera_measure" + task.vision_json = {"schema_version": 1, "name": "p", "tools": []} + subtask.vision_output = "t1.diameter" + await db_session.commit() + return recipe, task + + +async def _station_with_run(db_session, user, recipe, worker_url): + station = Station( + code="ST-W", name="Stazione W", active=True, created_by=user.id, + vision_worker_url=worker_url, + ) + db_session.add(station) + await db_session.commit() + await db_session.refresh(station) + run = await production_service.open_run( + db_session, + ProductionRunCreate(station_code=station.code, recipe_id=recipe.id), + user, + ) + await db_session.commit() + return station, run + + +async def _execute(client, user, task, run=None): + data = {"task_id": str(task.id)} + if run is not None: + data["production_run_id"] = str(run.id) + response = await client.post( + "/api/vision/execute", + headers=auth_headers(user), + files={"image": ("p.png", an_image(), "image/png")}, + data=data, + ) + assert response.status_code == 200, response.text + + +async def _the_result(db_session) -> VisionResult: + return (await db_session.execute(select(VisionResult))).scalars().one() + + +@pytest.mark.asyncio +async def test_station_with_url_gets_the_call_and_the_record( + client, db_session, measurement_tec_user, worker_calls, +): + recipe, task = await _camera_task(db_session, measurement_tec_user.id) + station, run = await _station_with_run( + db_session, measurement_tec_user, recipe, STATION_URL) + + await _execute(client, measurement_tec_user, task, run) + + assert worker_calls[0]["worker_url"] == STATION_URL + result = await _the_result(db_session) + assert result.executed_on == "station" + assert result.worker_url == STATION_URL + assert result.station_id == station.id + + +@pytest.mark.asyncio +async def test_station_without_url_uses_the_central_worker( + client, db_session, measurement_tec_user, worker_calls, +): + recipe, task = await _camera_task(db_session, measurement_tec_user.id) + station, run = await _station_with_run( + db_session, measurement_tec_user, recipe, None) + + await _execute(client, measurement_tec_user, task, run) + + assert worker_calls[0]["worker_url"] is None + result = await _the_result(db_session) + assert result.executed_on == "server" + assert result.worker_url == settings.vision_worker_url.rstrip("/") + assert result.station_id == station.id + + +@pytest.mark.asyncio +async def test_without_a_run_the_central_worker_is_used( + client, db_session, measurement_tec_user, worker_calls, +): + _, task = await _camera_task(db_session, measurement_tec_user.id) + + await _execute(client, measurement_tec_user, task) + + assert worker_calls[0]["worker_url"] is None + result = await _the_result(db_session) + assert result.executed_on == "server" + assert result.worker_url == settings.vision_worker_url.rstrip("/") + assert result.station_id is None + + +class _Response: + status_code = 200 + + def json(self): + return {"outputs": {}, "failures": [], "engine_version": "e" * 40} + + +class _RecordingClient: + """httpx.AsyncClient finto: registra gli URL postati, opzionalmente fallisce.""" + + def __init__(self, posts: list, raises=None): + self._posts = posts + self._raises = raises + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc_info): + return False + + async def post(self, url, *args, **kwargs): + self._posts.append(url) + if self._raises is not None: + raise self._raises + return _Response() + + +def _install(monkeypatch, raises=None) -> list: + posts: list = [] + monkeypatch.setattr( + "src.backend.services.vision_service.httpx.AsyncClient", + lambda *a, **k: _RecordingClient(posts, raises), + ) + return posts + + +@pytest.mark.asyncio +async def test_call_worker_posts_to_the_given_url(monkeypatch): + posts = _install(monkeypatch) + # lo "/" finale non deve produrre "//run" + await _call_worker(b"img", {"tools": []}, worker_url=STATION_URL + "/") + assert posts == [f"{STATION_URL}/run"] + + +@pytest.mark.asyncio +async def test_call_worker_defaults_to_the_central_url(monkeypatch): + posts = _install(monkeypatch) + await _call_worker(b"img", {"tools": []}) + assert posts == [f"{settings.vision_worker_url.rstrip('/')}/run"] + + +@pytest.mark.asyncio +async def test_station_worker_down_is_502_with_its_url_and_no_fallback(monkeypatch): + posts = _install(monkeypatch, raises=httpx.ConnectError("refused")) + with pytest.raises(HTTPException) as exc: + await _call_worker(b"img", {"tools": []}, worker_url=STATION_URL) + assert exc.value.status_code == 502 + assert STATION_URL in exc.value.detail + # una sola POST: il centrale non viene tentato + assert posts == [f"{STATION_URL}/run"] diff --git a/src/frontend/flask_app/templates/admin/stations.html b/src/frontend/flask_app/templates/admin/stations.html index b916739..be9e765 100644 --- a/src/frontend/flask_app/templates/admin/stations.html +++ b/src/frontend/flask_app/templates/admin/stations.html @@ -204,6 +204,17 @@ maxlength="255"> + {# Worker visione (#45): vuoto = worker centrale #} +
{{ _('Vuoto = worker centrale. Indirizzo completo, con http o https') }}
+