Compare commits

...

20 Commits

Author SHA1 Message Date
Adriano e38e1a2eaf Calibrazione densa: finestra sub-pixel ristretta al bordo, test su scacchiera continua (#84)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
2026-10-24 00:00:36 +02:00
Adriano 7fa5bacdfe Acquisitore: calibrazione densa su C2/grab/pagina, test (#84)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
2026-10-23 23:56:15 +02:00
Adriano 896a30b376 Calibrazione densa da una foto: griglia, mappa di distorsione, archivio .npz (#84)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Jbf7qKuxsnAVjxthRy7dnB
2026-10-23 23:53:48 +02:00
Adriano 351739ec36 Acquisitore: calibrazione a scacchiera per sorgente, rotte e pagina /calibrazione (#83)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-23 22:43:32 +02:00
Adriano 9f2d624b4f Worker: endpoint /recipes/dxf-compare con test (#83)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-23 22:40:08 +02:00
Adriano 9fe82d7cef Costruttore ricetta DXF compare dal campo, con test (#83)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-23 22:39:21 +02:00
Adriano ae2353f7f8 Pacchetto Windows: registra gli strumenti senza entry point e controlla il catalogo nel build (#82)
rilascio-windows / setup-windows (push) Successful in 32m30s
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VMYrdz6uBbuwbaQ9MqDmMc
2026-10-19 23:03:48 +02:00
Adriano d652891c1c Merge lavoro/daitarn-tmf-81 (TieMeasureFlow#81)
rilascio-windows / setup-windows (push) Successful in 34m11s
2026-10-17 14:25:56 +02:00
Adriano 0b22643d69 Merge lavoro/daitarn-tmf-80 (TieMeasureFlow#80) 2026-10-17 14:25:56 +02:00
Adriano 15e31e2b8a Merge lavoro/daitarn-tmf-79 (TieMeasureFlow#79) 2026-10-17 14:25:55 +02:00
Adriano f8f5f248ff setup: init-db e reset-db stampano head in un thread (TieMeasureFlow#80)
_stamp_head esegue env.py di alembic, che fa asyncio.run() e dentro il loop
di FastAPI dava RuntimeError (500). Ora gira con asyncio.to_thread. Nuovo
test_setup_stamp.py senza monkeypatch di _stamp_head, su file SQLite.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-17 14:25:24 +02:00
Adriano f33415df6a Test frontend: niente piu' dipendenza dall'ordine sulla stazione (TieMeasureFlow#79)
test_measure.py non ricarica piu' config/blueprints.measure (monkeypatch.setattr,
annullato al teardown); TestTaskComplete e test_recipe_rules_ui.py fissano la
stazione esplicitamente con una fixture autouse.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-17 14:13:15 +02:00
Adriano 62b5e412ca static del frontend sempre LF (.gitattributes): impronte di VERSIONS.md e pacchetto Windows identici (TieMeasureFlow#81)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-17 14:10:39 +02:00
Adriano 0e92959022 Merge piano C: setup Windows di TieMeasureFlow (TieMeasureFlow#76)
rilascio-windows / setup-windows (push) Successful in 32m46s
2026-10-17 12:39:35 +02:00
Adriano f23e87e585 Merge piano B: report PDF su Windows offline (TieMeasureFlow#78) 2026-10-17 12:39:35 +02:00
Adriano 114e4a10ae Merge piano A: backend su SQLite (TieMeasureFlow#77) 2026-10-17 12:39:34 +02:00
Adriano 0a68a9995e Pacchetto Windows: LEGGIMI dei report PDF (DLL, variabili, flag Nuitka, licenze) (TieMeasureFlow#78)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-17 09:18:21 +02:00
Adriano 8431aa7095 Report PDF su Windows: MathJax locale per Kaleido (niente CDN offline) e test con WeasyPrint/Edge veri (TieMeasureFlow#78)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-17 09:16:29 +02:00
Adriano 23012fdfbb Pacchetto Windows: prendi-gtk.ps1 scarica le DLL Pango/GTK pinnate con sha256 (TieMeasureFlow#78)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-17 09:10:11 +02:00
Adriano 242add047b Backend su file SQLite locale: DB_URL, PRAGMA FK/WAL, UPLOAD_DIR assoluto (TieMeasureFlow#77)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-17 09:07:45 +02:00
26 changed files with 2765 additions and 62 deletions
+3
View File
@@ -0,0 +1,3 @@
# TieMeasureFlow#81: static del frontend sempre con LF, anche con core.autocrlf=true su Windows:
# le impronte sha256 di static/vendor/VERSIONS.md (test_offline.py) e il pacchetto Windows le vogliono identiche.
src/frontend/flask_app/static/** text=auto eol=lf
+3
View File
@@ -73,3 +73,6 @@ docker-compose.override.yml
# Subagent-driven development scratch: ledger, briefs, review packages
.superpowers/
# Generato da packaging/windows/build.ps1 (TieMeasureFlow#82): moduli-strumento dagli entry point
packaging/windows/_strumenti.py
+1
View File
@@ -347,6 +347,7 @@ Variabili d'ambiente in `.env` (copiare da `.env.example`):
- ⚠️ Alembic carica `env.py` e `versions\` dal disco, quindi Nuitka non può compilarli: `build.ps1` li copia come soli `.pyc` (`compileall -b`) e `avvio.py` esegue `upgrade head` con `sourceless=true`. È l'unica eccezione al «nessun `.py`/`.pyc`»: il controllo finale di `build.ps1` ammette `.pyc` solo sotto `src\backend\migrations`. `env.py` chiama `fileConfig`, che azzera i gestori del log: `avvio.py` li ricrea dopo le migrazioni.
- ⚠️ `numba` è escluso dal pacchetto (`--nofollow-import-to=numba`): le funzioni compilate da Nuitka non hanno bytecode e numba non può farne il JIT. Core e pm2d hanno il ripiego in Python puro (`HAS_NUMBA=False`): funziona ma il matching è più lento. Da misurare sul TC bench.
- ⚠️ L'exe è senza console: `sys.stdout`/`sys.stderr` possono essere `None` (`avvio.py` li sostituisce con il device nullo), uvicorn gira con `log_config=None` e il log va su file (`...\log\tiemeasureflow.log`). I server uvicorn girano in thread (nessun signal handler) e waitress si crea con `create_server`.
- ⚠️ Strumenti di visione (TieMeasureFlow#82): nel binario non ci sono i metadati di `vs-tools`, quindi `load_tools()` (entry point `visionsuite.task.tools`) non trova nulla e ogni ricetta fallisce con tipo sconosciuto. `build.ps1` genera `packaging/windows/_strumenti.py` (`MODULI`, gitignorato, letto dal venv del build), `avvio.py` `registra_strumenti()` lo importa prima dei server, e prima di Inno Setup `build.ps1` lancia l'exe con `--controlla-strumenti` (senza licenza né server; JSON in `...\log\controllo-strumenti.json`, exit 3 se il catalogo è vuoto o un modulo non si importa).
- ⚠️ `visionsuite` è un namespace condiviso da più distribuzioni: `build.ps1` elenca i sottopacchetti dagli `__path__` e i metadati con `--include-distribution-metadata`. Un nuovo pacchetto di Core si prende da solo se è in `[tool.uv.sources]` e nel gruppo.
- ⚠️ `tailwind.css` non è nel repo: `build.ps1` lo genera con `npx tailwindcss@3.4.19` (stessa versione di `Dockerfile.frontend`); serve Internet solo a chi costruisce. Il setup finale non ha bisogno di Internet.
- ⚠️ Chrome o Edge servono a kaleido per i grafici dei report: `avvio.py` imposta `BROWSER_PATH` al primo trovato. Le DLL di WeasyPrint le scarica `packaging/windows/pdf/prendi-gtk.ps1` (piano #78) in `gtk\` accanto all'exe.
+59
View File
@@ -17,6 +17,11 @@ Come si usa: e' l'exe compilato con Nuitka (build.ps1); da sorgente, dalla radic
con l'ambiente 3.13 del pacchetto: `python packaging/windows/avvio.py [--impronta] [--host H]`.
Opzioni di servizio: `--senza-finestre` (niente MessageBox, solo stderr: serve alle prove
automatiche; NON salta il controllo di licenza) e `--senza-browser`.
`--controlla-strumenti` (TieMeasureFlow#82): NON richiede la licenza, non avvia server, migrazioni
ne' database; registra gli strumenti, scrive su stdout e in
%PROGRAMDATA%\\Tielogic\\TieMeasureFlow\\log\\controllo-strumenti.json un JSON
{strumenti, kind, falliti, unavailable} ed esce con 0 (strumenti > 0 e nessun fallito) o 3.
La usa build.ps1 prima di Inno Setup.
Dipende da: visionsuite.license (vs-license, contratto licenza v1), uvicorn, waitress, alembic.
@@ -25,11 +30,15 @@ Dipende da: visionsuite.license (vs-license, contratto licenza v1), uvicorn, wai
⚠️ Le variabili d'ambiente vanno impostate PRIMA di importare backend e frontend (leggono
la configurazione all'import): per questo gli import pesanti stanno dentro le funzioni.
⚠️ La cartella ProgramData la rende scrivibile agli utenti l'installer (.iss).
⚠️ Nel binario Nuitka mancano i metadati di vs-tools, quindi `load_tools()` (che cerca gli
strumenti come entry point) non ne trova nessuno: `build.ps1` genera `_strumenti.py`
(MODULI, dagli entry point del venv) e `registra_strumenti()` li importa per nome (#82).
"""
from __future__ import annotations
import argparse
import ctypes
import importlib
import json
import logging
import os
@@ -264,6 +273,50 @@ def _prepara_log(dati: Path | None = None) -> None:
pass
# ---------------------------------------------------------------- strumenti di visione
def registra_strumenti() -> list[str]:
"""Registra gli strumenti di Core; torna i moduli che non si sono importati (#82).
Nel pacchetto Nuitka gli entry point non esistono: `load_tools()` non trova nulla e
l'elenco dei moduli arriva da `_strumenti.py`, generato da build.ps1. Da sorgente
`_strumenti` non c'e' e bastano gli entry point. Il worker di visione gira in un thread
di questo processo: registrare una volta sola qui vale per tutti."""
from visionsuite.task.registry import load_tools
load_tools()
try:
from _strumenti import MODULI # generato da build.ps1, accanto ad avvio.py
except ImportError:
MODULI = () # da sorgente: gli entry point di load_tools() bastano
falliti: list[str] = []
for modulo in MODULI:
try:
importlib.import_module(modulo) # eseguire il modulo registra i suoi @tool
except Exception: # un modulo rotto non ferma l'app: si logga e si raccoglie
log.error("strumenti %s non importati", modulo, exc_info=True)
falliti.append(modulo)
return falliti
def controlla_strumenti() -> int:
"""Senza licenza, server ne' DB: scrive il JSON del catalogo strumenti. Exit 0 se completo, 3 altrimenti."""
from visionsuite.task.graph import registry
from visionsuite.task.registry import unavailable
falliti = registra_strumenti()
kind = sorted(registry())
esito = {"strumenti": len(kind), "kind": kind, "falliti": falliti, "unavailable": unavailable()}
riga = json.dumps(esito, ensure_ascii=False)
if sys.stdout is not None:
print(riga)
try:
f = cartella_dati() / "log" / "controllo-strumenti.json"
f.parent.mkdir(parents=True, exist_ok=True)
f.write_text(riga + "\n", encoding="utf-8") # l'exe e' senza console: il file e' l'unica uscita sicura
except OSError as e:
log.warning("controllo-strumenti.json non scritto: %s", e)
return 0 if kind and not falliti else 3
# ---------------------------------------------------------------- rete
def _porta_aperta(porta: int) -> bool:
@@ -402,6 +455,8 @@ def main(argv=None) -> int:
p.add_argument("--host", default="127.0.0.1", help="indirizzo di ascolto dell'interfaccia web (default 127.0.0.1)")
p.add_argument("--senza-finestre", action="store_true", help="niente MessageBox (prove automatiche)")
p.add_argument("--senza-browser", action="store_true", help="non aprire il browser (prove automatiche)")
p.add_argument("--controlla-strumenti", action="store_true",
help="registra gli strumenti, scrive il JSON ed esce (senza licenza ne' server; usata da build.ps1)")
args = p.parse_args(argv)
_prepara_log()
@@ -415,6 +470,9 @@ def main(argv=None) -> int:
messaggio(f"Impronta di questo computer:\n{impronta}", errore=False, senza_finestre=args.senza_finestre)
return 0
if args.controlla_strumenti:
return controlla_strumenti() # prima della licenza: la build non ha una licenza
if not controlla_licenza(args.senza_finestre):
return 1
@@ -449,6 +507,7 @@ def main(argv=None) -> int:
_prepara_log(dati) # alembic ha riconfigurato il logging
log.info("migrazioni applicate (primo avvio: %s)", primo_avvio)
registra_strumenti() # prima del server: il worker di visione legge registry() gia' popolato (#82)
thread = avvia_server(args.host, dati)
if not args.senza_browser:
# database senza utenti: si apre la pagina di setup del backend (SETUP_PASSWORD in config.env)
+37 -2
View File
@@ -26,6 +26,10 @@ segue da solo, quindi i sottopacchetti si elencano uno per uno dagli __path__ in
⚠️ numba e' escluso di proposito: le funzioni compilate da Nuitka non hanno bytecode e numba
non puo' farne il JIT. Core e pm2d hanno gia' il ripiego in Python puro (HAS_NUMBA=False):
funziona ma e' piu' lento nel matching.
⚠️ Gli strumenti di vs-tools si registrano solo come entry point (visionsuite.task.tools) e i
metadati di vs-tools non stanno nel binario (TieMeasureFlow#82): il build genera
packaging\windows\_strumenti.py (MODULI, gitignorato) che avvio.py importa, e prima di
Inno Setup esegue l'exe con --controlla-strumenti: se il catalogo e' vuoto il build fallisce.
⚠️ Il frontend Flask non e' un pacchetto (`from config import Config`, `blueprints`, `services`):
si compila mettendo flask_app e scripts su PYTHONPATH, come fa avvio.py in sviluppo.
#>
@@ -85,8 +89,9 @@ try {
$pacchetti = & $py -c "import visionsuite, pkgutil; print('\n'.join('visionsuite.' + m.name for m in pkgutil.iter_modules(visionsuite.__path__) if m.ispkg))"
if ($LASTEXITCODE -ne 0 -or -not $pacchetti) { throw "Elenco dei pacchetti visionsuite non riuscito" }
$includi = @($pacchetti -split "\r?\n" | Where-Object { $_ } | ForEach-Object { "--include-package=$_" })
# Nessun --include-distribution-metadata per vs-*: nulla li legge e Nuitka rifiuta i metadati di una
# distribuzione il cui pacchetto (visionsuite.camera, namespace) non e' di primo livello.
# Nessun --include-distribution-metadata per vs-*: Nuitka rifiuta i metadati di una distribuzione il cui
# pacchetto (visionsuite.camera, namespace) non e' di primo livello. Gli entry point di vs-tools, che
# load_tools() cercherebbe nei metadati, si rimpiazzano con _strumenti.py (sotto, TieMeasureFlow#82).
Write-Host "Pacchetti visionsuite: $($includi -join ' ')"
# Flag Nuitka del piano PDF (WeasyPrint, kaleido, pyphen, plotly, tzdata...): le righe "--include-..."
@@ -103,6 +108,25 @@ if (Test-Path $leggimi) {
# Nuitka cerca i moduli non-pacchetto (config, services, blueprints, app, camera_service) su PYTHONPATH
$env:PYTHONPATH = (@($radice, $flask, (Join-Path $radice 'scripts')) -join ';')
# TieMeasureFlow#82: i moduli-strumento dagli entry point del venv del build (nel binario non ci sono i
# metadati di vs-tools, quindi load_tools() non li troverebbe). Si usa $py, non `uv run`.
$strumenti = Join-Path $radice 'packaging\windows\_strumenti.py'
$genera = @'
import sys
from importlib.metadata import entry_points
moduli = sorted({ep.value for ep in entry_points(group="visionsuite.task.tools")})
if not moduli:
sys.exit(1)
righe = ["# Generato da build.ps1 (TieMeasureFlow#82): non modificare, non committare.", "MODULI = ("]
righe += [f" {m!r}," for m in moduli] + [")"]
with open(sys.argv[1], "w", encoding="utf-8", newline="") as f:
print(*righe, sep=chr(10), file=f)
print(len(moduli))
'@
$nModuli = $genera | & $py - $strumenti
if ($LASTEXITCODE -ne 0) { throw "Entry point visionsuite.task.tools non trovati nel venv: nessuno strumento da includere" }
Write-Host "Moduli-strumento ($nModuli) in $strumenti"
$out = Join-Path $radice 'build\nuitka'
New-Item -ItemType Directory -Force $out | Out-Null
$nuitka = @(
@@ -201,6 +225,17 @@ $sporchi = Get-ChildItem $sorgente -Recurse -Include *.py, *.pyc -File |
Where-Object { $_.FullName -notlike "$mig\*" -or $_.Extension -eq '.py' }
if ($sporchi) { throw "Il pacchetto contiene sorgenti/bytecode:`n$($sporchi.FullName -join "`n")" }
# TieMeasureFlow#82: l'exe deve vedere gli strumenti (--controlla-strumenti non vuole licenza ne' apre server/DB)
$exe = Join-Path $sorgente 'TieMeasureFlow.exe'
$jsonControllo = Join-Path $env:PROGRAMDATA 'Tielogic\TieMeasureFlow\log\controllo-strumenti.json'
Remove-Item $jsonControllo -ErrorAction SilentlyContinue # un JSON vecchio non deve far passare il controllo
# l'exe e' GUI-subsystem: con & non si aspetta, serve Start-Process -Wait
$proc = Start-Process -FilePath $exe -ArgumentList '--controlla-strumenti' -Wait -PassThru
$esito = if (Test-Path $jsonControllo) { Get-Content $jsonControllo -Raw } else { '(controllo-strumenti.json non scritto)' }
if ($proc.ExitCode -ne 0) { throw "Controllo strumenti fallito (codice $($proc.ExitCode)): $esito" }
$nStrumenti = ($esito | ConvertFrom-Json).strumenti
Write-Host "Controllo strumenti: $nStrumenti tipi registrati nell'exe"
if ($SenzaSetup) { Write-Host "Fermo qui (-SenzaSetup): $sorgente"; return }
$dist = Join-Path $radice 'dist'
+3
View File
@@ -0,0 +1,3 @@
# Le DLL scaricate da prendi-gtk.ps1 non si committano: solo lo script (TieMeasureFlow#78).
gtk/
cache/
+149
View File
@@ -0,0 +1,149 @@
# Report PDF nel pacchetto Windows
Questa cartella contiene quello che serve perché i report PDF di TieMeasureFlow
(`src/backend/services/report_service.py`: Jinja2, Plotly con Kaleido 1.2.0, WeasyPrint 68.1)
funzionino su Windows, dentro il pacchetto, **senza Internet**. Riferimento: TieMeasureFlow#78
(parte del pacchetto Windows, madre #76).
## Come funziona
Un report SPC passa da due componenti nativi che su Windows non arrivano con `pip`:
1. **WeasyPrint** disegna il PDF e carica a runtime le DLL di Pango/GTK
(`libgobject-2.0-0`, `libpango-1.0-0`, `libharfbuzz-0`, `libharfbuzz-subset-0`,
`libfontconfig-1`, `libpangoft2-1.0-0` e le loro dipendenze).
2. **Kaleido 1.x** trasforma i grafici Plotly in SVG pilotando un browser Chromium già installato
(non porta più Chromium con sé). Su ogni Windows 10/11 c'è Edge.
### Le DLL: `prendi-gtk.ps1`
```powershell
powershell -NoProfile -ExecutionPolicy Bypass -File packaging\windows\pdf\prendi-gtk.ps1 -Destinazione <cartella>
```
- Scarica da `repo.msys2.org` (ucrt64) 21 pacchetti **pinnati con sha256** (nome file e hash stanno nello
script), verifica l'hash, estrae solo le 27 DLL necessarie, `etc\fonts` di fontconfig e le licenze.
- Idempotente: se la cartella è già pronta (file `.gtk-pin` uguale ai pin dello script) esce subito.
- Se un hash non coincide fallisce e non lascia mezza cartella.
- I pacchetti scaricati stanno in `%LOCALAPPDATA%\Tielogic\tmf-gtk-cache` (`-Cache` per cambiarla).
- Serve Internet **solo alla build**. Estrae `.zst` con `tar.exe` di Windows (verificato: bsdtar 3.8.9 con
zstd); se `tar.exe` non sa di zstd ripiega su `uv run --with zstandard`.
- Gira da utente non admin con PowerShell 5.1.
- ⚠️ `repo.msys2.org` tiene solo le versioni recenti: se un file pinnato sparisce (404) vanno aggiornati nome
e sha256 dal db `https://repo.msys2.org/mingw/ucrt64/ucrt64.db` e rifatto il test. Conservare la cache
(Wasabi o NAS) evita la corsa contro il repository.
Il piano C copia `<cartella>` in `P\gtk\` (P = cartella programma).
### Variabili d'ambiente (le imposta `avvio.py`, piano C)
| Variabile | Valore | Perché |
|---|---|---|
| `WEASYPRINT_DLL_DIRECTORIES` | `<P>\gtk` | WeasyPrint chiama `os.add_dll_directory` su ogni cartella (separatore `;`) **prima** di caricare le DLL. Va impostata prima di importare `weasyprint` (lo fa `report_service` all'import). |
| `BROWSER_PATH` | `msedge.exe` o `chrome.exe` | Choreographer (sotto Kaleido) la legge da sé e ha la precedenza su ogni ricerca (verificato: un percorso inesistente dà `FileNotFoundError`). Senza la variabile Kaleido cerca Edge/Chrome da solo (registro e percorsi tipici). Nessun codice in `report_service.py` per questo. |
Fontconfig non richiede variabili: trova `etc\fonts` accanto alle DLL (verificato). Se manca `etc\fonts`
WeasyPrint stampa "Fontconfig error: Cannot load default config file". I font dei report vengono da
`C:\Windows\Fonts`.
### MathJax: la modifica a `report_service.py`
Kaleido 1.x inserisce nella pagina di render uno `<script>` MathJax da `cdnjs.cloudflare.com` anche quando i
grafici non hanno formule. Misurato su daitarn con Edge: con rete che scarta i pacchetti ogni grafico aspetta
~19 s (due grafici per report SPC); con rete che rifiuta subito ~2 s. `report_service.py` imposta
`pio.defaults.mathjax` su un file `.js` vuoto in `%TEMP%` (`tmf-mathjax-vuoto.js`): nessuna richiesta di rete,
report SPC in ~7 s anche a rete bloccata. Se un giorno i grafici useranno LaTeX, qui serve un MathJax locale.
`plotly.min.js` invece è già locale (dentro il pacchetto `plotly`).
## Flag Nuitka e dati da includere (per `build.ps1`, piano C)
⚠️ Stato: ricavati leggendo i sorgenti installati (versioni del `uv.lock`); **non provati con una build
Nuitka**: la verifica è la prima build di C. I dati e i metadati sono quelli che il codice legge davvero da disco.
Dati dei pacchetti (file non `.py` letti a runtime):
```text
--include-package-data=weasyprint # css/html5_ua.css, html5_ph.css, html5_ua_form.css, pdf/sRGB2014.icc
--include-package-data=pyphen # dictionaries/*.dic (84 file, sillabazione)
--include-package-data=kaleido # vendor/kaleido_scopes.js (pagina di render)
--include-package-data=plotly # package_data/plotly.min.js (Kaleido lo cerca come file accanto a plotly/__init__), templates/*.json, validators/_validators.json
--include-package-data=tzdata # zoneinfo: Windows non ha il database IANA (vedi sotto)
--include-data-dir=src/backend/templates=src/backend/templates # Jinja2: report_service legge i .html dal disco
```
`choreographer/resources/last_known_good_chrome.json` serve solo al comando che scarica Chrome: non serve.
Metadati delle distribuzioni (`importlib.metadata.version` al momento dell'import): senza, `import plotly` e
Kaleido falliscono con `PackageNotFoundError`:
```text
--include-distribution-metadata=plotly
--include-distribution-metadata=kaleido
```
Import dinamici che l'analisi statica di Nuitka può perdere:
```text
--include-package=fontTools # ttLib/ttFont.py importa le tabelle con __import__("fontTools.ttLib.tables." + tag)
--include-package=plotly # _plotly_utils/importers.py: importlib.import_module lazy dei moduli graph_objs
--include-package=_plotly_utils
--include-module=_cffi_backend # WeasyPrint usa cffi in modalita' ABI (dlopen), serve il modulo C di cffi
--include-package=weasyprint
--include-package=kaleido
--include-package=choreographer
```
Altre note per C:
- WeasyPrint salta `os.add_dll_directory` se `sys.frozen` esiste (`weasyprint/text/ffi.py`). Nuitka di norma
non lo imposta (PyInstaller sì): se in `TieMeasureFlow.exe` le DLL non si trovassero, chiamare
`os.add_dll_directory(<P>\gtk)` da `avvio.py` prima dell'import.
- Le DLL **non** vanno compilate né incluse da Nuitka: restano file in `P\gtk\`.
- Il controllo "nessun `.py`" di `build.ps1` non deve scandire `gtk\` (non ci sono `.py`, ma è meglio
saperlo) né `licenze\`.
- **`tzdata` manca da `pyproject.toml`**: su Windows `ZoneInfo("Europe/Rome")` (usato da
`src/backend/utils/tempo.py`, quindi da ogni report) dà `ZoneInfoNotFoundError`. Va aggiunto
`tzdata; sys_platform == 'win32'` agli extra server e client (o al gruppo `windows`) e il frontend
(`src/frontend/flask_app/app.py` usa `ZoneInfo`) ne ha bisogno allo stesso modo.
- `pdfplumber` è già nell'extra `server` (il test lo usa per leggere i PDF).
## Licenze delle DLL
Le DLL sono binari MSYS2/MinGW (ucrt64) non modificati, collegati dinamicamente. `prendi-gtk.ps1` mette in
`<cartella>\licenze\` i testi di licenza di ogni pacchetto più un `<pacchetto>.txt` con nome del file e licenza.
**L'installer deve includere `gtk\licenze\`.**
| Libreria | Licenza |
|---|---|
| Pango, GLib/GObject/GIO, fribidi, libthai, libdatrie, graphite2, libintl (gettext-runtime), libiconv | LGPL 2.1 o successive |
| FreeType | FTL oppure GPL 2+ (si sceglie FTL) |
| HarfBuzz, expat, libffi, brotli, fontconfig | MIT / permissiva |
| libpng, zlib-ng, bzip2, pcre2 | permissive (libpng, Zlib, bzip2, BSD-3) |
| libgcc, libstdc++ | GPL 3 con GCC Runtime Library Exception (ridistribuzione libera) |
| libwinpthread | MIT e BSD-3 |
Per le librerie LGPL, chi riceve il pacchetto può sostituire le DLL (sono file separati in `gtk\`); i sorgenti
sono pubblici su MSYS2 (`https://packages.msys2.org/`), e l'elenco esatto delle versioni è nello script.
## Test
`src/backend/tests/test_report_windows.py` (saltato fuori da Windows, senza DLL, senza Edge/Chrome o senza
`tzdata`): in un sottoprocesso con WeasyPrint **vero** (il `conftest.py` altrimenti lo sostituisce con un
mock) genera su SQLite in memoria il report SPC (carta di controllo e istogramma) e quello a tabella, e
controlla con `pdfplumber` byte `%PDF-`, almeno una pagina, titoli dei grafici e segni vettoriali.
```powershell
# DLL (una volta) e test
powershell -NoProfile -ExecutionPolicy Bypass -File packaging\windows\pdf\prendi-gtk.ps1 -Destinazione packaging\windows\pdf\gtk
uv run --python 3.13 --extra server --extra dev pytest src/backend/tests/test_report_windows.py -q
```
`packaging/windows/pdf/gtk/` e `cache/` sono in `.gitignore`: le DLL non si committano.
### Prova a rete bloccata
`HTTP_PROXY`/`HTTPS_PROXY` non bastano: Edge e Chrome usano il **proxy di sistema** di Windows. Per provare davvero
il PC senza Internet: impostare temporaneamente il proxy di sistema (Impostazioni, Rete, Proxy) su un indirizzo
morto (`127.0.0.1:9` rifiuta subito; `10.255.255.1:8080` scarta i pacchetti, il caso peggiore), generare i
report e **ripristinare** il proxy. Esito del 2026-10-17 con proxy che scarta i pacchetti: report SPC e report a
tabella completi in ~7 s.
+143
View File
@@ -0,0 +1,143 @@
<#
prendi-gtk.ps1 - prepara le DLL di Pango/GTK che WeasyPrint 68.1 vuole su Windows.
COSA FA: scarica da repo.msys2.org (ucrt64) un elenco FISSO di pacchetti, ne verifica lo sha256,
estrae solo le DLL che servono a WeasyPrint (+ etc\fonts di fontconfig e le licenze) in -Destinazione.
COME SI USA: powershell -NoProfile -ExecutionPolicy Bypass -File prendi-gtk.ps1 -Destinazione <cartella>
Opzionali: -Cache <cartella> (pacchetti scaricati, default %LOCALAPPDATA%\Tielogic\tmf-gtk-cache),
-Forza (rifa' anche se la cartella e' gia' pronta).
DA COSA DIPENDE: Windows 10/11 con tar.exe (bsdtar con zstd) oppure uv + Python (modulo zstandard);
Internet solo al momento della build, mai a runtime. PowerShell 5.1 va bene, non serve admin.
IDEMPOTENTE: se la cartella ha gia' le DLL e il file .gtk-pin coincide con questi pin, esce senza scaricare.
FALLISCE se lo sha256 di un pacchetto non coincide (nessun file parziale resta in -Destinazione).
TRAPPOLA: repo.msys2.org tiene solo le versioni recenti. Se un file pinnato sparisce (404) serve aggiornare
i pin qui sotto (nome file + sha256 dal db https://repo.msys2.org/mingw/ucrt64/ucrt64.db) e riprovare i test.
La -Cache conserva i pacchetti gia' scaricati: copiarla su Wasabi/NAS evita il problema.
#>
param(
[Parameter(Mandatory = $true)][string]$Destinazione,
[string]$Cache = (Join-Path $env:LOCALAPPDATA 'Tielogic\tmf-gtk-cache'),
[switch]$Forza
)
$ErrorActionPreference = 'Stop'
# PS 5.1 usa TLS 1.0 di default: repo.msys2.org vuole 1.2 (cosa/perche')
[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12
$Base = 'https://repo.msys2.org/mingw/ucrt64/'
# Pin: pacchetto, file, sha256 (dal db ucrt64 di MSYS2 del 2026-10-17), licenza dichiarata.
$Pacchetti = @(
@{ N='brotli'; F='mingw-w64-ucrt-x86_64-brotli-1.2.0-1-any.pkg.tar.zst'; H='9cc89665496ea504751476eafba15c71c554aaf3004babc2004f7c07ffd60514'; L='MIT' }
@{ N='bzip2'; F='mingw-w64-ucrt-x86_64-bzip2-1.0.8-4-any.pkg.tar.zst'; H='f03a2174034ddd2d96cecd34f617c5f8e2ef86c812b8d2bb3b8875257f2c8bfa'; L='bzip2 (BSD)' }
@{ N='expat'; F='mingw-w64-ucrt-x86_64-expat-2.9.0-1-any.pkg.tar.zst'; H='f52364dc874009316e1ee1fe46ee89784594b80d10cbcc1cc856945d575308e0'; L='MIT' }
@{ N='fontconfig'; F='mingw-w64-ucrt-x86_64-fontconfig-2.18.3-1-any.pkg.tar.zst'; H='d0b364f3ee37e3b6f6c136688ce8fa909bb6d8a69df7e08de1aa40870d9e1d60'; L='fontconfig (MIT-like)' }
@{ N='freetype'; F='mingw-w64-ucrt-x86_64-freetype-2.15.0-1-any.pkg.tar.zst'; H='a9a691f71d1a553099c54ca666e611abfb2b4ff08dae24f2a5af72e22a7e5feb'; L='GPL-2.0+ OR FTL' }
@{ N='fribidi'; F='mingw-w64-ucrt-x86_64-fribidi-1.0.17-1-any.pkg.tar.zst'; H='3581ee03cf83a9bb0e98bf688c4f532a29553500e317532d90746de295b0f8a8'; L='LGPL-2.1+' }
@{ N='gettext-runtime'; F='mingw-w64-ucrt-x86_64-gettext-runtime-1.0-1-any.pkg.tar.zst'; H='ba693dda4ac375af76ce481ff3a6e7481286546cc7dc6d56c7021dae34084157'; L='LGPL-2.1+ (libintl)' }
@{ N='glib2'; F='mingw-w64-ucrt-x86_64-glib2-2.90.1-1-any.pkg.tar.zst'; H='9a7826349279bcca0c935b01880d812be9de9609aef2205be68fd7a6128c1695'; L='LGPL-2.1+' }
@{ N='graphite2'; F='mingw-w64-ucrt-x86_64-graphite2-1.3.15-1-any.pkg.tar.zst'; H='34ca1f88bff10ef182a6b1c8e5622d5029f13b05748e361af13b552b422ea849'; L='LGPL-2.1+' }
@{ N='harfbuzz'; F='mingw-w64-ucrt-x86_64-harfbuzz-14.6.0-1-any.pkg.tar.zst'; H='8781d486c864bf91f482768cc5bd7267e49331636583d9f855af3974e6378e69'; L='MIT' }
@{ N='libdatrie'; F='mingw-w64-ucrt-x86_64-libdatrie-0.2.14-1-any.pkg.tar.zst'; H='ea92777801ad325278193a362967540752ae37035016df2f6bcee2d89928a195'; L='LGPL' }
@{ N='libffi'; F='mingw-w64-ucrt-x86_64-libffi-3.8.0-1-any.pkg.tar.zst'; H='99ad12f4ecfa00a889ef9c5c0188592368a5bac109d0c508ce4e04f29ea95a76'; L='MIT' }
@{ N='libgcc'; F='mingw-w64-ucrt-x86_64-libgcc-16.2.0-6-any.pkg.tar.zst'; H='80b2497f019977dd9528b1abc8621b1328a0115ab1f724047ccbb56bf323f11d'; L='GPL-3.0+ con GCC runtime exception' }
@{ N='libiconv'; F='mingw-w64-ucrt-x86_64-libiconv-1.19-1-any.pkg.tar.zst'; H='9a500f38c2b91808741c62fae746b3e9110b33a1ecf5c30fa0c66dbedddf7e16'; L='LGPL-2.1+' }
@{ N='libpng'; F='mingw-w64-ucrt-x86_64-libpng-1.6.59-1-any.pkg.tar.zst'; H='7f8448e06eceba8f8ce130390addea49f767bdf8f6b7e7eede8ca6d595e3d3a8'; L='libpng (permissiva)' }
@{ N='libstdc++'; F='mingw-w64-ucrt-x86_64-libstdc++-16.2.0-6-any.pkg.tar.zst'; H='33c5e58a85a538cf843a8672991cce591f0231d66da3b66a57629712d64be17f'; L='GPL-3.0+ con GCC runtime exception' }
@{ N='libthai'; F='mingw-w64-ucrt-x86_64-libthai-0.1.30-1-any.pkg.tar.zst'; H='c90ebbca3e7b4c34f4b3ffa501ba8d00cdaf6340cff7db3c46d79d5e01ef207c'; L='LGPL-2.1+' }
@{ N='libwinpthread'; F='mingw-w64-ucrt-x86_64-libwinpthread-14.0.0.r426.g4564ee4b5-2-any.pkg.tar.zst'; H='a02f5e09d666d3be657c78c86fe67c975d04fe33b5211cd5ff1b23a91c118524'; L='MIT AND BSD-3-Clause' }
@{ N='pcre2'; F='mingw-w64-ucrt-x86_64-pcre2-10.49-1-any.pkg.tar.zst'; H='63172179db724ac71e2439a7981134c1bcab8049bdddfd349b32b8f76ed52d1e'; L='BSD-3-Clause' }
@{ N='pango'; F='mingw-w64-ucrt-x86_64-pango-1.58.2-1-any.pkg.tar.zst'; H='8d86e4a018ff139b010bf357fa1c9f9030fa32ee195c54ed7e89d8ea701851c8'; L='LGPL-2.1' }
@{ N='zlib-ng-compat'; F='mingw-w64-ucrt-x86_64-zlib-ng-compat-2.3.3-2-any.pkg.tar.zst'; H='b3ca6b437fb154a0fa26652002cdda591498ee4a0e08d91cc653c3702d090f25'; L='Zlib' }
)
# DLL che WeasyPrint carica (6 radici) piu' la loro chiusura di dipendenze (calcolata con pefile il 2026-10-17).
$Dll = @(
'libbrotlicommon.dll','libbrotlidec.dll','libbz2-1.dll','libdatrie-1.dll','libexpat-1.dll','libffi-8.dll',
'libfontconfig-1.dll','libfreetype-6.dll','libfribidi-0.dll','libgcc_s_seh-1.dll','libgio-2.0-0.dll',
'libglib-2.0-0.dll','libgmodule-2.0-0.dll','libgobject-2.0-0.dll','libgraphite2.dll','libharfbuzz-0.dll',
'libharfbuzz-subset-0.dll','libiconv-2.dll','libintl-8.dll','libpango-1.0-0.dll','libpangoft2-1.0-0.dll',
'libpcre2-8-0.dll','libpng16-16.dll','libstdc++-6.dll','libthai-0.dll','libwinpthread-1.dll','zlib1.dll'
)
function Scrivi([string]$m) { Write-Host ("[prendi-gtk] " + $m) }
# Impronta dei pin: se cambia, la cartella gia' pronta non e' piu' valida.
$pinTesto = (($Pacchetti | ForEach-Object { $_.F + ':' + $_.H }) -join "`n") + "`n" + ($Dll -join ',')
$sha = [Security.Cryptography.SHA256]::Create()
$impronta = -join ($sha.ComputeHash([Text.Encoding]::UTF8.GetBytes($pinTesto)) | ForEach-Object { $_.ToString('x2') })
$marker = Join-Path $Destinazione '.gtk-pin'
$dllMancanti = @($Dll | Where-Object { -not (Test-Path (Join-Path $Destinazione $_)) })
if (-not $Forza -and (Test-Path $marker) -and ((Get-Content $marker -Raw).Trim() -eq $impronta) -and $dllMancanti.Count -eq 0) {
Scrivi "gia' pronto in $Destinazione (pin $($impronta.Substring(0,12)))"
exit 0
}
# --- estrazione .zst: tar.exe di Windows se sa di zstd, altrimenti Python zstandard via uv ---
$tar = Join-Path $env:SystemRoot 'System32\tar.exe'
$tarZstd = $false
if (Test-Path $tar) { $tarZstd = ((& $tar --version) -join ' ') -match 'zstd' }
function Estrai([string]$pkg, [string]$dest) {
if ($tarZstd) {
& $tar -xf $pkg -C $dest
if ($LASTEXITCODE -ne 0) { throw "tar.exe ha fallito su $pkg" }
} else {
$py = "import sys,tarfile,zstandard`nd=zstandard.ZstdDecompressor()`nwith open(sys.argv[1],'rb') as f, d.stream_reader(f) as r:`n tarfile.open(fileobj=r,mode='r|').extractall(sys.argv[2])"
$tmpPy = Join-Path ([IO.Path]::GetTempPath()) ('estrai-zst-' + [guid]::NewGuid().ToString('N') + '.py')
Set-Content -Path $tmpPy -Value $py -Encoding ASCII
try {
& uv run --quiet --with zstandard python -I $tmpPy $pkg $dest
if ($LASTEXITCODE -ne 0) { throw "estrazione zstd via uv fallita su $pkg" }
}
finally { Remove-Item $tmpPy -ErrorAction SilentlyContinue }
}
}
New-Item -ItemType Directory -Force -Path $Cache | Out-Null
# Lavoro in una cartella temporanea, poi sposto: un errore a meta' non lascia mezza cartella in -Destinazione.
$tmp = Join-Path ([IO.Path]::GetTempPath()) ('tmf-gtk-' + [guid]::NewGuid().ToString('N'))
$stage = Join-Path $tmp 'stage'
New-Item -ItemType Directory -Force -Path $stage | Out-Null
try {
foreach ($p in $Pacchetti) {
$file = Join-Path $Cache $p.F
$ok = $false
if (Test-Path $file) {
$ok = ((Get-FileHash $file -Algorithm SHA256).Hash.ToLower() -eq $p.H)
if (-not $ok) { Scrivi "cache corrotta, riscarico $($p.F)"; Remove-Item $file -Force }
}
if (-not $ok) {
Scrivi "scarico $($p.F)"
$parz = $file + '.parz'
# il '+' di libstdc++ va codificato nell'URL
Invoke-WebRequest -UseBasicParsing -Uri ($Base + [Uri]::EscapeDataString($p.F)) -OutFile $parz
$h = (Get-FileHash $parz -Algorithm SHA256).Hash.ToLower()
if ($h -ne $p.H) { Remove-Item $parz -Force; throw "sha256 diverso per $($p.F): atteso $($p.H), trovato $h" }
Move-Item $parz $file -Force
}
$ex = Join-Path $tmp $p.N
New-Item -ItemType Directory -Force -Path $ex | Out-Null
Estrai $file $ex
$bin = Join-Path $ex 'ucrt64\bin'
if (Test-Path $bin) { Get-ChildItem $bin -Filter *.dll | Copy-Item -Destination $stage -Force }
# fontconfig: senza etc\fonts WeasyPrint non trova i font (errore "Cannot load default config file")
if ($p.N -eq 'fontconfig') { Copy-Item (Join-Path $ex 'ucrt64\etc\fonts') (Join-Path $stage 'etc\fonts') -Recurse -Force }
# licenze: le DLL LGPL si ridistribuiscono con il testo della licenza
New-Item -ItemType Directory -Force -Path (Join-Path $stage 'licenze') | Out-Null
$lic = Join-Path $ex 'ucrt64\share\licenses'
if (Test-Path $lic) { Copy-Item $lic (Join-Path $stage ('licenze\' + $p.N)) -Recurse -Force }
Set-Content -Path (Join-Path $stage ('licenze\' + $p.N + '.txt')) -Value ($p.F + "`nlicenza: " + $p.L + "`nsorgente: https://packages.msys2.org/") -Encoding UTF8
}
# Tengo solo le DLL in whitelist: le altre DLL dei pacchetti non servono.
Get-ChildItem $stage -Filter *.dll | Where-Object { $Dll -notcontains $_.Name } | Remove-Item -Force
$mancanti = @($Dll | Where-Object { -not (Test-Path (Join-Path $stage $_)) })
if ($mancanti.Count -gt 0) { throw "DLL attese non trovate: $($mancanti -join ', ')" }
Set-Content -Path (Join-Path $stage '.gtk-pin') -Value $impronta -Encoding ASCII
if (Test-Path $Destinazione) { Get-ChildItem $Destinazione -Force | Remove-Item -Recurse -Force }
New-Item -ItemType Directory -Force -Path $Destinazione | Out-Null
Get-ChildItem $stage -Force | Move-Item -Destination $Destinazione -Force
Scrivi "pronto: $($Dll.Count) DLL in $Destinazione"
}
finally {
Remove-Item $tmp -Recurse -Force -ErrorAction SilentlyContinue
}
+76
View File
@@ -13,6 +13,8 @@ import importlib.util
import json
import os
import sqlite3
import sys
import types
from datetime import date, timedelta
from pathlib import Path
@@ -195,3 +197,77 @@ def test_chiavi_extra_di_config_env_passano_ma_non_sovrascrivono_il_contratto(av
avvio.main(["--senza-finestre", "--senza-browser"])
assert os.environ["STATION_CODE"] == "ST-FIERA" and os.environ["STATION_SWITCH_ENABLED"] == "1"
assert os.environ["APP_ENV"] == "production" # le variabili fisse vincono
# --- TieMeasureFlow#82: strumenti nel pacchetto (nel binario non ci sono gli entry point) ---
def _modulo_strumenti(monkeypatch, moduli):
# finto _strumenti.py generato da build.ps1
finto = types.ModuleType("_strumenti")
finto.MODULI = tuple(moduli)
monkeypatch.setitem(sys.modules, "_strumenti", finto)
def test_registra_strumenti_da_sorgente_senza_modulo_generato(avvio, monkeypatch):
# _strumenti.py esiste solo dopo build.ps1: se manca (sorgente) bastano gli entry point
monkeypatch.setitem(sys.modules, "_strumenti", None) # None in sys.modules = ImportError all'import
from visionsuite.task.graph import registry
assert avvio.registra_strumenti() == []
assert len(registry()) > 0
def test_registra_strumenti_modulo_inesistente_non_solleva(avvio, monkeypatch):
_modulo_strumenti(monkeypatch, ("visionsuite.tools.image", "visionsuite.tools.non_esiste"))
assert avvio.registra_strumenti() == ["visionsuite.tools.non_esiste"] # raccolto, non sollevato
def test_registra_strumenti_importa_i_moduli_elencati(avvio, monkeypatch):
# senza entry point (come nell'exe) i moduli di MODULI sono l'unica via: li si importa per nome
_modulo_strumenti(monkeypatch, ("visionsuite.tools.image",))
visti = []
monkeypatch.setattr(avvio.importlib, "import_module", lambda nome: visti.append(nome))
assert avvio.registra_strumenti() == [] and visti == ["visionsuite.tools.image"]
def test_controlla_strumenti_senza_licenza_scrive_json(avvio, monkeypatch, capsys, tmp_path):
monkeypatch.setitem(sys.modules, "_strumenti", None)
# la licenza non deve essere nemmeno chiesta, e server e migrazioni non devono partire
def mai(*a, **k):
raise AssertionError("licenza, server o migrazioni chiamati da --controlla-strumenti")
monkeypatch.setattr(avvio.lic, "verifica", mai)
monkeypatch.setattr(avvio.lic, "cerca_licenza", mai)
monkeypatch.setattr(avvio, "avvia_server", mai)
monkeypatch.setattr(avvio, "esegui_migrazioni", mai)
assert avvio.main(["--controlla-strumenti", "--senza-finestre"]) == 0
dati = json.loads(capsys.readouterr().out)
assert dati["strumenti"] > 0 and dati["strumenti"] == len(dati["kind"])
assert dati["falliti"] == []
scritto = tmp_path / "Tielogic" / "TieMeasureFlow" / "log" / "controllo-strumenti.json"
assert json.loads(scritto.read_text(encoding="utf-8")) == dati
assert avvio.aperti == [] and not (tmp_path / "Tielogic" / "TieMeasureFlow" / "tmf.db").exists() # nessun DB
def test_controlla_strumenti_exit_3_se_un_modulo_fallisce(avvio, monkeypatch, capsys):
_modulo_strumenti(monkeypatch, ("visionsuite.tools.non_esiste",))
assert avvio.main(["--controlla-strumenti"]) == 3
assert json.loads(capsys.readouterr().out)["falliti"] == ["visionsuite.tools.non_esiste"]
def test_controlla_strumenti_exit_3_se_il_catalogo_e_vuoto(avvio, monkeypatch, capsys):
# niente strumenti registrati (es. entry point assenti e _strumenti vuoto): il controllo deve fallire
import visionsuite.task.graph as graph
monkeypatch.setattr(avvio, "registra_strumenti", lambda: [])
monkeypatch.setattr(graph, "registry", lambda: {})
assert avvio.main(["--controlla-strumenti"]) == 3
assert json.loads(capsys.readouterr().out)["strumenti"] == 0
def test_avvio_normale_registra_gli_strumenti_prima_del_server(avvio, tmp_path, monkeypatch):
# main() chiama registra_strumenti() dopo licenza e migrazioni e prima di avvia_server
_scrivi_licenza(tmp_path)
monkeypatch.setattr(avvio, "registra_strumenti", lambda: avvio.chiamate.append("strumenti") or [])
assert avvio.main(["--senza-finestre", "--senza-browser"]) == 0
dati = tmp_path / "Tielogic" / "TieMeasureFlow"
assert avvio.chiamate == ["migrazioni", "strumenti", ("127.0.0.1", dati)]
+566
View File
@@ -0,0 +1,566 @@
"""Calibrazione a scacchiera dell'acquisitore (TieMeasureFlow, #83).
Cosa fa: da una foto di scacchiera piana calcola la scala (px/mm) con
TieVision-Core (visionsuite.calibration), e conserva un risultato per sorgente
camera in un JSON (scrittura atomica).
#84: `calibra_scacchiera_densa(pixels, square_mm)` calibra da UNA foto di
scacchiera che riempie tutto il campo (ottica telecentrica): trova da sola la
griglia, costruisce la mappa di distorsione (DistortionMap) e il mm/px
dell'immagine raddrizzata. L'Archivio salva la mappa accanto al JSON (.npz) e
`raddrizza()` la applica ai frame.
Uso: importato da scripts/camera_service.py (rotte /camera/calibrate e
/camera/calibration); `calibra_scacchiera(gray, cols, rows, square_mm)` e
`calibra_scacchiera_densa(gray, square_mm)` sono logica pura, senza FastAPI,
e si provano da sole.
Dipende da: numpy, opencv, visionsuite (vs-core). Niente altro (no scipy).
"""
from __future__ import annotations
import base64
import hashlib
import json
import math
import os
import re
import tempfile
import threading
from collections import deque
from datetime import datetime, timezone
from pathlib import Path
from typing import NamedTuple
import cv2
import numpy as np
from visionsuite.calibration import calibrate_planar, find_chessboard
from visionsuite.calibration.distortion_map import DistortionMap # #84 mappa densa
from visionsuite.core.image import Image
class CalibrazioneError(ValueError):
"""Parametri invalidi o scacchiera non trovata: input sbagliato, non un guasto (422)."""
def _gray_u8(pixels: np.ndarray) -> np.ndarray:
"""Immagine -> grigio uint8 2D."""
px = np.asarray(pixels)
if px.ndim == 3:
px = cv2.cvtColor(px, cv2.COLOR_BGRA2GRAY if px.shape[2] == 4 else cv2.COLOR_BGR2GRAY)
if px.dtype != np.uint8:
# float/uint16 -> scala sul massimo, come fa vs-core
m = float(px.max()) if px.size else 1.0
px = np.clip(px.astype(np.float32) / max(m, 1e-9) * 255, 0, 255).astype(np.uint8)
return np.ascontiguousarray(px)
def calibra_scacchiera(pixels, cols: int, rows: int, square_mm: float) -> dict:
"""Scala (px/mm) da una foto di scacchiera piana. cols/rows = angoli INTERNI.
Ritorna {px_per_mm, mm_per_px, rms_px, n_corners, cols, rows, square_mm,
image_width, image_height, scale_spread_pct, metodo, distorsione,
overlay_png (b64)}.
"""
if cols < 2 or rows < 2:
raise CalibrazioneError("cols e rows (angoli interni) devono essere almeno 2")
if not (square_mm > 0 and math.isfinite(square_mm)):
raise CalibrazioneError("square_mm deve essere > 0")
frame = _gray_u8(pixels)
image = Image(frame)
try:
image_pts, world_pts = find_chessboard(image, (cols, rows), square_mm)
except RuntimeError as exc:
# find_chessboard solleva RuntimeError quando la scacchiera non c'e'
raise CalibrazioneError(
f"scacchiera {cols}x{rows} non trovata nell'immagine: {exc}") from exc
model = calibrate_planar(image, image_pts, world_pts) # omografia pura
# Scala al centro della scacchiera (puo' non stare al centro immagine)
x0, y0 = image_pts.min(axis=0)
x1, y1 = image_pts.max(axis=0)
mm_per_px = model.pixel_size((x0 + x1) / 2.0, (y0 + y1) / 2.0)
if not (mm_per_px > 0 and math.isfinite(mm_per_px)):
raise CalibrazioneError("calibrazione degenere: scala non valida")
# Scarto massimo % della scala ai quattro angoli dell'area della scacchiera
# rispetto al centro: ottica telecentrica/piana ~ 0, prospettiva = grande.
spread = max(abs(model.pixel_size(x, y) - mm_per_px) / mm_per_px * 100.0
for x in (x0, x1) for y in (y0, y1))
# Overlay: angoli disegnati sull'immagine (PNG base64)
bgr = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
cv2.drawChessboardCorners(
bgr, (cols, rows), image_pts.reshape(-1, 1, 2).astype(np.float32), True)
ok, png = cv2.imencode(".png", bgr)
if not ok:
raise RuntimeError("PNG dell'overlay non codificabile")
h, w = frame.shape
return {
"px_per_mm": 1.0 / mm_per_px,
"mm_per_px": mm_per_px,
"rms_px": float(model.rms_reprojection),
"n_corners": int(len(image_pts)),
"cols": int(cols),
"rows": int(rows),
"square_mm": float(square_mm),
"image_width": int(w),
"image_height": int(h),
"scale_spread_pct": float(spread),
"metodo": "planare", # #84: il metodo con cols/rows resta quello di prima
"distorsione": False,
"overlay_png": base64.b64encode(png.tobytes()).decode("ascii"),
}
# ---- #84: calibrazione densa (scacchiera che riempie il campo, griglia non nota)
# #84: findChessboardCornersSBWithMeta (verificato su opencv 5.0.0) su una scacchiera
# grande quanto l'immagine trova solo frammenti (2064x1544 con quadrati da 40 px:
# 5x5, 28x6...). Per questo la griglia si cerca qui: sellini della Hessiana a piu'
# scale + crescita del reticolo; poi cornerSubPix.
_SIGME = (1.5, 2.2, 3.2, 4.5, 6.5, 9.0, 13.0) # scale dei sellini (passo ~ 5..120 px)
_MAX_CANDIDATI = 6000
def _candidati(gray: np.ndarray, sigma: float) -> np.ndarray:
"""Massimi locali del 'selliness' (-det Hessiana) = angoli di scacchiera, (n,2) x,y."""
b = cv2.GaussianBlur(gray.astype(np.float32), (0, 0), sigma)
ixx = cv2.Sobel(b, cv2.CV_32F, 2, 0, ksize=3)
iyy = cv2.Sobel(b, cv2.CV_32F, 0, 2, ksize=3)
ixy = cv2.Sobel(b, cv2.CV_32F, 1, 1, ksize=3)
r = ixy * ixy - ixx * iyy # >0 solo dove l'immagine ha un sellino
massimo = float(r.max())
if massimo <= 0:
return np.zeros((0, 2))
k = 2 * int(math.ceil(1.5 * sigma)) + 1
picchi = (r >= cv2.dilate(r, np.ones((k, k), np.uint8))) & (r > 0.1 * massimo)
ys, xs = np.nonzero(picchi)
if len(xs) > _MAX_CANDIDATI: # tiene i piu' forti
forti = np.argsort(r[ys, xs])[::-1][:_MAX_CANDIDATI]
ys, xs = ys[forti], xs[forti]
return np.stack([xs, ys], axis=1).astype(np.float64)
def _e_angolo_x(lisc: np.ndarray, delta: float, p: np.ndarray, u: np.ndarray, v: np.ndarray) -> bool:
"""True se attorno a p i quattro quadranti alternano chiaro/scuro (vero angolo X).
Scarta le giunzioni a L/T al bordo della scacchiera (3 quadranti uguali), che
il selliness da solo accetta e che spostano la griglia.
"""
h, w = lisc.shape
campioni = []
for q in (0.25 * (u + v), -0.25 * (u + v), 0.25 * (u - v), -0.25 * (u - v)):
x, y = int(round(p[0] + q[0])), int(round(p[1] + q[1]))
if not (0 <= x < w and 0 <= y < h):
return False
campioni.append(float(lisc[y, x]))
s1, s2, t1, t2 = campioni
return (min(s1, s2) > max(t1, t2) + delta) or (max(s1, s2) < min(t1, t2) - delta)
def _cresci_reticolo(pts: np.ndarray, centro: tuple[float, float],
lisc: np.ndarray, delta: float) -> dict | None:
"""Assegna un indice (riga, colonna) ai candidati che stanno su un reticolo.
Parte dal candidato piu' vicino al centro, ricava i due vettori di passo dai
vicini e cresce a macchia d'olio: la base si aggiorna a ogni passo, quindi
segue una distorsione lenta. Ritorna {(i, j): indice candidato} o None.
"""
if len(pts) < 9:
return None
seme = int(np.argmin(np.hypot(pts[:, 0] - centro[0], pts[:, 1] - centro[1])))
d = np.hypot(pts[:, 0] - pts[seme, 0], pts[:, 1] - pts[seme, 1])
vicini = np.argsort(d)[1:9]
u = pts[vicini[0]] - pts[seme]
lu = float(np.hypot(*u))
v = None
for k in vicini[1:]:
w = pts[k] - pts[seme]
lw = float(np.hypot(*w))
# il secondo vettore: quasi ortogonale al primo e di lunghezza simile
if lu > 0 and lw > 0 and abs(float(u @ w)) / (lu * lw) < 0.4 and lw < 1.5 * lu:
v = w
break
if v is None or lu < 3:
return None
posto = {(0, 0): seme}
indice = {seme: (0, 0)}
basi = {seme: (u, v)}
coda = deque([seme])
while coda:
a = coda.popleft()
i, j = indice[a]
u, v = basi[a]
tol = 0.3 * min(float(np.hypot(*u)), float(np.hypot(*v)))
for di, dj, vec in ((0, 1, u), (0, -1, -u), (1, 0, v), (-1, 0, -v)):
if (i + di, j + dj) in posto:
continue
prev = pts[a] + vec
dist = np.hypot(pts[:, 0] - prev[0], pts[:, 1] - prev[1])
b = int(np.argmin(dist))
if dist[b] > tol or b in indice:
continue
if not _e_angolo_x(lisc, delta, pts[b], u, v): # niente L/T del bordo
continue
posto[(i + di, j + dj)] = b
indice[b] = (i + di, j + dj)
reale = pts[b] - pts[a]
basi[b] = (reale * dj, v) if dj else (u, reale * di) # base locale aggiornata
coda.append(b)
return posto
def _rettangolo_pieno(posto: dict) -> tuple[int, int, int, int] | None:
"""Il rettangolo di indici tutto occupato di area massima: (i0, j0, righe, colonne)."""
if not posto:
return None
i_min = min(i for i, _ in posto)
j_min = min(j for _, j in posto)
nr = max(i for i, _ in posto) - i_min + 1
nc = max(j for _, j in posto) - j_min + 1
m = np.zeros((nr, nc), bool)
for (i, j) in posto:
m[i - i_min, j - j_min] = True
migliore = (0, 0, 0, 0)
alt = np.zeros(nc, int)
for r in range(nr): # istogramma delle altezze + pila, O(righe*colonne)
alt = np.where(m[r], alt + 1, 0)
pila: list[int] = []
for c in range(nc + 1):
h = int(alt[c]) if c < nc else 0
while pila and alt[pila[-1]] >= h:
altezza = int(alt[pila.pop()])
sx = pila[-1] + 1 if pila else 0
larg = c - sx
if altezza * larg > migliore[2] * migliore[3]:
migliore = (r - altezza + 1 + i_min, sx + j_min, altezza, larg)
pila.append(c)
return migliore if migliore[2] * migliore[3] else None
def _orienta(g: np.ndarray) -> np.ndarray:
"""Riordina (nr,nc,2) con colonne verso destra e righe verso il basso."""
a = (g[:, -1] - g[:, 0]).mean(axis=0) # direzione dell'indice di colonna
b = (g[-1, :] - g[0, :]).mean(axis=0) # direzione dell'indice di riga
if abs(b[0]) > abs(a[0]): # assi scambiati (scacchiera ruotata di ~90 gradi)
g = g.transpose(1, 0, 2)
a, b = b, a
if a[0] < 0:
g = g[:, ::-1]
if b[1] < 0:
g = g[::-1]
return np.ascontiguousarray(g)
def _passo_mediano(g: np.ndarray) -> float:
"""Distanza mediana (px) fra corner adiacenti."""
passi = np.concatenate([np.diff(g, axis=1).reshape(-1, 2), np.diff(g, axis=0).reshape(-1, 2)])
return float(np.median(np.hypot(passi[:, 0], passi[:, 1])))
def _raffina(img: np.ndarray, pts: np.ndarray, passo: float) -> np.ndarray:
"""cornerSubPix su (n,2) con finestra ~0.3 passo, ristretta per chi sta vicino al bordo.
Una finestra che esce dall'immagine vede il bordo replicato e sposta il corner
di ~1 px (visto sull'ultima colonna di una scacchiera a 14 px dal bordo): per
questi punti si usa la finestra piu' grande che sta dentro; sotto 3 px resta
la posizione grezza.
"""
h, w = img.shape
out = pts.astype(np.float32).copy()
win = int(min(20, max(3, 0.3 * passo)))
dentro = np.minimum.reduce([out[:, 0], w - 1 - out[:, 0], out[:, 1], h - 1 - out[:, 1]])
per_punto = np.minimum(win, np.floor(dentro).astype(int) - 2)
crit = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_COUNT, 40, 0.001)
for wi in np.unique(per_punto[per_punto >= 3]):
sel = per_punto == wi
out[sel] = cv2.cornerSubPix(img, out[sel].reshape(-1, 1, 2).copy(),
(int(wi), int(wi)), (-1, -1), crit).reshape(-1, 2)
return out
def _trova_griglia(frame: np.ndarray) -> np.ndarray:
"""Griglia di angoli (n_righe, n_colonne, 2) x,y, in ordine di riga dall'alto."""
h, w = frame.shape
migliore = None
lisc = cv2.GaussianBlur(frame.astype(np.float32), (0, 0), 1.0)
p5, p95 = np.percentile(frame, (5, 95))
delta = 0.25 * float(p95 - p5) # contrasto minimo fra quadranti opposti
for sigma in _SIGME:
pts = _candidati(frame, sigma)
posto = _cresci_reticolo(pts, (w / 2.0, h / 2.0), lisc, delta)
rett = _rettangolo_pieno(posto) if posto else None
if rett is None:
continue
i0, j0, nr, nc = rett
if nr < 3 or nc < 3:
continue
if migliore is not None and nr * nc < 0.5 * migliore[0] * migliore[1]:
break # oltre il picco: le scale piu' grosse fondono gli angoli
if migliore is None or nr * nc > migliore[0] * migliore[1]:
g = np.array([[pts[posto[(i0 + i, j0 + j)]] for j in range(nc)]
for i in range(nr)])
migliore = (nr, nc, g)
if migliore is None:
raise CalibrazioneError(
"griglia della scacchiera non trovata: serve una scacchiera nitida "
"che riempia il campo")
g = _orienta(migliore[2])
nr, nc, _ = g.shape
passo = _passo_mediano(g)
# sub-pixel: finestra proporzionale al passo, abbastanza piccola da non vedere i vicini
grezzi = g.reshape(-1, 2).astype(np.float32)
raff = _raffina(frame, grezzi, passo)
spostati = np.hypot(*(raff - grezzi).T) > 0.3 * passo
raff[spostati] = grezzi[spostati] # un raffinamento che scappa e' peggio del grezzo
return raff.reshape(nr, nc, 2).astype(np.float64)
def _estrapola(g: np.ndarray, k: int, prima: bool) -> np.ndarray:
"""Aggiunge k colonne (asse 1) per estrapolazione lineare, prima o dopo."""
if k <= 0:
return g
if prima:
passo = g[:, 0] - g[:, 1]
extra = np.stack([g[:, 0] + m * passo for m in range(k, 0, -1)], axis=1)
return np.concatenate([extra, g], axis=1)
passo = g[:, -1] - g[:, -2]
extra = np.stack([g[:, -1] + m * passo for m in range(1, k + 1)], axis=1)
return np.concatenate([g, extra], axis=1)
def _estendi_asse(g: np.ndarray, lungh: int, c: int, passo: float,
grid_step: int) -> tuple[np.ndarray, int]:
"""Estende la griglia (asse 1) fino a coprire l'immagine; ritorna (griglia, k_prima).
`c` e' la coordinata (0 = x, 1 = y) e `lungh` la dimensione dell'immagine su
quell'asse. `from_grid` (crop_domain=False) tiene costante la mappa fuori
dalla griglia: senza anelli in piu' i bordi dell'immagine raddrizzata
sarebbero sbavati.
"""
margine = max(0.0, float(g[:, 0, c].min()))
k_pre = int(math.ceil(margine / passo)) + 1
g = _estrapola(g, k_pre, True)
offset = int(g[0, 0, c]) # lo stesso int() che usa from_grid
# colonne totali perche' l'ultimo nodo, a passo grid_step, superi il bordo
servono = int(math.ceil((lungh - 1 - offset) / grid_step)) + 2
g = _estrapola(g, max(0, servono - g.shape[1]), False)
return g, k_pre
class MappaPronta(NamedTuple):
"""Mappa gia' convertita in virgola fissa per un cv2.remap veloce (#84)."""
map1: np.ndarray
map2: np.ndarray
pixel_size: float
shape: tuple # (altezza, larghezza) dei frame per cui vale
def _prepara_mappa(mappa: DistortionMap) -> MappaPronta:
# CV_16SC2: remap bilineare in virgola fissa, piu' veloce di float32 su 5 MP
m1, m2 = cv2.convertMaps(mappa.map_col, mappa.map_row, cv2.CV_16SC2)
return MappaPronta(m1, m2, float(mappa.pixel_size), mappa.shape)
def raddrizza(frame: np.ndarray, mappa: MappaPronta) -> np.ndarray | None:
"""Frame raddrizzato, o None se le dimensioni non sono quelle della calibrazione."""
if frame.shape[:2] != tuple(mappa.shape):
return None
return cv2.remap(frame, mappa.map1, mappa.map2, cv2.INTER_LINEAR,
borderMode=cv2.BORDER_CONSTANT, borderValue=0)
def calibra_scacchiera_densa(pixels, square_mm: float) -> tuple[dict, DistortionMap]:
"""Calibra da UNA foto di scacchiera densa (riempie il campo). Input: solo square_mm.
Ritorna (risultato, mappa). Risultato: {px_per_mm, mm_per_px (dell'immagine
RADDRIZZATA), rms_px, residuo_rms_px, residuo_max_px, n_corners, cols, rows,
square_mm, image_width, image_height, scale_spread_pct (0: la mappa uniforma
la scala), copertura_pct, metodo, distorsione, overlay_png (b64)}.
"""
if not (square_mm > 0 and math.isfinite(square_mm)):
raise CalibrazioneError("square_mm deve essere > 0")
frame = _gray_u8(pixels)
h, w = frame.shape
g = _trova_griglia(frame)
nr, nc, _ = g.shape
passo = _passo_mediano(g)
grid_step = int(round(passo))
if grid_step < 2:
raise CalibrazioneError("calibrazione degenere: passo della griglia troppo piccolo")
# anelli in piu' ai bordi (x e poi y), poi la mappa dai punti estesi
e, kl = _estendi_asse(g, w, 0, passo, grid_step)
et, kt = _estendi_asse(e.transpose(1, 0, 2), h, 1, passo, grid_step)
e = et.transpose(1, 0, 2)
try:
mappa = DistortionMap.from_grid(
e[..., 0].ravel(), e[..., 1].ravel(), e.shape[1], e.shape[0],
image_size=(w, h), grid_step=grid_step, crop_domain=False,
grid_step_measure=square_mm)
except ValueError as exc:
raise CalibrazioneError(f"mappa di distorsione non costruibile: {exc}") from exc
mm_per_px = float(mappa.pixel_size)
if not (mm_per_px > 0 and math.isfinite(mm_per_px)):
raise CalibrazioneError("calibrazione degenere: scala non valida")
# qualita': si raddrizza la stessa foto e si rilegge ogni corner nel punto ideale
off_c, off_r = int(e[0, 0, 0]), int(e[0, 0, 1])
ideali = np.array([[(off_c + (kl + j) * grid_step, off_r + (kt + i) * grid_step)
for j in range(nc)] for i in range(nr)], np.float32).reshape(-1, 1, 2)
rett = cv2.remap(frame, mappa.map_col, mappa.map_row, cv2.INTER_LINEAR,
borderMode=cv2.BORDER_CONSTANT, borderValue=0)
# solo i corner ideali dentro l'immagine raddrizzata (cornerSubPix rifiuta quelli fuori)
xy = ideali.reshape(-1, 2)
dentro = (xy[:, 0] >= 1) & (xy[:, 0] <= w - 2) & (xy[:, 1] >= 1) & (xy[:, 1] <= h - 2)
if int(dentro.sum()) < 4:
raise CalibrazioneError("griglia raddrizzata fuori dall'immagine: calibrazione non verificabile")
ideali = ideali[dentro].reshape(-1, 2)
letti = _raffina(rett, ideali, grid_step)
scarto = (letti - ideali).astype(np.float64)
scarto -= np.median(scarto, axis=0) # int() dell'offset sposta tutto di una traslazione
dist = np.hypot(scarto[:, 0], scarto[:, 1])
residuo_max = float(dist.max())
residuo_rms = float(math.sqrt(float((dist ** 2).mean())))
# copertura: area del quadrilatero dei quattro angoli esterni / area immagine
quad = np.array([g[0, 0], g[0, -1], g[-1, -1], g[-1, 0]], np.float32)
copertura = float(cv2.contourArea(quad)) / float(w * h) * 100.0
# overlay: corner trovati sull'originale (ridotto a max 1600 px: 5 MP in PNG pesano)
f = min(1.0, 1600.0 / w)
bgr = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
pts = g.reshape(-1, 1, 2).astype(np.float32)
if f < 1.0:
bgr = cv2.resize(bgr, None, fx=f, fy=f, interpolation=cv2.INTER_AREA)
pts = pts * np.float32(f)
cv2.drawChessboardCorners(bgr, (nc, nr), pts, True)
ok, png = cv2.imencode(".png", bgr)
if not ok:
raise RuntimeError("PNG dell'overlay non codificabile")
return {
"px_per_mm": 1.0 / mm_per_px,
"mm_per_px": mm_per_px,
"rms_px": residuo_rms,
"residuo_rms_px": residuo_rms,
"residuo_max_px": residuo_max,
"n_corners": int(nr * nc),
"cols": int(nc),
"rows": int(nr),
"square_mm": float(square_mm),
"image_width": int(w),
"image_height": int(h),
"scale_spread_pct": 0.0,
"copertura_pct": copertura,
"metodo": "densa",
"distorsione": True,
"overlay_png": base64.b64encode(png.tobytes()).decode("ascii"),
}, mappa
def ora_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
class Archivio:
"""Un JSON per sorgente in `cartella`; nome file dalla sorgente ripulito.
#84: se la calibrazione e' densa, accanto al JSON sta la mappa (stesso nome,
.npz); `leggi_mappa` la tiene in memoria finche' il file non cambia.
"""
def __init__(self, cartella: Path | str):
self.cartella = Path(cartella)
self._cache: dict = {} # percorso .npz -> (mtime_ns, MappaPronta)
self._lock = threading.Lock()
def _file(self, sorgente: str) -> Path:
# caratteri sicuri + hash dell'originale: due sorgenti diverse non si pestano
pulito = re.sub(r"[^A-Za-z0-9._-]+", "_", sorgente).strip("._") or "sorgente"
sigla = hashlib.sha1(sorgente.encode("utf-8")).hexdigest()[:8]
return self.cartella / f"{pulito[:60]}-{sigla}.json"
def _file_mappa(self, sorgente: str) -> Path:
return self._file(sorgente).with_suffix(".npz") # #84 stesso nome del JSON
def leggi(self, sorgente: str) -> dict | None:
try:
return json.loads(self._file(sorgente).read_text(encoding="utf-8"))
except (OSError, ValueError):
return None # assente o illeggibile = non calibrata
def salva(self, sorgente: str, dati: dict, mappa: DistortionMap | None = None) -> None:
destinazione = self._file(sorgente)
self.cartella.mkdir(parents=True, exist_ok=True)
# #84: prima la mappa (o la rimozione di una vecchia), poi il JSON che la dichiara
if mappa is not None:
self._salva_mappa(sorgente, mappa)
else:
self._togli_mappa(sorgente) # ricalibrazione planare: la vecchia mappa non vale piu'
# scrittura atomica: file temporaneo nella stessa cartella, poi replace
fd, tmp = tempfile.mkstemp(dir=self.cartella, suffix=".tmp")
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
json.dump(dati, f, ensure_ascii=False, indent=2)
os.replace(tmp, destinazione)
except BaseException:
try:
os.unlink(tmp)
except OSError:
pass
raise
def _salva_mappa(self, sorgente: str, mappa: DistortionMap) -> None:
destinazione = self._file_mappa(sorgente)
fd, tmp = tempfile.mkstemp(dir=self.cartella, suffix=".tmp.npz") # .npz: numpy non aggiunge nulla
os.close(fd)
try:
mappa.save(tmp)
os.replace(tmp, destinazione) # atomica come il JSON
except BaseException:
try:
os.unlink(tmp)
except OSError:
pass
raise
with self._lock:
self._cache.pop(str(destinazione), None)
def _togli_mappa(self, sorgente: str) -> None:
destinazione = self._file_mappa(sorgente)
try:
destinazione.unlink()
except FileNotFoundError:
pass
with self._lock:
self._cache.pop(str(destinazione), None)
def leggi_mappa(self, sorgente: str) -> MappaPronta | None:
"""La mappa salvata, pronta per `raddrizza`; None se la sorgente non ne ha una."""
percorso = self._file_mappa(sorgente)
chiave = str(percorso)
try:
mtime = percorso.stat().st_mtime_ns
except OSError:
with self._lock:
self._cache.pop(chiave, None)
return None
with self._lock:
voce = self._cache.get(chiave)
if voce is not None and voce[0] == mtime:
return voce[1]
try:
pronta = _prepara_mappa(DistortionMap.load(percorso))
except (OSError, ValueError, KeyError):
return None # illeggibile = nessuna correzione
with self._lock:
self._cache[chiave] = (mtime, pronta)
return pronta
def elimina(self, sorgente: str) -> None:
try:
self._file(sorgente).unlink()
except FileNotFoundError:
pass
self._togli_mappa(sorgente) # #84 toglie anche la mappa
+231
View File
@@ -0,0 +1,231 @@
"""Pagina HTML della calibrazione a scacchiera (servita da camera_service su /calibrazione, #83).
Cosa fa: una pagina sola, autonoma e offline (niente CDN, JS e CSS inline), in
italiano. Parla con le rotte dello stesso servizio (/camera/sources,
/camera/calibration, /camera/calibrate).
#84: calibrazione densa, un solo dato dell'operatore (lato dello scacco in mm):
scacchiera che riempie il campo, griglia trovata da sola, mappa di distorsione.
Dipende da: niente. E' una stringa, cosi' il pacchetto compilato non ha file dati.
"""
PAGINA = """<!doctype html>
<html lang="it">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Calibrazione camera</title>
<style>
body { font-family: system-ui, sans-serif; margin: 0; background: #f3f4f6; color: #1f2937; }
main { max-width: 960px; margin: 0 auto; padding: 16px; }
h1 { font-size: 1.4rem; margin: 0 0 12px; }
.scheda { background: #fff; border: 1px solid #d1d5db; border-radius: 8px; padding: 14px; margin-bottom: 14px; }
label { display: block; font-size: .85rem; margin-bottom: 4px; color: #4b5563; }
select, input[type=number] { padding: 6px 8px; font-size: 1rem; border: 1px solid #9ca3af; border-radius: 6px; }
.riga { display: flex; flex-wrap: wrap; gap: 12px; align-items: flex-end; }
button { padding: 8px 14px; font-size: 1rem; border: 0; border-radius: 6px; background: #2563eb; color: #fff; cursor: pointer; }
button.secondario { background: #6b7280; }
button.pericolo { background: #b91c1c; }
button:disabled { opacity: .5; cursor: default; }
.messaggio { margin-top: 10px; font-weight: 600; }
.errore { color: #b91c1c; }
.ok { color: #15803d; }
table { border-collapse: collapse; }
td { padding: 3px 14px 3px 0; }
td:first-child { color: #4b5563; }
img { max-width: 100%; border: 1px solid #d1d5db; border-radius: 6px; margin-top: 8px; }
.avviso { color: #92400e; }
</style>
</head>
<body>
<main>
<h1>Calibrazione camera con scacchiera</h1>
<div class="scheda">
<div class="riga">
<div><label for="sorgente">Sorgente</label><select id="sorgente"></select></div>
<div><label for="quadrato">Lato dello scacco (mm)</label><input id="quadrato" type="number" min="0.001" step="any" value="2"></div>
</div>
<p style="font-size:.85rem;color:#4b5563;margin:10px 0 0">
Servono una scacchiera che riempie tutto il campo della camera, piana, ferma e ben illuminata, e il lato di uno scacco.
La griglia si trova da sola; il risultato corregge anche la distorsione dell'ottica.</p>
</div>
<div class="scheda">
<div class="riga">
<button id="acquisisci">Acquisisci e calibra</button>
<div><label for="file">oppure un file immagine</label><input id="file" type="file" accept="image/*"></div>
<button id="carica" class="secondario">Carica immagine e calibra</button>
</div>
<div id="messaggio" class="messaggio"></div>
</div>
<div class="scheda" id="risultato" hidden>
<strong>Risultato del calcolo</strong>
<table id="tabella"></table>
<div id="avviso" class="avviso"></div>
<div class="riga" style="margin-top:10px">
<button id="salva">Salva</button>
<span style="font-size:.85rem;color:#4b5563">Da camera, salvando si acquisisce e si calcola di nuovo: tenere la scacchiera ferma.</span>
</div>
<img id="overlay" alt="Angoli trovati sulla scacchiera">
</div>
<div class="scheda">
<strong>Calibrazione salvata</strong>
<div id="stato">-</div>
<div class="riga" style="margin-top:10px">
<button id="elimina" class="pericolo" disabled>Elimina</button>
</div>
</div>
</main>
<script>
"use strict";
var $ = function (id) { return document.getElementById(id); };
var ultimo = null; // ultimo calcolo, per «Salva»
function messaggio(testo, classe) {
var m = $("messaggio");
m.textContent = testo || "";
m.className = "messaggio " + (classe || "");
}
function numero(v, cifre) { return Number(v).toFixed(cifre); }
function righe(d) {
return [
["Scala", numero(d.px_per_mm, 4) + " px/mm (" + numero(d.mm_per_px, 5) + " mm/px)"],
["Metodo", d.metodo !== "densa" ? "planare (senza correzione della distorsione)" : "densa (immagine raddrizzata)"],
["Griglia trovata", d.cols + " x " + d.rows + " angoli (" + d.n_corners + ")"]
].concat(d.metodo !== "densa" ? [
["Errore di riproiezione (RMS)", numero(d.rms_px, 3) + " px"],
["Variazione di scala nell'area", numero(d.scale_spread_pct, 2) + " %"]
] : [
["Residuo massimo", numero(d.residuo_max_px, 3) + " px"],
["Residuo RMS", numero(d.residuo_rms_px, 3) + " px"],
["Copertura del campo", numero(d.copertura_pct, 1) + " %"]
]).concat([
["Lato dello scacco", d.square_mm + " mm"],
["Immagine", d.image_width + " x " + d.image_height + " px"]
]);
}
function tabella(el, d) {
el.textContent = "";
righe(d).forEach(function (r) {
var tr = document.createElement("tr");
r.forEach(function (c) {
var td = document.createElement("td");
td.textContent = c;
tr.appendChild(td);
});
el.appendChild(tr);
});
}
async function dettaglio(risposta) {
try { var j = await risposta.json(); return j.detail || JSON.stringify(j); }
catch (e) { return "errore " + risposta.status; }
}
async function caricaSorgenti() {
var sel = $("sorgente");
try {
var r = await fetch("/camera/sources");
var j = await r.json();
(j.sources || []).forEach(function (s) {
var o = document.createElement("option");
o.value = s.name;
o.textContent = s.label || s.name;
sel.appendChild(o);
});
} catch (e) { messaggio("Elenco delle sorgenti non disponibile.", "errore"); }
if (!sel.options.length) messaggio("Nessuna sorgente camera trovata.", "errore");
aggiornaStato();
}
async function aggiornaStato() {
var sorgente = $("sorgente").value;
var stato = $("stato");
$("elimina").disabled = true;
if (!sorgente) { stato.textContent = "-"; return; }
var r = await fetch("/camera/calibration?source=" + encodeURIComponent(sorgente));
if (r.status === 404) { stato.textContent = "Sorgente non calibrata."; return; }
if (!r.ok) { stato.textContent = await dettaglio(r); return; }
var d = await r.json();
stato.textContent = "";
var t = document.createElement("table");
tabella(t, d);
var quando = document.createElement("div");
quando.textContent = "Calibrata il " + d.calibrated_at;
stato.appendChild(t);
stato.appendChild(quando);
$("elimina").disabled = false;
}
function parametri() {
return { quadrato: $("quadrato").value };
}
// avvisi sulla qualita' (solo calibrazione densa): copertura bassa o residuo alto
function avvisi(d) {
var a = [];
if (d.metodo !== "densa") {
if (d.scale_spread_pct > 1) a.push("Attenzione: la scala varia di " + numero(d.scale_spread_pct, 2) + " % nell'area: scacchiera inclinata o ottica non telecentrica.");
} else {
if (d.copertura_pct < 70) a.push("Attenzione: la scacchiera copre solo il " + numero(d.copertura_pct, 1) + " % del campo (meno del 70 %): la correzione ai bordi non e' misurata.");
if (d.residuo_max_px > 0.5) a.push("Attenzione: residuo massimo " + numero(d.residuo_max_px, 3) + " px (oltre 0,5 px): scacchiera non piana, sfocata o con scacchi irregolari.");
}
return a.join(" ");
}
async function calcola(conFile, salva) {
var sorgente = $("sorgente").value;
if (!sorgente) { messaggio("Scegliere una sorgente.", "errore"); return; }
var p = parametri();
var f = new FormData();
f.append("source", sorgente);
f.append("square_mm", p.quadrato);
f.append("save", salva ? "true" : "false");
if (conFile) {
var file = $("file").files[0];
if (!file) { messaggio("Scegliere un file immagine.", "errore"); return; }
f.append("image", file);
}
messaggio("Calcolo in corso...", "");
["acquisisci", "carica", "salva"].forEach(function (i) { $(i).disabled = true; });
try {
var r = await fetch("/camera/calibrate", { method: "POST", body: f });
if (!r.ok) { messaggio(await dettaglio(r), "errore"); return; }
var d = await r.json();
ultimo = { conFile: conFile, dati: d };
tabella($("tabella"), d);
$("overlay").src = "data:image/png;base64," + d.overlay_png;
$("avviso").textContent = avvisi(d);
$("risultato").hidden = false;
messaggio(d.saved ? "Calibrazione salvata." : "Calibrazione riuscita: controllare e salvare.", "ok");
if (d.saved) aggiornaStato();
} catch (e) {
messaggio("Servizio non raggiungibile.", "errore");
} finally {
["acquisisci", "carica", "salva"].forEach(function (i) { $(i).disabled = false; });
}
}
$("acquisisci").onclick = function () { calcola(false, false); };
$("carica").onclick = function () { calcola(true, false); };
// «Salva» rifa il calcolo con save=true sulla stessa origine (altra acquisizione se da camera)
$("salva").onclick = function () { calcola(ultimo ? ultimo.conFile : false, true); };
$("elimina").onclick = async function () {
var sorgente = $("sorgente").value;
if (!confirm("Eliminare la calibrazione salvata di " + sorgente + "?")) return;
var r = await fetch("/camera/calibration?source=" + encodeURIComponent(sorgente), { method: "DELETE" });
messaggio(r.ok ? "Calibrazione eliminata." : await dettaglio(r), r.ok ? "ok" : "errore");
aggiornaStato();
};
$("sorgente").onchange = aggiornaStato;
caricaSorgenti();
</script>
</body>
</html>
"""
+143 -23
View File
@@ -4,19 +4,44 @@ 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 TieVision-Studio, src/visionsuite/gui/server.py):
GET /camera/sources -> {"sources": [{name, kind, label}]}
POST /camera/grab json {source, trigger="continuo", timeout_ms=5000} -> PNG
POST /camera/grab json {source, trigger="continuo", timeout_ms=5000,
grezzo=false} -> PNG
#84: se la sorgente ha una mappa di distorsione salvata il frame esce
raddrizzato (header X-Raddrizzato: si|no); "grezzo": true lo dà com'è.
Errori: 400 parametri, 404 sorgente inesistente, 503 guasto camera, 504 timeout.
Calibrazione a scacchiera per sorgente (#83, logica in scripts/calibrazione.py):
GET /camera/calibration?source=<src> -> 200 {source, px_per_mm, mm_per_px,
rms_px, n_corners, cols, rows, square_mm, image_width, image_height,
scale_spread_pct, calibrated_at} | 404 "sorgente non calibrata"
POST /camera/calibrate multipart: source, square_mm, image (file, opzionale:
senza, grab GREZZO dalla sorgente), save=false, cols/rows (opzionali).
#84: SENZA cols/rows = calibrazione densa (default): una foto di
scacchiera che riempie il campo, griglia trovata da sola, mappa di
distorsione salvata accanto al JSON (.npz); mm/px e px/mm sono quelli
dell'immagine RADDRIZZATA. CON cols/rows (angoli interni) = metodo
planare di prima. Campi: quelli di C1 + overlay_png (b64) + saved +
metodo ("densa"|"planare"), distorsione (bool) e, per la densa,
residuo_rms_px, residuo_max_px, copertura_pct;
422 scacchiera non trovata o parametri invalidi
DELETE /camera/calibration?source=<src> -> 204 (toglie JSON e mappa)
GET /calibrazione -> pagina HTML autonoma (offline) per fare tutto a mano
Il risultato salvato sta in un JSON per sorgente in --config-dir (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
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>]
[--host 127.0.0.1] [--port 8765] [--folder <cartella immagini>] \
[--config-dir <cartella calibrazioni>]
--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.
vs-camera, fastapi, uvicorn, opencv, python-multipart (rotta /camera/calibrate),
visionsuite.calibration (vs-core).
AVVERTENZA: nessuna autenticazione. Un --host diverso dal loopback espone la
camera a chiunque raggiunga la porta: solo su rete isolata.
@@ -24,17 +49,26 @@ camera a chiunque raggiunga la porta: solo su rete isolata.
from __future__ import annotations
import argparse
import os
import sys
import threading
from contextlib import asynccontextmanager
from pathlib import Path
import cv2
from fastapi import Body, FastAPI, HTTPException
from fastapi.responses import Response
import numpy as np
from fastapi import Body, FastAPI, File, Form, HTTPException, UploadFile
from fastapi.concurrency import run_in_threadpool # #83: il grab bloccante fuori dall'event loop
from fastapi.responses import HTMLResponse, Response
from visionsuite.camera.camera_interface import CameraError, CameraGrabError
from visionsuite.camera.sources import SourceNotFound, list_sources, open_source
# #83: calibrazione e pagina stanno accanto a questo file (che non e' un pacchetto)
sys.path.insert(0, str(Path(__file__).resolve().parent))
import calibrazione # noqa: E402
from calibrazione_pagina import PAGINA # noqa: E402
_TRIGGER = ("continuo",)
# errori noti di vs-camera (inglese) -> italiano; esatti, poi per prefisso
@@ -61,7 +95,18 @@ def _traduci(messaggio: str) -> str:
return messaggio
def create_app(folder: Path | str | None = None) -> FastAPI:
def _config_dir_default() -> Path:
"""#83: cartella delle calibrazioni salvate."""
if os.name == "nt":
base = os.environ.get("ProgramData") or "C:/ProgramData"
return Path(base) / "Tielogic" / "TieMeasureFlow" / "acquisitore"
return Path("acquisitore")
def create_app(folder: Path | str | None = None,
config_dir: Path | str | None = None) -> FastAPI:
archivio = calibrazione.Archivio(config_dir if config_dir is not None
else _config_dir_default()) # #83
aperte: dict = {}
lock = threading.Lock() # protegge SOLO i dizionari, mai un'apertura
lock_sorgenti: dict = {} # nome sorgente -> Lock che serializza apertura e grab
@@ -95,6 +140,24 @@ def create_app(folder: Path | str | None = None) -> FastAPI:
return {"sources": [{"name": v.name, "kind": v.kind, "label": v.label}
for v in voci if v.kind in ("camera", "virtual")]}
def _grab(nome: str, timeout: int):
"""Un frame dalla sorgente (apre e tiene aperta); errori camera -> HTTP."""
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
return 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)))
@app.post("/camera/grab")
def camera_grab(corpo: dict = Body(default=None)):
nome = corpo.get("source") if isinstance(corpo, dict) else None
@@ -104,28 +167,83 @@ def create_app(folder: Path | str | None = None) -> FastAPI:
trigger = corpo.get("trigger", "continuo")
if trigger not in _TRIGGER:
raise HTTPException(400, "«trigger» supportato: solo continuo")
grezzo = corpo.get("grezzo", False) # #84: true = frame non raddrizzato
if not isinstance(grezzo, bool):
raise HTTPException(400, "«grezzo» deve essere true o false")
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)))
frame = _grab(nome, timeout) # #83: corpo estratto in _grab, invariato
raddrizzato = False
if not grezzo: # #84: sorgente calibrata (densa) -> frame raddrizzato dalla mappa in cache
mappa = archivio.leggi_mappa(nome)
if mappa is not None:
corretto = calibrazione.raddrizza(frame, mappa)
if corretto is not None: # None = dimensioni diverse da quelle calibrate
frame, raddrizzato = corretto, True
ok, buf = cv2.imencode(".png", frame)
if not ok:
raise HTTPException(500, "codifica PNG fallita")
return Response(buf.tobytes(), media_type="image/png")
return Response(buf.tobytes(), media_type="image/png",
headers={"X-Raddrizzato": "si" if raddrizzato else "no"})
# ---- #83: calibrazione a scacchiera per sorgente
def _sorgente(valore) -> str:
if not isinstance(valore, str) or not valore.strip():
raise HTTPException(400, "manca «source»")
return valore.strip()
@app.get("/camera/calibration")
def calibrazione_leggi(source: str = ""):
dati = archivio.leggi(_sorgente(source))
if dati is None:
raise HTTPException(404, "sorgente non calibrata")
return dati
@app.delete("/camera/calibration", status_code=204)
def calibrazione_elimina(source: str = ""):
archivio.elimina(_sorgente(source))
return Response(status_code=204)
@app.post("/camera/calibrate")
async def calibrazione_calcola(
source: str = Form(...),
square_mm: float = Form(...),
cols: int | None = Form(None), # #84: opzionali, senza = calibrazione densa
rows: int | None = Form(None),
image: UploadFile | None = File(None),
save: bool = Form(False),
):
nome = _sorgente(source)
contenuto = await image.read() if image is not None else b""
if contenuto:
frame = cv2.imdecode(np.frombuffer(contenuto, np.uint8), cv2.IMREAD_UNCHANGED)
if frame is None:
raise HTTPException(422, "l'immagine caricata non è leggibile")
else:
frame = await run_in_threadpool(_grab, nome, 5000) # come /camera/grab (404/503/504)
if (cols is None) != (rows is None):
raise HTTPException(422, "cols e rows vanno indicati insieme (o nessuno dei due)")
mappa = None
try:
if cols is None: # #84: densa, nessuna griglia da dichiarare
risultato, mappa = await run_in_threadpool(
calibrazione.calibra_scacchiera_densa, frame, square_mm)
else: # metodo planare di #83, invariato
risultato = calibrazione.calibra_scacchiera(frame, cols, rows, square_mm)
except calibrazione.CalibrazioneError as e:
raise HTTPException(422, str(e))
corpo = {"source": nome,
**{k: v for k, v in risultato.items() if k != "overlay_png"},
"calibrated_at": calibrazione.ora_iso()}
if save:
archivio.salva(nome, corpo, mappa) # #84: la mappa (se densa) va in .npz accanto
return {**corpo, "overlay_png": risultato["overlay_png"], "saved": bool(save)}
@app.get("/calibrazione", response_class=HTMLResponse)
def calibrazione_pagina():
return HTMLResponse(PAGINA)
return app
@@ -136,8 +254,10 @@ def main() -> None:
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")
p.add_argument("--config-dir", default=None,
help="cartella delle calibrazioni salvate (default: ProgramData su Windows)") # #83
a = p.parse_args()
uvicorn.run(create_app(a.folder), host=a.host, port=a.port)
uvicorn.run(create_app(a.folder, a.config_dir), host=a.host, port=a.port)
if __name__ == "__main__":
+376
View File
@@ -0,0 +1,376 @@
"""Test della calibrazione dell'acquisitore (#83): logica pura e rotte di camera_service.
Uso: dalla radice, `uv run --python 3.13 --extra vision-worker --extra dev
--with-editable vendor/visionsuite/packages/vs-camera pytest scripts/tests -q`
(o il venv di VisionSuite come per test_camera_service.py).
"""
import base64
import importlib.util
import json
from pathlib import Path
import cv2
import numpy as np
import pytest
from fastapi.testclient import TestClient
SCRIPTS = Path(__file__).resolve().parents[1]
MODULO = SCRIPTS / "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
def una_scacchiera(square_px: int = 40, nx: int = 10, ny: int = 7, margin: int = 60) -> np.ndarray:
"""nx x ny quadrati => (nx-1) x (ny-1) angoli interni, su fondo bianco."""
img = np.full((ny * square_px + 2 * margin, nx * square_px + 2 * margin), 255, np.uint8)
for j in range(ny):
for i in range(nx):
if (i + j) % 2 == 0:
y, x = margin + j * square_px, margin + i * square_px
img[y:y + square_px, x:x + square_px] = 0
return img
def una_scacchiera_densa(w: int = 640, h: int = 480, square_px: int = 40, k: float = 1e-8) -> np.ndarray:
"""#84: scacchiera che riempie tutto il campo, con distorsione a barilotto (k sui px dal centro)."""
m = 3 * square_px # la scacchiera continua oltre il campo: nessun bordo finto nell'immagine
yy, xx = np.mgrid[0:h + 2 * m, 0:w + 2 * m]
piana = (((xx // square_px) + (yy // square_px)) % 2 * 255).astype(np.uint8)
cx, cy = w / 2.0, h / 2.0
dx, dy = (xx[m:m + h, m:m + w] - m - cx).astype(np.float32), (yy[m:m + h, m:m + w] - m - cy).astype(np.float32)
f = 1 + k * (dx * dx + dy * dy)
img = cv2.remap(piana, cx + m + dx * f, cy + m + dy * f, cv2.INTER_LINEAR)
return cv2.GaussianBlur(img, (3, 3), 0.8) # bordi non perfettamente netti, come in camera
def residuo_reticolo(griglia: np.ndarray) -> float:
"""Residuo massimo (px) della griglia (nr,nc,2) da un reticolo affine regolare."""
nr, nc, _ = griglia.shape
ii, jj = np.mgrid[0:nr, 0:nc]
A = np.stack([np.ones(nr * nc), ii.ravel(), jj.ravel()], axis=1)
P = griglia.reshape(-1, 2)
coef, *_ = np.linalg.lstsq(A, P, rcond=None)
return float(np.hypot(*(A @ coef - P).T).max())
def _png(img: np.ndarray) -> bytes:
ok, buf = cv2.imencode(".png", img)
assert ok
return buf.tobytes()
@pytest.fixture
def cartella(tmp_path):
# la sorgente camera:folder fa il loop su una scacchiera: il grab la trova
cartella = tmp_path / "immagini"
cartella.mkdir()
cv2.imwrite(str(cartella / "scacchiera.png"), una_scacchiera())
return cartella
@pytest.fixture
def client(cartella, tmp_path):
mod = _carica()
with TestClient(mod.create_app(cartella, config_dir=tmp_path / "config")) as c:
c.config = tmp_path / "config"
yield c
CAMPI = {"cols": "9", "rows": "6", "square_mm": "2"}
CHIAVI_C1 = {"source", "px_per_mm", "mm_per_px", "rms_px", "n_corners", "cols", "rows",
"square_mm", "image_width", "image_height", "scale_spread_pct", "calibrated_at"}
# #84: campi in piu' per metodo planare (C2 con cols/rows) e per la densa
CHIAVI_PLANARE = CHIAVI_C1 | {"metodo", "distorsione"}
CHIAVI_DENSA = CHIAVI_PLANARE | {"residuo_rms_px", "residuo_max_px", "copertura_pct"}
SORGENTE = "camera:folder"
@pytest.fixture
def cartella_densa(tmp_path):
# #84: la sorgente camera:folder fa il loop su una scacchiera densa distorta
cartella = tmp_path / "immagini_dense"
cartella.mkdir()
cv2.imwrite(str(cartella / "densa.png"), una_scacchiera_densa(k=1e-7))
return cartella
@pytest.fixture
def client_denso(cartella_densa, tmp_path):
mod = _carica()
with TestClient(mod.create_app(cartella_densa, config_dir=tmp_path / "config_densa")) as c:
c.config = tmp_path / "config_densa"
yield c
NON_IMMAGINE = b"questo non e una immagine"
# ---- logica pura
def test_scacchiera_sintetica_da_la_scala():
_carica() # mette scripts/ in sys.path
import calibrazione
r = calibrazione.calibra_scacchiera(una_scacchiera(), 9, 6, 2.0)
assert r["px_per_mm"] == pytest.approx(20.0, rel=0.005)
assert r["mm_per_px"] == pytest.approx(0.05, rel=0.005)
assert r["rms_px"] < 0.5
assert r["n_corners"] == 54
assert (r["image_width"], r["image_height"]) == (520, 400)
assert r["scale_spread_pct"] < 1.0
assert base64.b64decode(r["overlay_png"])[:8] == b"\x89PNG\r\n\x1a\n"
def test_immagine_bianca_solleva():
_carica()
import calibrazione
with pytest.raises(calibrazione.CalibrazioneError, match="non trovata"):
calibrazione.calibra_scacchiera(np.full((300, 300), 255, np.uint8), 9, 6, 2.0)
@pytest.mark.parametrize("cols,rows,quadrato", [(1, 6, 2.0), (9, 6, 0.0), (9, 6, -1.0)])
def test_parametri_invalidi_sollevano(cols, rows, quadrato):
_carica()
import calibrazione
with pytest.raises(calibrazione.CalibrazioneError):
calibrazione.calibra_scacchiera(una_scacchiera(), cols, rows, quadrato)
# ---- rotte
def test_c1_non_calibrata_404(client):
r = client.get("/camera/calibration", params={"source": "camera:folder"})
assert r.status_code == 404
assert r.json()["detail"] == "sorgente non calibrata"
def test_c1_senza_source_400(client):
assert client.get("/camera/calibration").status_code == 400
def test_c2_immagine_caricata_senza_salvare(client):
r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI},
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
assert r.status_code == 200, r.text
corpo = r.json()
assert CHIAVI_PLANARE | {"overlay_png", "saved"} == set(corpo)
assert corpo["metodo"] == "planare" and corpo["distorsione"] is False # #84: con cols/rows nulla cambia
assert corpo["saved"] is False
assert corpo["px_per_mm"] == pytest.approx(20.0, rel=0.005)
# senza save non si scrive nulla
assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404
assert not client.config.exists() or not list(client.config.glob("*.json"))
def test_c2_c1_c3_ciclo_con_grab_dalla_sorgente(client):
# nessun file: grab da camera:folder (la scacchiera in cartella), con save
r = client.post("/camera/calibrate",
data={"source": "camera:folder", "save": "true", **CAMPI})
assert r.status_code == 200, r.text
assert r.json()["saved"] is True
r1 = client.get("/camera/calibration", params={"source": "camera:folder"})
assert r1.status_code == 200
assert set(r1.json()) == CHIAVI_PLANARE # nessun overlay nel file salvato
assert r1.json()["px_per_mm"] == pytest.approx(20.0, rel=0.005)
assert r1.json()["source"] == "camera:folder"
file_salvati = list(client.config.glob("*.json"))
assert len(file_salvati) == 1
assert json.loads(file_salvati[0].read_text(encoding="utf-8"))["cols"] == 9
assert client.delete("/camera/calibration", params={"source": "camera:folder"}).status_code == 204
assert client.get("/camera/calibration", params={"source": "camera:folder"}).status_code == 404
# cancellare di nuovo e' idempotente
assert client.delete("/camera/calibration", params={"source": "camera:folder"}).status_code == 204
def test_c2_scacchiera_non_trovata_422(client):
bianca = np.full((300, 300), 255, np.uint8)
r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI},
files={"image": ("b.png", _png(bianca), "image/png")})
assert r.status_code == 422
assert "non trovata" in r.json()["detail"]
def test_c2_parametri_invalidi_422(client):
r = client.post("/camera/calibrate",
data={"source": "camera:folder", "cols": "1", "rows": "6", "square_mm": "2"},
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
assert r.status_code == 422
def test_c2_immagine_illeggibile_422(client):
r = client.post("/camera/calibrate", data={"source": "camera:folder", **CAMPI},
files={"image": ("x.png", NON_IMMAGINE, "image/png")})
assert r.status_code == 422
def test_c2_errori_camera_come_grab(client):
r = client.post("/camera/calibrate", data={"source": "camera:non-esiste", **CAMPI})
assert r.status_code == 404
def test_due_sorgenti_hanno_file_diversi(client):
for nome in ("camera:a b", "camera:a_b"):
r = client.post("/camera/calibrate", data={"source": nome, "save": "true", **CAMPI},
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
assert r.status_code == 200
assert len(list(client.config.glob("*.json"))) == 2
def test_c4_pagina_html(client):
r = client.get("/calibrazione")
assert r.status_code == 200
assert r.headers["content-type"].startswith("text/html")
assert "Acquisisci e calibra" in r.text
assert "Lato dello scacco" in r.text and 'id="cols"' not in r.text # #84: un solo campo
assert "http://" not in r.text and "https://" not in r.text # offline, niente CDN
# ---- #84: calibrazione densa (logica pura)
def test_densa_trova_la_griglia_da_sola_e_la_scala():
_carica()
import calibrazione
r, mappa = calibrazione.calibra_scacchiera_densa(una_scacchiera_densa(k=1e-8), 2.0)
assert (r["cols"], r["rows"]) == (15, 11) # 640x480 con quadrati da 40 px
assert r["n_corners"] == 15 * 11
assert r["mm_per_px"] == pytest.approx(2.0 / 40, rel=0.005)
assert r["px_per_mm"] == pytest.approx(20.0, rel=0.005)
assert r["metodo"] == "densa" and r["distorsione"] is True
assert r["residuo_max_px"] < 0.3 and r["residuo_rms_px"] <= r["residuo_max_px"]
assert 60 < r["copertura_pct"] <= 100
assert (r["image_width"], r["image_height"]) == (640, 480)
assert base64.b64decode(r["overlay_png"])[:8] == b"\x89PNG\r\n\x1a\n"
assert mappa.shape == (480, 640) and mappa.pixel_size == pytest.approx(0.05, rel=0.005)
def test_densa_raddrizza_su_griglia_regolare():
_carica()
import calibrazione
img = una_scacchiera_densa(k=1e-7) # distorsione marcata: ai bordi ~5 %
r, mappa = calibrazione.calibra_scacchiera_densa(img, 2.0)
assert (r["cols"], r["rows"]) == (15, 11)
# prima la griglia e' curva, dopo e' un reticolo regolare
assert residuo_reticolo(calibrazione._trova_griglia(img)) > 1.0
pronta = calibrazione._prepara_mappa(mappa)
dritta = calibrazione.raddrizza(img, pronta)
assert dritta.shape == img.shape
assert residuo_reticolo(calibrazione._trova_griglia(dritta)) < 0.3
@pytest.mark.parametrize("cosa", ["bianca", "scacchiera_piccola"])
def test_densa_senza_griglia_solleva(cosa):
_carica()
import calibrazione
img = (np.full((300, 300), 255, np.uint8) if cosa == "bianca"
else una_scacchiera_densa(80, 60, 20)) # 3x2 angoli: meno di 3x3
with pytest.raises(calibrazione.CalibrazioneError, match="non trovata"):
calibrazione.calibra_scacchiera_densa(img, 2.0)
@pytest.mark.parametrize("quadrato", [0.0, -1.0, float("nan")])
def test_densa_quadrato_invalido_solleva(quadrato):
_carica()
import calibrazione
with pytest.raises(calibrazione.CalibrazioneError):
calibrazione.calibra_scacchiera_densa(una_scacchiera_densa(), quadrato)
def test_archivio_mappa_cache_e_invalidazione(tmp_path):
_carica()
import calibrazione
_, mappa = calibrazione.calibra_scacchiera_densa(una_scacchiera_densa(), 2.0)
arch = calibrazione.Archivio(tmp_path)
assert arch.leggi_mappa("s") is None
arch.salva("s", {"a": 1}, mappa)
prima = arch.leggi_mappa("s")
assert prima is not None and arch.leggi_mappa("s") is prima # cache in memoria
assert not list(tmp_path.glob("*.tmp*")) # niente temporanei rimasti
arch.salva("s", {"a": 2}, mappa) # file riscritto -> cache invalidata
assert arch.leggi_mappa("s") is not prima
arch.salva("s", {"a": 3}) # salvataggio senza mappa (planare): la vecchia sparisce
assert arch.leggi_mappa("s") is None and not list(tmp_path.glob("*.npz"))
# ---- #84: rotte con la calibrazione densa
def _grigio(r) -> np.ndarray:
return cv2.imdecode(np.frombuffer(r.content, np.uint8), cv2.IMREAD_GRAYSCALE)
def test_c2_senza_cols_rows_e_densa(client_denso):
r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"})
assert r.status_code == 200, r.text
corpo = r.json()
assert CHIAVI_DENSA | {"overlay_png", "saved"} == set(corpo)
assert corpo["metodo"] == "densa" and corpo["distorsione"] is True
assert (corpo["cols"], corpo["rows"]) == (15, 11)
assert corpo["px_per_mm"] == pytest.approx(1.0 / corpo["mm_per_px"])
assert corpo["saved"] is False
assert not client_denso.config.exists() or not list(client_denso.config.iterdir())
def test_c2_solo_cols_o_solo_rows_422(client_denso):
r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "cols": "9"})
assert r.status_code == 422
def test_grab_raddrizzato_se_calibrata_e_grezzo_a_richiesta(client_denso, cartella_densa):
originale = cv2.imread(str(cartella_densa / "densa.png"), cv2.IMREAD_GRAYSCALE)
prima = client_denso.post("/camera/grab", json={"source": SORGENTE})
assert prima.headers["x-raddrizzato"] == "no" # non calibrata: frame com'e'
assert np.array_equal(_grigio(prima), originale)
r = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"})
assert r.status_code == 200, r.text
assert r.json()["saved"] is True
json_salvati, mappe = list(client_denso.config.glob("*.json")), list(client_denso.config.glob("*.npz"))
assert len(json_salvati) == 1 and len(mappe) == 1
assert mappe[0].stem == json_salvati[0].stem # stesso nome del JSON
saved = client_denso.get("/camera/calibration", params={"source": SORGENTE}).json()
assert set(saved) == CHIAVI_DENSA and saved["metodo"] == "densa"
dopo = client_denso.post("/camera/grab", json={"source": SORGENTE})
assert dopo.status_code == 200 and dopo.headers["x-raddrizzato"] == "si"
import calibrazione
dritta = _grigio(dopo)
assert dritta.shape == originale.shape
assert not np.array_equal(dritta, originale)
assert residuo_reticolo(calibrazione._trova_griglia(dritta)) < 0.3 # griglia regolare
grezzo = client_denso.post("/camera/grab", json={"source": SORGENTE, "grezzo": True})
assert grezzo.headers["x-raddrizzato"] == "no"
assert np.array_equal(_grigio(grezzo), originale)
# la calibrazione parte sempre dal frame grezzo: ricalibrare dalla sorgente da' lo stesso mm/px
di_nuovo = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2"})
assert di_nuovo.json()["mm_per_px"] == pytest.approx(r.json()["mm_per_px"])
assert di_nuovo.json()["residuo_max_px"] < 0.3
def test_grab_grezzo_non_booleano_400(client_denso):
r = client_denso.post("/camera/grab", json={"source": SORGENTE, "grezzo": "si"})
assert r.status_code == 400
def test_delete_toglie_anche_la_mappa(client_denso):
client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"})
assert list(client_denso.config.glob("*.npz"))
assert client_denso.delete("/camera/calibration", params={"source": SORGENTE}).status_code == 204
assert not list(client_denso.config.glob("*.npz")) and not list(client_denso.config.glob("*.json"))
assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no"
def test_ricalibrare_planare_toglie_la_mappa_densa(client_denso):
client_denso.post("/camera/calibrate", data={"source": SORGENTE, "square_mm": "2", "save": "true"})
assert list(client_denso.config.glob("*.npz"))
planare = client_denso.post("/camera/calibrate", data={"source": SORGENTE, "save": "true", **CAMPI},
files={"image": ("s.png", _png(una_scacchiera()), "image/png")})
assert planare.status_code == 200 and planare.json()["metodo"] == "planare"
assert not list(client_denso.config.glob("*.npz")) # niente mappa vecchia che raddrizza a vuoto
assert client_denso.post("/camera/grab", json={"source": SORGENTE}).headers["x-raddrizzato"] == "no"
+7 -2
View File
@@ -3,6 +3,7 @@
Protected by SETUP_PASSWORD env var. When empty/None, all endpoints return 404.
NOT protected by API Key auth (standalone access).
"""
import asyncio
import hmac
import random
from datetime import datetime, timedelta
@@ -312,7 +313,9 @@ async def init_db(body: PasswordBody):
await _require_not_initialised()
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
_stamp_head()
# #80: env.py di alembic fa asyncio.run(), vietato dentro il loop di FastAPI;
# in un thread non c'e' un loop attivo e funziona.
await asyncio.to_thread(_stamp_head)
return {"status": "ok", "message": "Database tables created successfully"}
@@ -748,6 +751,8 @@ async def reset_db(body: ResetDbBody):
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
_stamp_head()
# #80: env.py di alembic fa asyncio.run(), vietato dentro il loop di FastAPI;
# in un thread non c'e' un loop attivo e funziona.
await asyncio.to_thread(_stamp_head)
return {"status": "ok", "message": "Database reset successfully"}
+17 -2
View File
@@ -27,6 +27,10 @@ class Settings(BaseSettings):
db_name: str = "tiemeasureflow"
db_user: str = "tmflow"
db_password: str = _DEFAULT_DB_PASSWORD
# Full SQLAlchemy async URL (env DB_URL). When set it wins over the
# DB_HOST/DB_PORT/DB_NAME/DB_USER/DB_PASSWORD parts. Used by the Windows
# package: sqlite+aiosqlite:///C:/ProgramData/.../tmf.db (forward slashes).
db_url: str | None = None
# Presentation time zone (DB is UTC, #42)
display_timezone: str = "Europe/Rome"
@@ -78,7 +82,9 @@ class Settings(BaseSettings):
@property
def database_url(self) -> str:
"""Async MySQL connection string."""
"""Async connection string: DB_URL if set, else the MySQL one."""
if self.db_url:
return self.db_url
return (
f"mysql+asyncmy://{self.db_user}:{self.db_password}"
f"@{self.db_host}:{self.db_port}/{self.db_name}"
@@ -97,7 +103,12 @@ class Settings(BaseSettings):
(mounted as a Docker volume), not inside the backend tree.
"""
# Path(__file__) = src/backend/config.py → parents[2] = project root
return Path(__file__).resolve().parents[2] / self.upload_dir
# UPLOAD_DIR may be absolute (Windows package: C:/ProgramData/...):
# in that case it is used as is instead of being joined to the root.
configured = Path(self.upload_dir)
if configured.is_absolute():
return configured
return Path(__file__).resolve().parents[2] / configured
# Always resolve .env against the project root regardless of cwd
# (pydantic-settings would otherwise treat the path as cwd-relative).
@@ -122,6 +133,10 @@ class Settings(BaseSettings):
"SERVER_SECRET_KEY is still the committed default value; "
"set it before running with APP_ENV=" + self.app_env
)
# A SQLite file has no DB password: the default-password guard only
# makes sense for the MySQL connection built from the DB_* parts.
if self.db_url and self.db_url.startswith("sqlite"):
return self
if self.db_password == _DEFAULT_DB_PASSWORD:
raise RuntimeError(
"DB_PASSWORD is still the committed default value; "
+25 -8
View File
@@ -1,6 +1,7 @@
"""Async SQLAlchemy database engine and session management."""
from collections.abc import AsyncGenerator
from sqlalchemy import event
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
@@ -10,14 +11,30 @@ from sqlalchemy.orm import DeclarativeBase
from src.backend.config import settings
# Create async engine
engine = create_async_engine(
settings.database_url,
echo=False,
pool_size=10,
max_overflow=20,
pool_recycle=3600,
)
_IS_SQLITE = settings.database_url.startswith("sqlite")
# Create async engine. MySQL gets a connection pool; SQLite (local file of the
# Windows package) does not accept pool_size/max_overflow/pool_recycle.
if _IS_SQLITE:
engine = create_async_engine(settings.database_url, echo=False)
@event.listens_for(engine.sync_engine, "connect")
def _sqlite_pragmas(dbapi_connection, _connection_record) -> None:
"""Per-connection PRAGMAs: SQLite ignores FKs unless asked, and WAL +
busy_timeout let the web, vision and backend threads share the file."""
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.execute("PRAGMA journal_mode=WAL")
cursor.execute("PRAGMA busy_timeout=5000")
cursor.close()
else:
engine = create_async_engine(
settings.database_url,
echo=False,
pool_size=10,
max_overflow=20,
pool_recycle=3600,
)
# Session factory
async_session_factory = async_sessionmaker(
+18
View File
@@ -1,6 +1,7 @@
"""Report generation service — PDF reports via Jinja2 + Plotly/Kaleido + WeasyPrint."""
import asyncio
import base64
import tempfile
from datetime import datetime
from src.backend.utils.tempo import to_local, utcnow
from pathlib import Path
@@ -25,6 +26,23 @@ from src.backend.services.spc_service import (
compute_summary,
)
# Kaleido 1.x aggiunge alla pagina di render uno <script> MathJax dal CDN (cdnjs) anche se i grafici
# non usano LaTeX: senza Internet (stand, PC di produzione) ogni grafico aspetta il timeout di rete
# (~19 s con rete che scarta i pacchetti, TieMeasureFlow#78). I grafici qui sotto non hanno formule:
# si passa a Kaleido un file locale vuoto al posto del CDN. Il file sta nella cartella temporanea.
def _mathjax_vuoto() -> str:
"""Percorso di un .js vuoto da usare come MathJax: nessuna richiesta di rete per i grafici."""
path = Path(tempfile.gettempdir()) / "tmf-mathjax-vuoto.js"
if not path.exists():
path.write_text("/* MathJax non usato: grafici senza LaTeX */", encoding="utf-8")
return str(path)
try:
pio.defaults.mathjax = _mathjax_vuoto()
except (AttributeError, OSError): # plotly vecchio senza pio.defaults o temp non scrivibile: resta il CDN
pass
# Jinja2 environment for report templates
_templates_dir = Path(__file__).parent.parent / "templates" / "reports"
_jinja_env = Environment(
+164
View File
@@ -0,0 +1,164 @@
"""Report PDF veri su Windows (WeasyPrint + DLL GTK + Kaleido con Edge/Chrome), offline.
Pacchetto Windows (TieMeasureFlow#78): senza Internet i report devono uscire lo stesso.
Il test gira solo su Windows, con le DLL di packaging/windows/pdf/prendi-gtk.ps1 e un browser
Chromium (Edge o Chrome) installati; altrimenti viene saltato.
Il conftest dei test sostituisce WeasyPrint con un mock: qui serve quello vero, quindi la
generazione gira in un sottoprocesso che importa weasyprint PRIMA del conftest (il mock scatta
solo se weasyprint non e' gia' in sys.modules).
Come preparare le DLL: powershell -File packaging/windows/pdf/prendi-gtk.ps1 -Destinazione packaging/windows/pdf/gtk
(oppure WEASYPRINT_DLL_DIRECTORIES gia' impostata su una cartella con le DLL).
"""
import json
import os
import subprocess
import sys
from pathlib import Path
import pytest
REPO = Path(__file__).resolve().parents[3]
GTK_DEFAULT = REPO / "packaging" / "windows" / "pdf" / "gtk"
# Browser Chromium cercati in ordine (Edge e' presente su ogni Windows 10/11).
_BROWSER_CANDIDATI = [
r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe",
r"C:\Program Files\Microsoft\Edge\Application\msedge.exe",
r"C:\Program Files\Google\Chrome\Application\chrome.exe",
r"C:\Program Files (x86)\Google\Chrome\Application\chrome.exe",
]
def _cartella_gtk() -> Path | None:
"""Cartella DLL: WEASYPRINT_DLL_DIRECTORIES (primo elemento) o quella di prendi-gtk.ps1."""
env = os.environ.get("WEASYPRINT_DLL_DIRECTORIES", "").split(";")[0]
for c in (Path(env) if env else None, GTK_DEFAULT):
if c and (c / "libpango-1.0-0.dll").exists():
return c
return None
def _browser() -> str | None:
env = os.environ.get("BROWSER_PATH")
if env and Path(env).exists():
return env
return next((b for b in _BROWSER_CANDIDATI if Path(b).exists()), None)
def _tz_ok() -> bool:
"""Windows non ha il database IANA: serve il pacchetto tzdata (to_local usa ZoneInfo)."""
try:
from zoneinfo import ZoneInfo
ZoneInfo("Europe/Rome")
return True
except Exception:
return False
pytestmark = [
pytest.mark.skipif(sys.platform != "win32", reason="solo Windows"),
pytest.mark.skipif(_cartella_gtk() is None, reason="DLL GTK assenti: lancia prendi-gtk.ps1"),
pytest.mark.skipif(_browser() is None, reason="nessun Edge/Chrome installato"),
# Senza tzdata i report falliscono su Windows (ZoneInfoNotFoundError): il pacchetto deve dichiararlo.
pytest.mark.skipif(not _tz_ok(), reason="manca tzdata: aggiungere 'tzdata; sys_platform == \"win32\"' a pyproject"),
]
# Sottoprocesso: dati veri nel DB SQLite in memoria del conftest, poi i due report reali.
_WORKER = r"""
import asyncio, json, sys
sys.path.insert(0, sys.argv[1])
import weasyprint # VERO, prima del conftest (che altrimenti lo sostituisce con un mock)
assert not type(weasyprint.HTML).__module__.startswith("unittest"), "WeasyPrint mockato"
from datetime import datetime, timedelta
from src.backend.tests.conftest import Base, TestSessionFactory, test_engine, _create_user, create_test_recipe
from src.backend.models.orm.measurement import Measurement
from src.backend.models.orm.task import RecipeSubtask
from src.backend.services import report_service
from sqlalchemy import select
VALORI = [10.02, 9.98, 10.05, 10.11, 9.94, 10.0, 10.07, 9.96, 10.03, 10.09, 9.99, 10.04, 10.38, 10.01, 9.97]
async def main():
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with TestSessionFactory() as db:
user = await _create_user(db, "metrologo", roles=["Metrologist"])
recipe = await create_test_recipe(db, user.id)
version = recipe.versions[0]
sub = (await db.execute(select(RecipeSubtask))).scalars().first()
t0 = datetime.utcnow() - timedelta(hours=1)
for i, v in enumerate(VALORI):
pf = "warning" if v > 10.3 else "pass"
db.add(Measurement(subtask_id=sub.id, version_id=version.id, measured_by=user.id,
value=v, pass_fail=pf, measured_at=t0 + timedelta(minutes=i)))
await db.commit()
spc = await report_service.generate_spc_report(db, recipe.id, sub.id)
mis = await report_service.generate_measurement_report(db, recipe.id)
open(sys.argv[2] + "/spc.pdf", "wb").write(spc)
open(sys.argv[2] + "/misure.pdf", "wb").write(mis)
# MathJax locale: senza, Kaleido chiede lo script al CDN e senza Internet ogni grafico attende ~19 s.
from pathlib import Path
mj = report_service.pio.defaults.mathjax
print(json.dumps({"ok": True, "mathjax_locale": bool(mj) and Path(mj).is_file()}))
asyncio.run(main())
"""
@pytest.fixture(scope="module")
def pdf_generati(tmp_path_factory):
"""Genera i due PDF in un sottoprocesso con proxy finto: se serve Internet, fallisce."""
out = tmp_path_factory.mktemp("report_win")
env = dict(os.environ)
env.update(
WEASYPRINT_DLL_DIRECTORIES=str(_cartella_gtk()),
BROWSER_PATH=_browser(),
# Proxy morto per la parte Python (Chrome/Edge usano invece il proxy di sistema: vedi LEGGIMI.md).
HTTP_PROXY="http://127.0.0.1:9", HTTPS_PROXY="http://127.0.0.1:9",
PYTHONIOENCODING="utf-8",
)
# Il sottoprocesso non deve leggere il .env di sviluppo ne' un DB reale.
env.pop("DB_URL", None)
res = subprocess.run(
[sys.executable, "-W", "ignore", "-c", _WORKER, str(REPO), str(out)],
cwd=REPO, env=env, capture_output=True, text=True, timeout=180,
)
assert res.returncode == 0, f"sottoprocesso fallito:\n{res.stdout}\n{res.stderr}"
esito = json.loads(res.stdout.strip().splitlines()[-1])
assert esito["ok"] is True
assert esito["mathjax_locale"], "Kaleido userebbe MathJax dal CDN: niente rete = report lenti"
return out
def _pdf(path: Path):
import pdfplumber
return pdfplumber.open(path)
def test_spc_ha_carta_di_controllo_e_istogramma(pdf_generati):
"""Il report SPC e' un PDF vero con una pagina almeno e i due grafici disegnati."""
path = pdf_generati / "spc.pdf"
assert path.read_bytes()[:5] == b"%PDF-"
with _pdf(path) as pdf:
assert len(pdf.pages) >= 1
testo = "\n".join(p.extract_text() or "" for p in pdf.pages)
# I titoli stanno dentro l'SVG di Plotly: se ci sono, Kaleido ha reso e WeasyPrint ha disegnato.
assert "Control Chart" in testo
assert "Histogram" in testo
# Grafico presente = segni vettoriali (tracce, assi, barre) oltre al testo.
segni = sum(len(p.curves) + len(p.lines) + len(p.rects) for p in pdf.pages)
assert segni > 40, f"troppo pochi segni vettoriali ({segni}): grafico assente?"
def test_report_misure_e_pdf_vero(pdf_generati):
"""Anche il report a tabella (solo WeasyPrint, niente Kaleido) esce con le righe misurate."""
path = pdf_generati / "misure.pdf"
assert path.read_bytes()[:5] == b"%PDF-"
with _pdf(path) as pdf:
assert len(pdf.pages) >= 1
testo = "\n".join(p.extract_text() or "" for p in pdf.pages)
assert "10.3800" in testo # il valore anomalo e' in tabella
+81
View File
@@ -0,0 +1,81 @@
"""init-db / reset-db con _stamp_head REALE su un file SQLite (TieMeasureFlow#80).
_stamp_head esegue migrations/env.py, che chiama asyncio.run(): dentro il loop
gia' attivo di FastAPI dava RuntimeError -> 500. test_setup.py e
test_station_seed.py sostituiscono _stamp_head con un no-op e non lo vedevano:
qui NON si fa monkeypatch di _stamp_head.
"""
import logging.config
import sqlite3
import pytest
from alembic.config import Config
from alembic.script import ScriptDirectory
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from src.backend.config import settings
from src.backend.main import app
SETUP_PWD = "test-setup-password"
@pytest.fixture
def file_db(tmp_path, monkeypatch):
"""Punta settings, engine e session factory del router setup a un file SQLite."""
import src.backend.api.routers.setup as setup_mod
db_file = tmp_path / "t.db"
url = f"sqlite+aiosqlite:///{db_file.as_posix()}"
# env.py ricava sqlalchemy.url da settings.database_url (property su db_url)
monkeypatch.setattr(settings, "db_url", url)
monkeypatch.setattr(settings, "setup_password", SETUP_PWD)
file_engine = create_async_engine(url, echo=False)
monkeypatch.setattr(setup_mod, "engine", file_engine)
monkeypatch.setattr(
setup_mod, "async_session_factory",
async_sessionmaker(file_engine, expire_on_commit=False),
)
# env.py chiama fileConfig, che disattiverebbe i logger degli altri test
monkeypatch.setattr(logging.config, "fileConfig", lambda *a, **k: None)
return db_file, setup_mod, file_engine
def _head() -> str:
from src.backend.api.routers.setup import _alembic_ini
cfg = Config(str(_alembic_ini))
return ScriptDirectory.from_config(cfg).get_current_head()
def _versions(db_file) -> list[str]:
con = sqlite3.connect(db_file)
try:
return [r[0] for r in con.execute("SELECT version_num FROM alembic_version")]
finally:
con.close()
@pytest.fixture
async def http():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as ac:
yield ac
@pytest.mark.asyncio
async def test_init_db_stamps_head(file_db, http):
db_file, _, file_engine = file_db
resp = await http.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200, resp.text
assert _versions(db_file) == [_head()]
await file_engine.dispose()
@pytest.mark.asyncio
async def test_reset_db_stamps_head(file_db, http):
db_file, _, file_engine = file_db
resp = await http.post("/api/setup/init-db", json={"password": SETUP_PWD})
assert resp.status_code == 200, resp.text
resp = await http.post("/api/setup/reset-db", json={"password": SETUP_PWD, "confirm": True})
assert resp.status_code == 200, resp.text
assert _versions(db_file) == [_head()]
await file_engine.dispose()
+144
View File
@@ -0,0 +1,144 @@
"""Backend su un file SQLite locale (pacchetto Windows, TieMeasureFlow#77).
Il modulo `database` costruisce l'engine all'import leggendo l'ambiente, quindi
i test che riguardano DB_URL girano in un sottoprocesso con DB_URL/UPLOAD_DIR
impostati: e' lo stesso percorso di avvio.py nel pacchetto, senza toccare
l'engine in memoria usato dal resto della suite.
"""
import json
import os
import subprocess
import sys
import textwrap
from pathlib import Path
from src.backend.config import Settings
_ROOT = Path(__file__).resolve().parents[3]
# Script eseguito nel sottoprocesso: upgrade head da DB vuoto, poi app reale
# (health, login con utente creato via ORM), PRAGMA e vincolo FK.
_SCRIPT = textwrap.dedent(
"""
import asyncio, json, sys
from pathlib import Path
from unittest.mock import MagicMock
# WeasyPrint senza GTK: stesso mock di conftest.py (le DLL sono affare del piano B)
sys.modules.setdefault("weasyprint", MagicMock())
from alembic import command
from alembic.config import Config
from src.backend.config import settings
ini = Path(r"{ini}")
command.upgrade(Config(str(ini)), "head")
from httpx import ASGITransport, AsyncClient
from sqlalchemy import select, text
from sqlalchemy.exc import IntegrityError
from src.backend.database import async_session_factory, engine
from src.backend.main import app
from src.backend.models.orm.user import User
from src.backend.models.orm.recipe import RecipeVersion
from src.backend.services.auth_service import hash_password
async def main():
out = {{}}
async with async_session_factory() as s:
s.add(User(username="fiera", password_hash=hash_password("Pass12345"),
display_name="Fiera", roles=["Maker"], is_admin=True))
await s.commit()
async with async_session_factory() as s:
out["utente"] = (await s.execute(select(User.username))).scalars().all()
out["fk"] = (await s.execute(text("PRAGMA foreign_keys"))).scalar()
out["journal"] = (await s.execute(text("PRAGMA journal_mode"))).scalar()
out["busy"] = (await s.execute(text("PRAGMA busy_timeout"))).scalar()
# FK attive: una versione con recipe_id inesistente deve essere rifiutata
try:
async with async_session_factory() as s:
s.add(RecipeVersion(recipe_id=99999, version_number=1, created_by=1))
await s.commit()
out["fk_violation"] = False
except IntegrityError:
out["fk_violation"] = True
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://t") as c:
out["health"] = (await c.get("/api/health")).status_code
r = await c.post("/api/auth/login", json={{"username": "fiera", "password": "Pass12345"}})
out["login"] = r.status_code
out["upload_path"] = str(settings.upload_path)
out["database_url"] = settings.database_url
await engine.dispose()
print("RISULTATO=" + json.dumps(out))
asyncio.run(main())
"""
)
def _run(tmp_path: Path, extra_env: dict[str, str]) -> dict:
ini = _ROOT / "src" / "backend" / "migrations" / "alembic.ini"
env = {k: v for k, v in os.environ.items() if k not in ("DB_URL", "UPLOAD_DIR")}
env.update(extra_env)
env["PYTHONPATH"] = str(_ROOT)
env["PYTHONIOENCODING"] = "utf-8"
proc = subprocess.run(
[sys.executable, "-c", _SCRIPT.format(ini=str(ini))],
cwd=str(_ROOT), env=env, capture_output=True, text=True, timeout=180,
)
assert proc.returncode == 0, proc.stderr[-3000:]
line = [l for l in proc.stdout.splitlines() if l.startswith("RISULTATO=")][-1]
return json.loads(line[len("RISULTATO="):])
def test_file_sqlite_upgrade_app_pragma_e_upload_assoluto(tmp_path):
db = tmp_path / "tmf.db"
uploads = tmp_path / "uploads"
# barre "/" nell'URL come nel pacchetto Windows (D:/... non va in escape)
url = f"sqlite+aiosqlite:///{db.as_posix()}"
out = _run(tmp_path, {"DB_URL": url, "UPLOAD_DIR": str(uploads)})
assert out["database_url"] == url
assert db.exists() # il file e' stato creato da upgrade head
assert out["utente"] == ["fiera"]
assert out["health"] == 200
assert out["login"] == 200
assert out["fk"] == 1 # PRAGMA foreign_keys attivo
assert out["fk_violation"] is True
assert out["journal"] == "wal"
assert out["busy"] == 5000
# UPLOAD_DIR assoluto rispettato (non accodato alla radice del progetto)
assert Path(out["upload_path"]) == uploads
def test_database_url_senza_db_url_resta_mysql(monkeypatch):
monkeypatch.delenv("DB_URL", raising=False)
s = Settings(_env_file=None, db_url=None)
assert s.database_url.startswith("mysql+asyncmy://")
def test_db_url_vince_sulle_parti_db(monkeypatch):
s = Settings(_env_file=None, db_url="sqlite+aiosqlite:///C:/x/tmf.db", db_host="altro")
assert s.database_url == "sqlite+aiosqlite:///C:/x/tmf.db"
def test_produzione_con_sqlite_non_richiede_db_password():
# niente RuntimeError: con SQLite la password DB di default non conta
Settings(
_env_file=None, app_env="production", server_secret_key="x" * 40,
db_url="sqlite+aiosqlite:///C:/x/tmf.db",
)
def test_produzione_mysql_rifiuta_ancora_password_di_default():
import pytest
with pytest.raises(Exception, match="DB_PASSWORD"):
Settings(_env_file=None, app_env="production", server_secret_key="x" * 40)
def test_upload_dir_relativo_resta_sotto_la_radice():
s = Settings(_env_file=None, upload_dir="uploads")
assert s.upload_path == _ROOT / "uploads"
def test_upload_dir_assoluto(tmp_path):
s = Settings(_env_file=None, upload_dir=str(tmp_path / "u"))
assert s.upload_path == tmp_path / "u"
+16 -18
View File
@@ -12,12 +12,9 @@ class TestSelectRecipe:
def test_select_recipe_renders(self, logged_in_client, mock_api_client, monkeypatch):
"""Recipe selection page renders for MeasurementTec role."""
monkeypatch.setenv("STATION_CODE", "ST-TEST")
import config
import importlib
importlib.reload(config)
import blueprints.measure
importlib.reload(blueprints.measure)
# Patch the class attribute instead of reload()ing config/blueprints.measure:
# monkeypatch undoes it at teardown, so no ST-TEST leaks into later tests (#79).
monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", "ST-TEST")
mock_api_client.get_station_recipes.return_value = [
{"id": 1, "code": "REC-001", "name": "Test Recipe"},
@@ -47,12 +44,9 @@ class TestSelectRecipe:
import re
from html.parser import HTMLParser
monkeypatch.setenv("STATION_CODE", "ST-TEST")
import config
import importlib
importlib.reload(config)
import blueprints.measure
importlib.reload(blueprints.measure)
# Patch the class attribute instead of reload()ing config/blueprints.measure:
# monkeypatch undoes it at teardown, so no ST-TEST leaks into later tests (#79).
monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", "ST-TEST")
mock_api_client.get_station_recipes.return_value = []
@@ -79,12 +73,9 @@ class TestSelectRecipe:
self, logged_in_client, mock_api_client, monkeypatch
):
"""#10: a lot number like ``L'23`` must not break the page."""
monkeypatch.setenv("STATION_CODE", "ST-TEST")
import config
import importlib
importlib.reload(config)
import blueprints.measure
importlib.reload(blueprints.measure)
# Patch the class attribute instead of reload()ing config/blueprints.measure:
# monkeypatch undoes it at teardown, so no ST-TEST leaks into later tests (#79).
monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", "ST-TEST")
mock_api_client.get_station_recipes.return_value = []
@@ -393,6 +384,13 @@ class TestStartSequence:
class TestTaskComplete:
"""GET /measure/complete/<recipe_id> tests."""
@pytest.fixture(autouse=True)
def _station(self, monkeypatch):
# The side_effect lists below include the /production-runs/current reply,
# which is only requested when a station is configured: pin it explicitly
# instead of relying on whatever an earlier test left behind (#79).
monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", "ST-TEST")
def test_task_complete_renders(self, logged_in_client, mock_api_client):
"""The summary carries the measurement clock, so it has to render with it."""
mock_api_client.get.side_effect = [
@@ -6,6 +6,17 @@ a start button that will not start, a description that reads the way it was
written.
"""
import pytest
@pytest.fixture(autouse=True)
def _no_station(monkeypatch):
# These tests assume no station is configured (e.g. _task_list feeds the mock a
# fixed number of replies, with no /production-runs/current among them); pin it
# so a STATION_CODE left behind by an earlier test cannot change the call count (#79).
# Tests that need a station set it themselves, after this fixture has run.
monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", None)
# ---------------------------------------------------------------------------
# Point 11 - line breaks and bold, and nothing else
@@ -197,13 +208,9 @@ def test_avvia_is_untouched_when_nothing_is_required(
def test_selection_screen_carries_the_traceability_rules(
logged_in_client, mock_api_client, monkeypatch,
):
monkeypatch.setenv("STATION_CODE", "ST-TEST")
import importlib
import config
importlib.reload(config)
import blueprints.measure
importlib.reload(blueprints.measure)
# Patch instead of reload()ing config/blueprints.measure: the patch is undone at
# teardown, a reload would leave ST-TEST behind for later tests (#79).
monkeypatch.setattr("blueprints.measure.Config.STATION_CODE", "ST-TEST")
mock_api_client.get_station_recipes.return_value = [
{
+228
View File
@@ -0,0 +1,228 @@
"""Costruttore di ricette DXF compare dal campo (immagine + DXF + scala -> ricetta v2).
Logica pura, niente FastAPI ne' database: il worker (src/vision_worker/main.py)
lo espone come /recipes/dxf-compare, i test lo chiamano direttamente. Tutta la
visione sta in TieVision-Core (visionsuite.*). La calibrazione a scacchiera NON
sta qui: vive nell'acquisitore (scripts/calibrazione.py, #83).
Errori di input (pezzo non trovato, DXF illeggibile, parametri fuori range)
sono `CostruttoreError`: il worker li mappa a 422.
"""
from __future__ import annotations
import base64
import hashlib
import json
import math
import tempfile
from pathlib import Path
import numpy as np
from visionsuite.core.image import Image
from src.vision.runner import RunOutcome, run_graph
class CostruttoreError(ValueError):
"""Input che non permette di costruire (422 per il worker), non un guasto."""
# Nomi/id fissi della ricetta (contratto #83 con backend e frontend).
DXF_FILE = "disegno.dxf"
CALIBRATION_FILE = "calibrazione.json"
_POLARITIES = ("dark", "light")
# Pose iniziali del disegno provate (gradi): il verso del pezzo non e' noto.
_OUTPUT_KEYS = ("dxf.points_in_pct", "dxf.distance_avg", "dxf.distance_min", "dxf.distance_max")
_START_ANGLES = (0.0, 90.0, 180.0, 270.0)
def _b64(data: bytes) -> str:
return base64.b64encode(data).decode("ascii")
def _gray_u8(pixels: np.ndarray) -> np.ndarray:
"""Immagine -> grigio uint8 2D (LocateBlob accetta solo uint8)."""
px = np.asarray(pixels)
if px.ndim == 3:
px = px[..., 0] # multicanale: il primo, come fa LocateBlob
if px.dtype != np.uint8:
# float/uint16 -> scala sul massimo, come _to_gray_uint8 di vs-core
m = float(px.max()) if px.size else 1.0
px = np.clip(px.astype(np.float32) / max(m, 1e-9) * 255, 0, 255).astype(np.uint8)
return np.ascontiguousarray(px)
# ------------------------------------------------------------------ W2
def _graph(dxf_sha: str, cal_sha: str, band_mm: float, min_points_pct: float, polarity: str,
reference_pose: dict, anchor: tuple[float, float, float] | None) -> dict:
"""Ricetta v2 fissa: LocateBlob "pezzo" + CompareDxf "dxf" agganciato a `align`.
`anchor` = (riga, colonna, angolo) dell'origine del disegno nell'immagine di
riferimento; None = solo il localizzatore (passo intermedio).
"""
tools: list[dict] = [{
"id": "pezzo", "name": "Pezzo", "kind": "LocateBlob",
"values": {"polarity": polarity, "reference_pose": reference_pose},
"links": {},
}]
if anchor is not None:
row, col, angle = anchor
tools.append({
"id": "dxf", "name": "Confronto DXF", "kind": "CompareDxf",
"values": {
"dxf": "disegno",
"anchor": "align",
"anchor_offset_row_px": float(row),
"anchor_offset_col_px": float(col),
"offset_angle_deg": float(angle),
"band_plus": float(band_mm),
"band_minus": float(band_mm),
"nominal_unit": "mm",
# tolleranza di points_in_pct: nominale = limite, plus = 100 - limite
"points_in_pct_nominal": float(min_points_pct),
"points_in_pct_plus": float(100.0 - min_points_pct),
"points_in_pct_minus": 0.0,
},
"links": {"align": {"tool": "pezzo", "property": "poses"}},
})
return {
"schema_version": 2,
"name": "Confronto DXF",
# sha256 obbligatorio per vs-task (verifica i file della cartella ricetta)
"meta": {"dxf": {"disegno": {"path": DXF_FILE, "sha256": dxf_sha}},
"calibration": {"path": CALIBRATION_FILE, "sha256": cal_sha}},
"tools": tools,
}
def _locate_reference(gray: np.ndarray, polarity: str) -> dict:
"""Posa del pezzo sull'immagine di riferimento (LocateBlob, soglia Otsu)."""
from visionsuite.task import Task, load_tools
load_tools()
graph = _graph('', '', 1.0, 95.0, polarity, {"row": 0.0, "col": 0.0, "angle_deg": 0.0}, None)
task = Task.from_dict(graph)
task.run(context={"image": Image(gray)})
pezzo = task["Pezzo"]
if not pezzo.found_poses:
raise CostruttoreError(
"pezzo non trovato nell'immagine di riferimento "
f"(polarity={polarity}): controllare luce, sfondo e polarita'"
)
return pezzo.found_poses[0].to_dict()
def _candidate_anchors(gray: np.ndarray, dxf_path: Path, px_per_mm: float) -> list[tuple[float, float, float]]:
"""Pose (riga, col, angolo) del disegno sull'immagine: ICP da 4 orientamenti.
Il disegno e' in px (origine DXF, y capovolta); si parte da ognuno dei 4
angoli con i centroidi sovrapposti e l'ICP rifinisce. Ritorna le posa trovate
(anche per orientamenti che non convergono: le giudica la ricetta eseguita).
"""
from visionsuite.calibration import ScaleCalibration
from visionsuite.dxf.compare import DXFCompare, _stack
from visionsuite.task.pose import Pose
cmp_ = DXFCompare(dxf_path, ScaleCalibration(px_per_mm=px_per_mm))
drawing = _stack(cmp_.drawing_contours())
measured = _stack(cmp_.measured_contours(Image(gray)))
if len(drawing) < 3:
raise CostruttoreError("DXF senza geometria utilizzabile")
if len(measured) < 3:
raise CostruttoreError("pezzo non trovato: nessun bordo nell'immagine")
dc, mc = drawing.mean(axis=0), measured.mean(axis=0)
anchors = []
for theta in _START_ANGLES:
# traslazione che porta il centroide del disegno ruotato sul misurato
rr, cc = Pose(0.0, 0.0, theta).apply(dc[0], dc[1])
start = Pose(float(mc[0] - rr), float(mc[1] - cc), theta)
try:
result = cmp_.compare(Image(gray), initial_pose=start)
except Exception: # noqa: BLE001 - un orientamento che cade non ferma gli altri
continue
# transform: misurato -> disegno; la posa del disegno e' l'inversa
m = np.linalg.inv(result.transform)
angle = math.degrees(math.atan2(m[1, 0], m[0, 0]))
anchors.append((float(m[0, 2]), float(m[1, 2]), angle))
return anchors
def _run(gray, graph, px_per_mm, folder) -> RunOutcome:
return run_graph(gray, graph, calibration={"px_per_mm": px_per_mm}, recipe_dir=folder)
def build_dxf_compare(
gray,
dxf_bytes: bytes,
px_per_mm: float,
band_mm: float = 0.1,
min_points_pct: float = 95.0,
polarity: str = "dark",
) -> dict:
"""Ricetta DXF compare dall'immagine di riferimento + disegno + scala.
Ritorna {graph, files, outputs, overlay, outcome}. `files` sono i due file
della cartella ricetta (b64): disegno.dxf e calibrazione.json.
"""
if not (px_per_mm > 0 and math.isfinite(px_per_mm)):
raise CostruttoreError("px_per_mm deve essere > 0")
if not (band_mm > 0 and math.isfinite(band_mm)):
raise CostruttoreError("band_mm deve essere > 0")
if not (0 < min_points_pct <= 100):
raise CostruttoreError("min_points_pct deve stare in (0, 100]")
if polarity not in _POLARITIES:
raise CostruttoreError(f"polarity deve essere una di {list(_POLARITIES)}")
if not dxf_bytes:
raise CostruttoreError("DXF vuoto")
frame = _gray_u8(gray)
calibration_bytes = json.dumps({"px_per_mm": px_per_mm}).encode("utf-8")
dxf_sha = hashlib.sha256(dxf_bytes).hexdigest()
cal_sha = hashlib.sha256(calibration_bytes).hexdigest()
with tempfile.TemporaryDirectory() as tmp:
folder = Path(tmp)
dxf_path = folder / DXF_FILE
dxf_path.write_bytes(dxf_bytes)
(folder / CALIBRATION_FILE).write_bytes(calibration_bytes)
# DXF leggibile? (ezdxf solleva su file corrotto/non DXF)
from visionsuite.io.dxf import read_xld_dxf
try:
drawing = list(read_xld_dxf(str(dxf_path), sampling_distance=0.1,
scale=px_per_mm, flip_y=True))
except Exception as exc: # noqa: BLE001 - qualunque errore di lettura = DXF illeggibile
raise CostruttoreError(f"DXF illeggibile: {type(exc).__name__}: {exc}") from exc
if not drawing:
raise CostruttoreError("DXF illeggibile: nessuna entita' geometrica")
reference_pose = _locate_reference(frame, polarity)
anchors = _candidate_anchors(frame, dxf_path, px_per_mm)
if not anchors:
raise CostruttoreError("disegno non sovrapponibile al pezzo (ICP fallito)")
# Verita' = eseguire davvero la ricetta sul riferimento: vince il maggior points_in_pct
best: tuple[float, dict, RunOutcome] | None = None
for anchor in anchors:
graph = _graph(dxf_sha, cal_sha, band_mm, min_points_pct, polarity, reference_pose, anchor)
outcome = _run(frame, graph, px_per_mm, folder)
pct = outcome.outputs.get("dxf.points_in_pct")
if pct is not None and (best is None or pct > best[0]):
best = (pct, graph, outcome)
if best is None:
raise CostruttoreError(
"disegno non sovrapponibile al pezzo: il confronto non ha prodotto punti"
)
_, graph, outcome = best
# solo le quattro uscite del contratto (n_points resta fuori)
outputs = {k: v for k, v in outcome.outputs.items() if k in _OUTPUT_KEYS}
return {
"graph": graph,
"files": {DXF_FILE: _b64(dxf_bytes), CALIBRATION_FILE: _b64(calibration_bytes)},
"outputs": outputs,
"overlay": outcome.overlay,
"outcome": outcome.outcome,
}
+114
View File
@@ -0,0 +1,114 @@
"""Costruttore di ricette dal campo: immagine + DXF + scala -> ricetta v2."""
import pytest
# Come test_runner.py: l'albero vision richiede Python >=3.13; su 3.11 e' skipped.
pytest.importorskip(
"visionsuite.task",
reason=(
"the vision tree requires Python >=3.13; run it with "
"`uv run --python 3.13 --extra vision --extra dev pytest src/vision/tests`"
),
)
import base64
import json
from pathlib import Path
import cv2
import numpy as np
from src.vision.costruttori import (
CALIBRATION_FILE,
DXF_FILE,
CostruttoreError,
build_dxf_compare,
)
from src.vision.runner import run_graph
SAMPLE_DXF = (
Path(__file__).resolve().parents[3]
/ "vendor/visionsuite/packages/vs-dxf/tests/fixtures/sample.dxf"
)
DXF_WIDTH_MM, DXF_HEIGHT_MM, DXF_HOLE_R_MM = 200.0, 100.0, 10.0
PX_PER_MM = 4.0
def a_scene(angle_deg: float = 0.0, shift_px: tuple[float, float] = (0.0, 0.0),
margin_mm: float = 60.0) -> np.ndarray:
"""Il pezzo del sample.dxf (scuro su bianco) a 4 px/mm, ruotato e traslato."""
w = int((DXF_WIDTH_MM + 2 * margin_mm) * PX_PER_MM)
h = int((DXF_HEIGHT_MM + 2 * margin_mm) * PX_PER_MM)
img = np.full((h, w), 255, np.uint8)
cx, cy = w // 2, h // 2
hw, hh = int(DXF_WIDTH_MM * PX_PER_MM / 2), int(DXF_HEIGHT_MM * PX_PER_MM / 2)
cv2.rectangle(img, (cx - hw, cy - hh), (cx + hw, cy + hh), 0, -1)
cv2.circle(img, (cx, cy), int(DXF_HOLE_R_MM * PX_PER_MM), 255, -1)
if angle_deg or any(shift_px):
m = cv2.getRotationMatrix2D((cx, cy), angle_deg, 1.0) # >0 = antiorario a video
m[0, 2] += shift_px[0]
m[1, 2] += shift_px[1]
img = cv2.warpAffine(img, m, (w, h), flags=cv2.INTER_LINEAR, borderValue=255)
return img
# ------------------------------------------------------------------ ricetta DXF
def test_ricetta_dxf_gira_su_riferimento_e_su_scena_ruotata(tmp_path):
dxf = SAMPLE_DXF.read_bytes()
risultato = build_dxf_compare(a_scene(0.0), dxf, px_per_mm=PX_PER_MM, band_mm=2.0)
assert risultato["outputs"]["dxf.points_in_pct"] > 95.0 # sul riferimento
assert set(risultato["outputs"]) == {
"dxf.points_in_pct", "dxf.distance_avg", "dxf.distance_min", "dxf.distance_max"}
assert set(risultato["files"]) == {DXF_FILE, CALIBRATION_FILE}
assert json.loads(base64.b64decode(risultato["files"][CALIBRATION_FILE])) == {
"px_per_mm": PX_PER_MM}
assert risultato["overlay"] and risultato["outcome"] is not None
graph = risultato["graph"]
assert graph["schema_version"] == 2
# path come da contratto; lo sha256 lo esige vs-task per ogni file in meta
assert graph["meta"]["dxf"]["disegno"]["path"] == "disegno.dxf"
assert graph["meta"]["calibration"]["path"] == "calibrazione.json"
assert len(graph["meta"]["dxf"]["disegno"]["sha256"]) == 64
pezzo, dxf_tool = graph["tools"]
assert (pezzo["id"], pezzo["kind"], pezzo["name"]) == ("pezzo", "LocateBlob", "Pezzo")
assert (dxf_tool["id"], dxf_tool["kind"], dxf_tool["name"]) == ("dxf", "CompareDxf", "Confronto DXF")
assert dxf_tool["links"] == {"align": {"tool": "pezzo", "property": "poses"}}
# La cartella ricetta come la scriverebbe il backend, poi la scena ruotata di 5 gradi
for nome, b64 in risultato["files"].items():
(tmp_path / nome).write_bytes(base64.b64decode(b64))
ruotata = a_scene(5.0, shift_px=(30.0, 20.0))
esito = run_graph(ruotata, graph, calibration={"px_per_mm": PX_PER_MM}, recipe_dir=tmp_path)
assert not esito.failures, esito.failures
assert esito.outputs["dxf.points_in_pct"] > 95.0
def test_ricetta_dxf_con_riferimento_gia_ruotato(tmp_path):
# il riferimento non e' a 0 gradi: la ricetta si costruisce lo stesso
risultato = build_dxf_compare(a_scene(20.0, shift_px=(-15.0, 10.0)),
SAMPLE_DXF.read_bytes(), PX_PER_MM, band_mm=2.0)
assert risultato["outputs"]["dxf.points_in_pct"] > 95.0
def test_immagine_senza_pezzo_solleva():
bianca = np.full((600, 800), 255, np.uint8)
with pytest.raises(CostruttoreError, match="pezzo"):
build_dxf_compare(bianca, SAMPLE_DXF.read_bytes(), PX_PER_MM)
def test_dxf_illeggibile_solleva():
with pytest.raises(CostruttoreError, match="DXF"):
build_dxf_compare(a_scene(), b"questo non e' un dxf", PX_PER_MM)
@pytest.mark.parametrize("kwargs", [
{"px_per_mm": 0.0}, {"px_per_mm": PX_PER_MM, "band_mm": 0.0},
{"px_per_mm": PX_PER_MM, "min_points_pct": 0.0},
{"px_per_mm": PX_PER_MM, "min_points_pct": 101.0},
{"px_per_mm": PX_PER_MM, "polarity": "grigio"},
])
def test_parametri_ricetta_invalidi(kwargs):
with pytest.raises(CostruttoreError):
build_dxf_compare(a_scene(), SAMPLE_DXF.read_bytes(), **kwargs)
+65
View File
@@ -17,6 +17,7 @@ import numpy as np
from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
from PIL import Image, UnidentifiedImageError
from src.vision.costruttori import CostruttoreError, build_dxf_compare # #83
from src.vision.runner import (
GraphBuildError,
RunOutcome,
@@ -223,3 +224,67 @@ async def validate(
validate_graph, parsed_graph, resolved_dir
)
return {"ok": ok, "errors": errors, "outputs": outputs}
def _decode_and_build_dxf(content: bytes, dxf: bytes, px_per_mm: float,
band_mm: float, min_points_pct: float,
polarity: str) -> dict:
"""#83: meta' sincrona di /recipes/dxf-compare (decodifica + costruzione),
in un thread come `_decode_and_run`."""
try:
frame = np.array(Image.open(io.BytesIO(content)).convert("L"))
except (UnidentifiedImageError, OSError) as exc:
raise _ImageDecodeError(str(exc)) from exc
return build_dxf_compare(frame, dxf, px_per_mm, band_mm, min_points_pct, polarity)
@app.post("/recipes/dxf-compare")
async def recipes_dxf_compare(
image: UploadFile = File(...),
dxf: UploadFile = File(...),
px_per_mm: float = Form(...),
band_mm: float = Form(0.1),
min_points_pct: float = Form(95.0),
polarity: str = Form("dark"),
) -> dict:
"""#83: costruisce una ricetta DXF compare dall'immagine di riferimento, il
disegno e la scala. Stessi semaforo, timeout e mappa errori di /run."""
content = await image.read()
dxf_bytes = await dxf.read()
if _run_slots.locked():
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"worker busy: already running {_MAX_CONCURRENCY} job(s)",
)
async with _run_slots:
try:
return await asyncio.wait_for(
asyncio.to_thread(
_decode_and_build_dxf, content, dxf_bytes, px_per_mm,
band_mm, min_points_pct, polarity,
),
timeout=_RUN_TIMEOUT_S,
)
except asyncio.TimeoutError as exc:
raise HTTPException(
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
detail=f"vision run exceeded {_RUN_TIMEOUT_S}s",
) from exc
except _ImageDecodeError as exc:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail=f"image is not decodable: {exc}",
) from exc
except (CostruttoreError, GraphBuildError) as exc:
# pezzo non trovato, DXF illeggibile, parametri invalidi: input, non guasto
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
detail=str(exc),
) from exc
except Exception as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"recipe build failed: {type(exc).__name__}: {exc}",
) from exc
@@ -0,0 +1,82 @@
"""POST /recipes/dxf-compare del worker (#83): forma della risposta ed errori."""
import pytest
# Come test_worker.py: serve l'albero vision (Python >=3.13), altrimenti skipped.
pytest.importorskip(
"visionsuite.task",
reason="the vision tree requires Python >=3.13; see test_worker.py",
)
import base64
import io
import time
import numpy as np
from httpx import ASGITransport, AsyncClient
from PIL import Image
from src.vision.tests.test_costruttori import PX_PER_MM, SAMPLE_DXF, a_scene
from src.vision_worker.main import app
def _png(array: np.ndarray) -> bytes:
buffer = io.BytesIO()
Image.fromarray(array).save(buffer, format="PNG")
return buffer.getvalue()
async def _post(image: bytes, dxf: bytes, **data):
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://worker") as client:
return await client.post(
"/recipes/dxf-compare",
files={"image": ("riferimento.png", image, "image/png"),
"dxf": ("disegno.dxf", dxf, "application/octet-stream")},
data={"px_per_mm": str(PX_PER_MM), "band_mm": "2.0", **data},
)
@pytest.mark.asyncio
async def test_dxf_compare_200_con_le_chiavi_del_contratto():
inizio = time.monotonic()
response = await _post(_png(a_scene(0.0)), SAMPLE_DXF.read_bytes())
durata = time.monotonic() - inizio
assert response.status_code == 200, response.text
body = response.json()
assert {"graph", "files", "outputs", "overlay", "outcome"} <= set(body)
assert set(body["files"]) == {"disegno.dxf", "calibrazione.json"}
assert base64.b64decode(body["files"]["disegno.dxf"]) == SAMPLE_DXF.read_bytes()
assert set(body["outputs"]) == {
"dxf.points_in_pct", "dxf.distance_avg", "dxf.distance_min", "dxf.distance_max"}
assert body["outputs"]["dxf.points_in_pct"] > 95.0
assert body["graph"]["schema_version"] == 2
print(f"\nW2 sulla scena di prova: {durata:.2f}s") # riportato nel rapporto
@pytest.mark.asyncio
async def test_dxf_compare_immagine_senza_pezzo_e_422():
bianca = np.full((600, 800), 255, np.uint8)
response = await _post(_png(bianca), SAMPLE_DXF.read_bytes())
assert response.status_code == 422
assert "pezzo" in response.json()["detail"]
@pytest.mark.asyncio
async def test_dxf_compare_dxf_illeggibile_e_422():
response = await _post(_png(a_scene(0.0)), b"non e' un dxf")
assert response.status_code == 422
assert "DXF" in response.json()["detail"]
@pytest.mark.asyncio
async def test_dxf_compare_immagine_non_decodificabile_e_422():
response = await _post(b"non e' un'immagine", SAMPLE_DXF.read_bytes())
assert response.status_code == 422
assert "not decodable" in response.json()["detail"].lower()
@pytest.mark.asyncio
async def test_dxf_compare_parametro_fuori_range_e_422():
response = await _post(_png(a_scene(0.0)), SAMPLE_DXF.read_bytes(), polarity="grigio")
assert response.status_code == 422