diff --git a/scripts/camera_service.py b/scripts/camera_service.py index 800a065..e272e79 100644 --- a/scripts/camera_service.py +++ b/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 /settaggi/-.json; alla prima apertura della camera nel servizio il file, se c'e', viene ricaricato. -Le calibrazioni stanno in /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 /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(): diff --git a/scripts/tests/test_camera_orientamento.py b/scripts/tests/test_camera_orientamento.py new file mode 100644 index 0000000..6d10af9 --- /dev/null +++ b/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