Specchiatura orizzontale/verticale per sorgente nel servizio camera, salvata su disco. Refs #110

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

scripts/camera_service.py

@@ -61,7 +61,17 @@ Acquisizione (#87): anteprima, esposizione/guadagno, settaggi di default:
   POST   /camera/parametri/salva json {source} -> {source, salvati, file}
          scrive <config-dir>/settaggi/<nome>-<sha1>.json; alla prima apertura
          della camera nel servizio il file, se c'e', viene ricaricato.
-Le calibrazioni stanno in <config-dir>/calibrazioni, le attive in attive.json (default
+  #110: specchiatura per sorgente, in software (cv2.flip) su OGNI frame letto dalla
+  sorgente, prima di raddrizzatura/calibrazione (anteprima, grab, calibrazione da camera,
+  copia bloccata); non tocca l'immagine caricata a mano in /camera/calibrate.
+  GET /camera/orientamento?source=S -> {source, orizzontale, verticale} (default false/false);
+         400 se manca source
+  PUT /camera/orientamento json {source, orizzontale: bool, verticale: bool} -> come GET +
+         "avviso" (stringa o null: la calibrazione attiva ha una mappa fatta con l'orientamento
+         di prima e va rifatta). 400 se source manca o i due campi non sono bool. Salva subito
+         su disco; scarta i frame tenuti (anteprima/blocca).
+Le calibrazioni stanno in <config-dir>/calibrazioni, le attive in attive.json, l'orientamento
+per sorgente in orientamento.json (#110) (default
 %ProgramData%/Tielogic/TieMeasureFlow/acquisitore su Windows, ./acquisitore altrove).
 
 Serve perche' vs-gui ascolta solo su 127.0.0.1 e il container `client` non lo
@@ -87,6 +97,7 @@ camera a chiunque raggiunga la porta: solo su rete isolata.
 from __future__ import annotations
 
 import argparse
+import json  # #110: orientamento.json
 import logging
 import math  # #95: mm/px finito nella calibrazione manuale
 import os
@@ -167,6 +178,36 @@ def create_app(folder: Path | str | None = None,
     lock_sorgenti: dict = {}  # nome sorgente -> Lock che serializza apertura e grab
     calcoli: OrderedDict = OrderedDict()  # #91: calcolo_id -> (dati, mappa); gli ultimi _MAX_CALCOLI
     lock_calcoli = threading.Lock()
+    file_orientamento = archivio.cartella / "orientamento.json"  # #110: {sorgente: {orizzontale, verticale}}
+
+    def _carica_orientamenti() -> dict:
+        """#110: orientamenti salvati; file assente o rotto -> {} (mai eccezione)."""
+        if not file_orientamento.is_file():
+            return {}
+        try:
+            dati = json.loads(file_orientamento.read_text(encoding="utf-8"))
+            if not isinstance(dati, dict):
+                raise ValueError("non e' un oggetto JSON")
+            return {n: {"orizzontale": v.get("orizzontale") is True, "verticale": v.get("verticale") is True}
+                    for n, v in dati.items() if isinstance(n, str) and isinstance(v, dict)}
+        except (OSError, ValueError) as e:
+            _log.warning("orientamento non caricato (%s): %s", file_orientamento, e)
+            return {}
+
+    orientamenti: dict = _carica_orientamenti()  # #110: sorgente -> {"orizzontale": bool, "verticale": bool}; protetto da `lock`
+
+    def _specchia(nome: str, frame):
+        """#110: applica l'orientamento della sorgente al frame letto; nessuno -> lo stesso oggetto, senza copia."""
+        with lock:
+            o = orientamenti.get(nome) or {}
+        orizz, vert = o.get("orizzontale", False), o.get("verticale", False)
+        if orizz and vert:
+            return cv2.flip(frame, -1)  # entrambi = rotazione di 180 gradi
+        if orizz:
+            return cv2.flip(frame, 1)
+        if vert:
+            return cv2.flip(frame, 0)
+        return frame
 
     def _config() -> dict:
         if folder is None:
@@ -317,7 +358,7 @@ def create_app(folder: Path | str | None = None,
 
     def _grab(nome: str, timeout: int):
         """Un frame dalla sorgente (apre e tiene aperta); errori camera -> HTTP."""
-        return _con_sorgente(nome, lambda s: s.grab(timeout))  # #87: corpo in _con_sorgente
+        return _con_sorgente(nome, lambda s: _specchia(nome, s.grab(timeout)))  # #87: corpo in _con_sorgente; #110: specchiato prima di tutto il resto
 
     @app.post("/camera/grab")
     def camera_grab(corpo: dict = Body(default=None)):
@@ -544,7 +585,7 @@ def create_app(folder: Path | str | None = None,
         if isinstance(timeout, bool) or not isinstance(timeout, int) or not 1 <= timeout <= 10000:
             raise HTTPException(400, "«timeout_ms» deve essere un intero fra 1 e 10000")
         def leggi(s):  # #92: il frame grezzo si tiene sotto il lock della sorgente
-            f = s.grab(timeout)
+            f = _specchia(nome, s.grab(timeout))  # #110: specchiato PRIMA di tenerlo, cosi' blocca/calibrata lo ricevono gia' specchiato
             with lock:
                 grezzi[nome] = f.copy()
             return f
@@ -686,6 +727,49 @@ def create_app(folder: Path | str | None = None,
 
         return _con_sorgente(nome, salva)
 
+    # ---- #110: specchiatura orizzontale/verticale per sorgente
+
+    def _orientamento_di(nome: str) -> dict:
+        with lock:
+            o = orientamenti.get(nome) or {}
+        return {"source": nome, "orizzontale": bool(o.get("orizzontale")), "verticale": bool(o.get("verticale"))}
+
+    @app.get("/camera/orientamento")
+    def orientamento_leggi(source: str = ""):
+        return _orientamento_di(_sorgente(source))
+
+    @app.put("/camera/orientamento")
+    def orientamento_scrivi(corpo: dict = Body(default=None)):
+        corpo = corpo if isinstance(corpo, dict) else {}
+        nome = _sorgente(corpo.get("source"))
+        orizz, vert = corpo.get("orizzontale"), corpo.get("verticale")
+        if not isinstance(orizz, bool) or not isinstance(vert, bool):
+            raise HTTPException(400, "«orizzontale» e «verticale» devono essere true o false")
+        with lock:
+            prima = orientamenti.get(nome) or {}
+            cambiato = (bool(prima.get("orizzontale")), bool(prima.get("verticale"))) != (orizz, vert)
+            if orizz or vert:
+                orientamenti[nome] = {"orizzontale": orizz, "verticale": vert}
+            else:
+                orientamenti.pop(nome, None)  # tutti false: la sorgente esce dal file
+            # scrittura atomica di tutto il dizionario (tmp + replace)
+            try:
+                file_orientamento.parent.mkdir(parents=True, exist_ok=True)
+                tmp = file_orientamento.with_name(file_orientamento.name + ".tmp")
+                tmp.write_text(json.dumps(orientamenti, ensure_ascii=False, indent=1), encoding="utf-8")
+                os.replace(tmp, file_orientamento)
+            except OSError as e:
+                _log.warning("orientamento non salvato (%s): %s", file_orientamento, e)
+                raise HTTPException(500, "salvataggio dell'orientamento non riuscito")
+            # i frame tenuti hanno l'orientamento di prima
+            grezzi.pop(nome, None)
+            bloccati.pop(nome, None)
+        avviso = None
+        if cambiato and archivio.leggi_mappa(nome) is not None:
+            avviso = (f"la calibrazione attiva «{archivio.attiva_di(nome)}» corregge la distorsione "
+                      "con l'orientamento di prima: va rifatta")
+        return {**_orientamento_di(nome), "avviso": avviso}
+
     @app.get("/calibrazione", response_class=HTMLResponse)
     @app.get("/acquisizione", response_class=HTMLResponse)  # #87: stessa pagina
     def calibrazione_pagina():

scripts/tests/test_camera_orientamento.py

@@ -0,0 +1,168 @@
+"""Test della specchiatura orizzontale/verticale per sorgente (#110): scripts/camera_service.py.
+
+Uso: uv run --python 3.13 --extra vision-worker --extra dev \
+       --with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q
+Copre: GET/PUT /camera/orientamento, effetto su /camera/grab e /camera/anteprima,
+persistenza al riavvio, file rotto, validazione, scarto dei frame tenuti, avviso
+sulla calibrazione densa. Sorgente: camera:folder su un'immagine 4x3 asimmetrica.
+"""
+import importlib.util
+from pathlib import Path
+
+import cv2
+import numpy as np
+import pytest
+from fastapi.testclient import TestClient
+
+from test_calibrazione import (  # calibrazione densa reale (#84), per l'avviso
+    SORGENTE as SORGENTE_DENSA, cartella_densa, carica_su, client_denso, salva_con_nome,
+)
+
+__all__ = ["cartella_densa", "client_denso"]  # fixture importate, restano usate
+
+MODULO = Path(__file__).resolve().parents[1] / "camera_service.py"
+SORGENTE = "camera:folder"
+
+
+def _carica():
+    spec = importlib.util.spec_from_file_location("camera_service_orient", MODULO)
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod
+
+
+@pytest.fixture
+def cartella(tmp_path):
+    # #110: 4x3 tutta nera con UN pixel acceso in alto a sinistra (riga 0, colonna 0)
+    img = tmp_path / "foto"
+    img.mkdir()
+    f = np.zeros((3, 4, 3), np.uint8)
+    f[0, 0] = 255
+    cv2.imwrite(str(img / "a.png"), f)
+    return img
+
+
+@pytest.fixture
+def config(tmp_path):
+    return tmp_path / "cfg"
+
+
+@pytest.fixture
+def client(cartella, config):
+    mod = _carica()
+    with TestClient(mod.create_app(cartella, config)) as c:
+        yield c
+
+
+def _acceso(png: bytes):
+    """(riga, colonna) del pixel acceso nel PNG; controlla che sia uno solo."""
+    f = cv2.imdecode(np.frombuffer(png, np.uint8), cv2.IMREAD_COLOR)
+    assert f.shape[:2] == (3, 4)
+    r, c = np.nonzero(f.max(axis=2) > 128)
+    assert len(r) == 1
+    return int(r[0]), int(c[0])
+
+
+def _imposta(c, orizz, vert, source=SORGENTE):
+    return c.put("/camera/orientamento", json={"source": source, "orizzontale": orizz, "verticale": vert})
+
+
+def _grab(c):
+    r = c.post("/camera/grab", json={"source": SORGENTE})
+    assert r.status_code == 200, r.text
+    return _acceso(r.content)
+
+
+def test_get_senza_file_false_false(client):
+    r = client.get("/camera/orientamento", params={"source": SORGENTE})
+    assert r.status_code == 200
+    assert r.json() == {"source": SORGENTE, "orizzontale": False, "verticale": False}
+
+
+def test_get_senza_source_400(client):
+    assert client.get("/camera/orientamento").status_code == 400
+
+
+@pytest.mark.parametrize("orizz,vert,atteso", [
+    (False, False, (0, 0)),  # originale: alto a sinistra
+    (True, False, (0, 3)),   # orizzontale: alto a destra
+    (False, True, (2, 0)),   # verticale: basso a sinistra
+    (True, True, (2, 3)),    # entrambi: basso a destra (180 gradi)
+])
+def test_grab_specchiato(client, orizz, vert, atteso):
+    r = _imposta(client, orizz, vert)
+    assert r.status_code == 200, r.text
+    assert (r.json()["orizzontale"], r.json()["verticale"]) == (orizz, vert)
+    assert client.get("/camera/orientamento", params={"source": SORGENTE}).json()["orizzontale"] is orizz
+    assert _grab(client) == atteso
+
+
+def test_tutti_false_toglie_la_sorgente_dal_file(client, config):
+    _imposta(client, True, False)
+    assert SORGENTE in (config / "orientamento.json").read_text(encoding="utf-8")
+    _imposta(client, False, False)
+    assert SORGENTE not in (config / "orientamento.json").read_text(encoding="utf-8")
+    assert _grab(client) == (0, 0)
+
+
+def test_anteprima_specchiata_dimensioni_invariate(client):
+    _imposta(client, True, True)
+    r = client.post("/camera/anteprima", json={"source": SORGENTE, "lato_max": 4096})
+    assert r.status_code == 200, r.text
+    assert (r.headers["x-larghezza"], r.headers["x-altezza"]) == ("4", "3")
+    f = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_GRAYSCALE)
+    assert f.shape == (3, 4)
+    # JPEG con perdite: il pixel acceso e' il piu' chiaro, in basso a destra
+    assert np.unravel_index(int(np.argmax(f)), f.shape) == (2, 3)
+
+
+def test_persistenza_al_riavvio(cartella, config):
+    mod = _carica()
+    with TestClient(mod.create_app(cartella, config)) as c1:
+        assert _imposta(c1, True, False).status_code == 200
+    with TestClient(_carica().create_app(cartella, config)) as c2:
+        j = c2.get("/camera/orientamento", params={"source": SORGENTE}).json()
+        assert (j["orizzontale"], j["verticale"]) == (True, False)
+        assert _grab(c2) == (0, 3)
+
+
+def test_file_rotto_non_blocca_l_avvio(cartella, config):
+    config.mkdir(parents=True)
+    (config / "orientamento.json").write_text("{non e json", encoding="utf-8")
+    with TestClient(_carica().create_app(cartella, config)) as c:
+        j = c.get("/camera/orientamento", params={"source": SORGENTE}).json()
+        assert (j["orizzontale"], j["verticale"]) == (False, False)
+        assert _imposta(c, False, True).status_code == 200  # e si puo' riscrivere
+
+
+def test_put_campi_non_bool_400(client):
+    for corpo in ({"source": SORGENTE, "orizzontale": 1, "verticale": False},
+                  {"source": SORGENTE, "orizzontale": True, "verticale": "si"},
+                  {"source": SORGENTE, "orizzontale": True},
+                  {"orizzontale": True, "verticale": True}):
+        assert client.put("/camera/orientamento", json=corpo).status_code == 400, corpo
+
+
+def test_put_scarta_i_frame_tenuti(client):
+    assert client.post("/camera/anteprima", json={"source": SORGENTE}).status_code == 200
+    assert client.post("/camera/blocca", json={"source": SORGENTE}).status_code == 200
+    _imposta(client, True, False)
+    assert client.post("/camera/blocca", json={"source": SORGENTE}).status_code == 404
+    # col frame tenuto scartato anche l'ultimo grezzo dell'anteprima e' sparito
+    assert client.post("/camera/anteprima/calibrata", json={"source": SORGENTE}).status_code == 404
+
+
+def test_avviso_senza_mappa_e_null(client):
+    j = _imposta(client, True, False).json()
+    assert j["avviso"] is None
+
+
+def test_avviso_con_mappa_attiva(client_denso):
+    densa = client_denso.post("/camera/calibrate", data={"source": SORGENTE_DENSA, "square_mm": "2"}).json()
+    assert salva_con_nome(client_denso, densa, "Densa").json()["metodo"] == "densa"
+    assert carica_su(client_denso, SORGENTE_DENSA, "Densa").status_code == 200
+    r = _imposta(client_denso, True, False, source=SORGENTE_DENSA)
+    assert r.status_code == 200, r.text
+    assert "«Densa»" in r.json()["avviso"] and "va rifatta" in r.json()["avviso"]
+    # stesso valore di nuovo: nulla e' cambiato, nessun avviso
+    assert _imposta(client_denso, True, False, source=SORGENTE_DENSA).json()["avviso"] is None