Servizio camera minimo di stazione con test e README (#72)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,144 @@
|
||||
"""Servizio camera minimo di stazione (TieMeasureFlow, #72).
|
||||
|
||||
Cosa fa: espone via HTTP l'acquisizione di una camera (vs-camera: GigE/GenTL,
|
||||
OpenCV, cartella di immagini) con lo stesso contratto di `vs-gui`, ma solo
|
||||
`grab` (copiato da vendor/visionsuite/packages/vs-gui/.../server.py):
|
||||
GET /camera/sources -> {"sources": [{name, kind, label}]}
|
||||
POST /camera/grab json {source, trigger="continuo", timeout_ms=5000} -> PNG
|
||||
Errori: 400 parametri, 404 sorgente inesistente, 503 guasto camera, 504 timeout.
|
||||
Serve perche' vs-gui ascolta solo su 127.0.0.1 e il container `client` non lo
|
||||
raggiunge; questo servizio permette di scegliere l'host di ascolto.
|
||||
|
||||
Uso (dalla radice del repo):
|
||||
vendor/visionsuite/.venv/bin/python scripts/camera_service.py \
|
||||
[--host 127.0.0.1] [--port 8765] [--folder <cartella immagini>]
|
||||
--folder abilita la sorgente finta `camera:folder` (loop sulle immagini).
|
||||
Da Docker su Linux: --host 172.17.0.1 (gateway di docker0).
|
||||
|
||||
Dipende da: venv di VisionSuite (`uv sync --all-packages` in vendor/visionsuite):
|
||||
vs-camera, fastapi, uvicorn, opencv.
|
||||
|
||||
AVVERTENZA: nessuna autenticazione. Un --host diverso dal loopback espone la
|
||||
camera a chiunque raggiunga la porta: solo su rete isolata.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import threading
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
import cv2
|
||||
from fastapi import Body, FastAPI, HTTPException
|
||||
from fastapi.responses import Response
|
||||
|
||||
from visionsuite.camera.camera_interface import CameraError, CameraGrabError
|
||||
from visionsuite.camera.sources import SourceNotFound, list_sources, open_source
|
||||
|
||||
_TRIGGER = ("continuo",)
|
||||
|
||||
# errori noti di vs-camera (inglese) -> italiano; esatti, poi per prefisso
|
||||
_MESSAGGI = {
|
||||
"Camera is not open": "La camera non è aperta",
|
||||
"Camera is not grabbing": "La camera non sta acquisendo",
|
||||
"Grab cancelled (grabbing stopped)": "Acquisizione annullata (acquisizione fermata)",
|
||||
"Failed to read frame from RTSP stream": "Lettura del fotogramma dallo stream RTSP non riuscita",
|
||||
}
|
||||
_PREFISSI = (
|
||||
("Failed to open OpenCV device", "Apertura del dispositivo OpenCV non riuscita"),
|
||||
("Failed to open RTSP stream", "Apertura dello stream RTSP non riuscita"),
|
||||
("Failed to reconnect to RTSP stream", "Riconnessione allo stream RTSP non riuscita"),
|
||||
("Failed to read frame after", "Lettura del fotogramma non riuscita dopo"),
|
||||
)
|
||||
|
||||
|
||||
def _traduci(messaggio: str) -> str:
|
||||
if messaggio in _MESSAGGI:
|
||||
return _MESSAGGI[messaggio]
|
||||
for inglese, italiano in _PREFISSI:
|
||||
if messaggio.startswith(inglese):
|
||||
return italiano + messaggio[len(inglese):]
|
||||
return messaggio
|
||||
|
||||
|
||||
def create_app(folder: Path | str | None = None) -> FastAPI:
|
||||
aperte: dict = {}
|
||||
lock = threading.Lock() # protegge SOLO i dizionari, mai un'apertura
|
||||
lock_sorgenti: dict = {} # nome sorgente -> Lock che serializza apertura e grab
|
||||
|
||||
def _config() -> dict:
|
||||
if folder is None:
|
||||
return {}
|
||||
return {"folder": {"folder": str(folder), "loop": True}}
|
||||
|
||||
def _lock_di(nome: str) -> threading.Lock:
|
||||
with lock:
|
||||
return lock_sorgenti.setdefault(nome, threading.Lock())
|
||||
|
||||
@asynccontextmanager
|
||||
async def _vita(_app):
|
||||
yield
|
||||
with lock: # alla chiusura rilascia le camere aperte
|
||||
sorgenti = list(aperte.values())
|
||||
aperte.clear()
|
||||
for s in sorgenti:
|
||||
try:
|
||||
s.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
app = FastAPI(title="TMF camera service", lifespan=_vita)
|
||||
|
||||
@app.get("/camera/sources")
|
||||
def camera_sources():
|
||||
voci = list_sources(None, cameras=True, config=_config())
|
||||
return {"sources": [{"name": v.name, "kind": v.kind, "label": v.label}
|
||||
for v in voci if v.kind in ("camera", "virtual")]}
|
||||
|
||||
@app.post("/camera/grab")
|
||||
def camera_grab(corpo: dict = Body(default=None)):
|
||||
nome = corpo.get("source") if isinstance(corpo, dict) else None
|
||||
if not isinstance(nome, str) or not nome.strip():
|
||||
raise HTTPException(400, "manca «source»")
|
||||
nome = nome.strip()
|
||||
trigger = corpo.get("trigger", "continuo")
|
||||
if trigger not in _TRIGGER:
|
||||
raise HTTPException(400, "«trigger» supportato: solo continuo")
|
||||
timeout = corpo.get("timeout_ms", 5000)
|
||||
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")
|
||||
try:
|
||||
with _lock_di(nome): # solo questa sorgente; le altre non aspettano
|
||||
with lock:
|
||||
sorgente = aperte.get(nome)
|
||||
if sorgente is None:
|
||||
sorgente = open_source(nome, None, config=_config())
|
||||
with lock:
|
||||
aperte[nome] = sorgente
|
||||
frame = sorgente.grab(timeout)
|
||||
except SourceNotFound as e:
|
||||
raise HTTPException(404, _traduci(str(e)))
|
||||
except TimeoutError as e:
|
||||
raise HTTPException(504, _traduci(str(e)) or "timeout")
|
||||
except (CameraGrabError, CameraError) as e:
|
||||
raise HTTPException(503, _traduci(str(e)))
|
||||
ok, buf = cv2.imencode(".png", frame)
|
||||
if not ok:
|
||||
raise HTTPException(500, "codifica PNG fallita")
|
||||
return Response(buf.tobytes(), media_type="image/png")
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def main() -> None:
|
||||
import uvicorn
|
||||
p = argparse.ArgumentParser(description="Servizio camera minimo di stazione (#72)")
|
||||
p.add_argument("--host", default="127.0.0.1")
|
||||
p.add_argument("--port", type=int, default=8765)
|
||||
p.add_argument("--folder", default=None, help="cartella immagini per camera:folder")
|
||||
a = p.parse_args()
|
||||
uvicorn.run(create_app(a.folder), host=a.host, port=a.port)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,116 @@
|
||||
"""Test di scripts/camera_service.py (#72).
|
||||
|
||||
Uso: vendor/visionsuite/.venv/bin/python -m pytest scripts/tests/test_camera_service.py -q
|
||||
Dipende da: venv di VisionSuite (vs-camera, fastapi, httpx, opencv, pytest).
|
||||
"""
|
||||
import importlib.util
|
||||
from pathlib import Path
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
MODULO = Path(__file__).resolve().parents[1] / "camera_service.py"
|
||||
|
||||
|
||||
def _carica():
|
||||
spec = importlib.util.spec_from_file_location("camera_service", MODULO)
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod)
|
||||
return mod
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def cartella(tmp_path):
|
||||
for nome, (w, h) in {"a.png": (64, 48), "b.png": (64, 48)}.items():
|
||||
cv2.imwrite(str(tmp_path / nome), np.full((h, w, 3), 128, np.uint8))
|
||||
return tmp_path
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(cartella):
|
||||
mod = _carica()
|
||||
with TestClient(mod.create_app(cartella)) as c:
|
||||
c.mod = mod
|
||||
yield c
|
||||
|
||||
|
||||
def test_sources_contiene_folder(client):
|
||||
r = client.get("/camera/sources")
|
||||
assert r.status_code == 200
|
||||
nomi = [s["name"] for s in r.json()["sources"]]
|
||||
assert "camera:folder" in nomi
|
||||
voce = next(s for s in r.json()["sources"] if s["name"] == "camera:folder")
|
||||
assert {"name", "kind", "label"} <= set(voce)
|
||||
|
||||
|
||||
def test_grab_folder_restituisce_png(client):
|
||||
r = client.post("/camera/grab", json={"source": "camera:folder"})
|
||||
assert r.status_code == 200
|
||||
assert r.headers["content-type"] == "image/png"
|
||||
img = cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_COLOR)
|
||||
assert img.shape[:2] == (48, 64)
|
||||
|
||||
|
||||
def test_grab_ripetuto_in_loop(client):
|
||||
# due immagini in cartella, tre grab: il loop non deve esaurirsi
|
||||
for _ in range(3):
|
||||
assert client.post("/camera/grab", json={"source": "camera:folder"}).status_code == 200
|
||||
|
||||
|
||||
def test_source_mancante_400(client):
|
||||
assert client.post("/camera/grab", json={}).status_code == 400
|
||||
assert client.post("/camera/grab", json={"source": " "}).status_code == 400
|
||||
|
||||
|
||||
def test_source_inesistente_404(client):
|
||||
assert client.post("/camera/grab", json={"source": "camera:non-esiste"}).status_code == 404
|
||||
|
||||
|
||||
def test_timeout_non_valido_400(client):
|
||||
r = client.post("/camera/grab", json={"source": "camera:folder", "timeout_ms": 0})
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_errore_camera_503_tradotto(client, monkeypatch):
|
||||
from visionsuite.camera.camera_interface import CameraGrabError
|
||||
|
||||
class Rotta:
|
||||
def grab(self, timeout):
|
||||
raise CameraGrabError("Camera is not open")
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(client.mod, "open_source", lambda *a, **k: Rotta())
|
||||
r = client.post("/camera/grab", json={"source": "camera:finta"})
|
||||
assert r.status_code == 503
|
||||
assert r.json()["detail"] == "La camera non è aperta"
|
||||
|
||||
|
||||
def test_timeout_504(client, monkeypatch):
|
||||
class Lenta:
|
||||
def grab(self, timeout):
|
||||
raise TimeoutError("scaduto")
|
||||
|
||||
def close(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(client.mod, "open_source", lambda *a, **k: Lenta())
|
||||
r = client.post("/camera/grab", json={"source": "camera:lenta"})
|
||||
assert r.status_code == 504
|
||||
|
||||
|
||||
def test_sorgente_aperta_una_volta(client, monkeypatch):
|
||||
aperture = []
|
||||
reale = client.mod.open_source
|
||||
|
||||
def conta(*a, **k):
|
||||
aperture.append(a[0])
|
||||
return reale(*a, **k)
|
||||
|
||||
monkeypatch.setattr(client.mod, "open_source", conta)
|
||||
for _ in range(3):
|
||||
client.post("/camera/grab", json={"source": "camera:folder"})
|
||||
assert aperture == ["camera:folder"]
|
||||
Reference in New Issue
Block a user