Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Adriano
|
9f2d624b4f | ||
|
Adriano
|
9fe82d7cef |
@@ -1,458 +0,0 @@
|
|||||||
"""Seed per il PC della fiera: utenti, stazione, tre ricette e quattro settimane di misure.
|
|
||||||
|
|
||||||
Cosa fa
|
|
||||||
Popola un database TieMeasureFlow (SQLite del pacchetto Windows) cosi' che al primo
|
|
||||||
avvio NON compaia la procedura di setup e la demo abbia gia' qualcosa da mostrare:
|
|
||||||
|
|
||||||
- utenti adriano (admin), maker, capoturno, operatore (password nella costante
|
|
||||||
UTENTI qui sotto, decise da Adriano per la fiera);
|
|
||||||
- la stazione ST-DEFAULT (quella del pacchetto, vedi packaging/windows/avvio.py)
|
|
||||||
con camera_url http://127.0.0.1:8765 e camera_source camera:601020 (Itala);
|
|
||||||
- FIERA-FLANGIA «Flangia Ø80» (5 quote) e FIERA-ALBERO «Albero rettificato»
|
|
||||||
(4 quote), con circa 4 settimane di storico fino a ieri (lun-ven 8-17, una
|
|
||||||
produzione chiusa per giorno, lotti e seriali coerenti, misure con Cpk diversi
|
|
||||||
per quota, una deriva lenta nell'ultima settimana, una quota con 1-3% fuori
|
|
||||||
tolleranza);
|
|
||||||
- FIERA-DXF «Profilo DXF (visione)»: un task dxf_compare senza ricetta di visione
|
|
||||||
(si crea in fiera dal pannello «Crea ricetta da DXF»).
|
|
||||||
|
|
||||||
«Setup fatto» per il backend vuol dire: esiste almeno un utente admin
|
|
||||||
(api/routers/setup.py::_admin_exists), quindi basta creare adriano.
|
|
||||||
|
|
||||||
Come si usa (da daitarn, TMF fermo sul PC, su una COPIA del file SQLite)
|
|
||||||
uv run python scripts/seed_fiera.py --db C:/percorso/tmf.db [--replace] [--dry-run]
|
|
||||||
[--fine AAAA-MM-GG]
|
|
||||||
|
|
||||||
Il file deve avere gia' lo schema (alembic upgrade head: lo fa l'exe al primo
|
|
||||||
avvio). --replace toglie solo i dati FIERA-* (ricette, misure, produzioni) e li
|
|
||||||
rifa'; senza, un secondo giro non duplica le ricette. Gli utenti e la stazione
|
|
||||||
sono idempotenti: se esistono si aggiornano password, ruoli e camera.
|
|
||||||
--dry-run fa tutto e annulla (stampa i numeri). Il seme casuale e' fisso: stesso
|
|
||||||
comando e stessa data di fine, stessi dati.
|
|
||||||
|
|
||||||
Da cosa dipende
|
|
||||||
src.backend (modelli ORM, auth_service, measurement_service), SQLAlchemy +
|
|
||||||
aiosqlite, tzdata su Windows (gia' in pyproject). Non usa MySQL.
|
|
||||||
"""
|
|
||||||
import argparse
|
|
||||||
import asyncio
|
|
||||||
import random
|
|
||||||
import sys
|
|
||||||
from datetime import date, datetime, time, timedelta, timezone
|
|
||||||
from pathlib import Path
|
|
||||||
from zoneinfo import ZoneInfo
|
|
||||||
|
|
||||||
from sqlalchemy import delete, event, select
|
|
||||||
from sqlalchemy.ext.asyncio import (
|
|
||||||
AsyncSession, async_sessionmaker, create_async_engine,
|
|
||||||
)
|
|
||||||
|
|
||||||
# La radice del repo nel path, cosi' "src.backend" si importa lanciando lo script da ovunque
|
|
||||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
||||||
|
|
||||||
from src.backend.database import async_session_factory # noqa: E402
|
|
||||||
from src.backend.models.orm.measurement import Measurement # noqa: E402
|
|
||||||
from src.backend.models.orm.production import ( # noqa: E402
|
|
||||||
ProductionEvent, ProductionRun,
|
|
||||||
)
|
|
||||||
from src.backend.models.orm.recipe import Recipe, RecipeVersion # noqa: E402
|
|
||||||
from src.backend.models.orm.setting import RecipeVersionAudit # noqa: E402
|
|
||||||
from src.backend.models.orm.station import ( # noqa: E402
|
|
||||||
Station, StationRecipeAssignment,
|
|
||||||
)
|
|
||||||
from src.backend.models.orm.task import RecipeSubtask, RecipeTask # noqa: E402
|
|
||||||
from src.backend.models.orm.task_result import TaskResult # noqa: E402
|
|
||||||
from src.backend.models.orm.user import User # noqa: E402
|
|
||||||
from src.backend.services import auth_service # noqa: E402
|
|
||||||
from src.backend.services.measurement_service import calculate_pass_fail # noqa: E402
|
|
||||||
|
|
||||||
CODES = ("FIERA-FLANGIA", "FIERA-ALBERO", "FIERA-DXF")
|
|
||||||
STATION_CODE = "ST-DEFAULT" # STATION_CODE del pacchetto Windows (avvio.py PREDEFINITI)
|
|
||||||
CAMERA_URL = "http://127.0.0.1:8765"
|
|
||||||
CAMERA_SOURCE = "camera:601020" # Itala sul PC fiera
|
|
||||||
SEED = 20261026 # seme fisso: dati ripetibili
|
|
||||||
FUSO = ZoneInfo("Europe/Rome") # le misure sono salvate in UTC naive, l'orario di turno e' locale
|
|
||||||
GIORNI_STORICO = 28 # circa 4 settimane
|
|
||||||
|
|
||||||
# (username, password, nome, ruoli, admin). Password "123" per tutti, chiesta da Adriano per la fiera (#85); il login non controlla la lunghezza.
|
|
||||||
UTENTI = [
|
|
||||||
("adriano", "123", "Adriano Dal Pastro",
|
|
||||||
["Maker", "MeasurementTec", "Metrologist", "Supervisor"], True),
|
|
||||||
("maker", "123", "Marco Ferrari (Maker)", ["Maker"], False),
|
|
||||||
("capoturno", "123", "Luca Bianchi (Capoturno)",
|
|
||||||
["Supervisor", "MeasurementTec"], False),
|
|
||||||
("operatore", "123", "Giulia Rossi (Operatore)", ["MeasurementTec"], False),
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
def _quota(marker, descrizione, nominale, tol, unita="mm", dec=3):
|
|
||||||
"""Quota simmetrica: tolleranza = nominale +/- tol, warning al 70% della tolleranza."""
|
|
||||||
return dict(
|
|
||||||
marker_number=marker, description=descrizione, nominal=nominale,
|
|
||||||
ltl=round(nominale - tol, dec), lwl=round(nominale - 0.7 * tol, dec),
|
|
||||||
uwl=round(nominale + 0.7 * tol, dec), utl=round(nominale + tol, dec),
|
|
||||||
unit=unita,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
# Profilo statistico di una quota: scostamento medio (in frazioni di tol), sigma (in
|
|
||||||
# frazioni di tol) e deriva (frazione di tol aggiunta linearmente nell'ultima settimana).
|
|
||||||
# Cpk ~ (1 - |scostamento|) / (3 * sigma): sigma 0.2 -> 1,67; 0.33 -> 1,0.
|
|
||||||
FLANGIA = [
|
|
||||||
# (task, quota, scostamento, sigma, deriva)
|
|
||||||
(0, _quota(1, "Diametro esterno Ø80", 80.0, 0.05), 0.0, 0.20, 0.0), # Cpk ~1,67 capace
|
|
||||||
(0, _quota(2, "Diametro foro centrale Ø25", 25.0, 0.02), 0.0, 0.33, 0.0), # Cpk ~1,0 al limite
|
|
||||||
(1, _quota(3, "Spessore flangia", 12.0, 0.10), 0.0, 0.22, 0.55), # deriva ultima settimana
|
|
||||||
(1, _quota(4, "Interasse fori Ø60", 60.0, 0.10), 0.10, 0.25, 0.0), # Cpk ~1,2
|
|
||||||
(1, _quota(5, "Profondita' lamatura", 8.0, 0.15), 0.50, 0.244, 0.0), # ~2% fuori tolleranza
|
|
||||||
]
|
|
||||||
ALBERO = [
|
|
||||||
(0, _quota(1, "Diametro cuscinetto Ø20 h6", 20.0, 0.010), 0.0, 0.20, 0.0), # Cpk ~1,67
|
|
||||||
(0, _quota(2, "Diametro sede Ø25 j5", 25.0, 0.008), 0.0, 0.25, 0.0), # Cpk ~1,33
|
|
||||||
(1, _quota(3, "Lunghezza totale", 150.0, 0.10), 0.0, 0.33, 0.0), # Cpk ~1,0
|
|
||||||
(1, _quota(4, "Rugosita' Ra", 0.40, 0.20, unita="µm"), 0.20, 0.25, 0.0), # Cpk ~1,07
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
def _ora_utc(giorno: date, minuti_da_mezzanotte: int) -> datetime:
|
|
||||||
"""Ora locale di officina -> datetime UTC naive, come le salva il backend."""
|
|
||||||
locale = datetime.combine(giorno, time(0, 0), tzinfo=FUSO) + timedelta(minutes=minuti_da_mezzanotte)
|
|
||||||
return locale.astimezone(timezone.utc).replace(tzinfo=None)
|
|
||||||
|
|
||||||
|
|
||||||
def _giorni_lavorativi(fine: date) -> list[date]:
|
|
||||||
"""Lun-ven degli ultimi GIORNI_STORICO giorni, fino a `fine` compreso."""
|
|
||||||
giorni = [fine - timedelta(days=i) for i in range(GIORNI_STORICO)]
|
|
||||||
return sorted(g for g in giorni if g.weekday() < 5)
|
|
||||||
|
|
||||||
|
|
||||||
def _slot_giornalieri(rng: random.Random, n: int) -> list[int]:
|
|
||||||
"""n orari (minuti dalla mezzanotte) in 8-12 e 13-17, ogni ~30 min con un po' di scarto."""
|
|
||||||
mezz_ore = [8 * 60 + 30 * i for i in range(8)] + [13 * 60 + 30 * i for i in range(8)]
|
|
||||||
scelti = sorted(rng.sample(mezz_ore, n))
|
|
||||||
return [m + rng.randint(0, 9) for m in scelti]
|
|
||||||
|
|
||||||
|
|
||||||
async def _upsert_utenti(session: AsyncSession) -> dict[str, User]:
|
|
||||||
"""Crea gli utenti o, se esistono, ne aggiorna password, nome, ruoli e flag admin."""
|
|
||||||
utenti = {}
|
|
||||||
for username, password, nome, ruoli, admin in UTENTI:
|
|
||||||
esistente = (await session.execute(
|
|
||||||
select(User).where(User.username == username)
|
|
||||||
)).scalar_one_or_none()
|
|
||||||
if esistente is None:
|
|
||||||
utenti[username] = await auth_service.create_user(
|
|
||||||
session, username=username, password=password, display_name=nome,
|
|
||||||
roles=ruoli, is_admin=admin,
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
esistente.password_hash = auth_service.hash_password(password)
|
|
||||||
esistente.display_name = nome
|
|
||||||
esistente.roles = ruoli
|
|
||||||
esistente.is_admin = admin
|
|
||||||
esistente.active = True
|
|
||||||
utenti[username] = esistente
|
|
||||||
await session.flush()
|
|
||||||
return utenti
|
|
||||||
|
|
||||||
|
|
||||||
async def _upsert_stazione(session: AsyncSession, owner: User) -> Station:
|
|
||||||
"""La stazione del pacchetto, con la camera Itala; se c'e' gia' ne aggiorna la camera."""
|
|
||||||
stazione = (await session.execute(
|
|
||||||
select(Station).where(Station.code == STATION_CODE)
|
|
||||||
)).scalar_one_or_none()
|
|
||||||
if stazione is None:
|
|
||||||
stazione = Station(
|
|
||||||
code=STATION_CODE, name="Postazione fiera", location="Stand",
|
|
||||||
created_by=owner.id,
|
|
||||||
)
|
|
||||||
session.add(stazione)
|
|
||||||
stazione.camera_url = CAMERA_URL
|
|
||||||
stazione.camera_source = CAMERA_SOURCE
|
|
||||||
await session.flush()
|
|
||||||
return stazione
|
|
||||||
|
|
||||||
|
|
||||||
async def _conta_e_cancella(session: AsyncSession, ricette: list[Recipe]) -> dict:
|
|
||||||
"""Toglie le ricette FIERA-* e tutto cio' che e' misurato contro di esse. Ritorna i conteggi."""
|
|
||||||
conti = {"misure": 0, "produzioni": 0, "eventi": 0}
|
|
||||||
for ricetta in ricette:
|
|
||||||
versioni = (await session.execute(
|
|
||||||
select(RecipeVersion.id).where(RecipeVersion.recipe_id == ricetta.id)
|
|
||||||
)).scalars().all()
|
|
||||||
runs = (await session.execute(
|
|
||||||
select(ProductionRun.id).where(ProductionRun.recipe_id == ricetta.id)
|
|
||||||
)).scalars().all()
|
|
||||||
if versioni:
|
|
||||||
conti["misure"] += (await session.execute(
|
|
||||||
delete(Measurement).where(Measurement.version_id.in_(versioni))
|
|
||||||
)).rowcount
|
|
||||||
await session.execute( # esiti dei task di queste versioni (FK sulle versioni)
|
|
||||||
delete(TaskResult).where(TaskResult.version_id.in_(versioni))
|
|
||||||
)
|
|
||||||
if runs:
|
|
||||||
conti["eventi"] += (await session.execute(
|
|
||||||
delete(ProductionEvent).where(ProductionEvent.run_id.in_(runs))
|
|
||||||
)).rowcount
|
|
||||||
conti["produzioni"] += (await session.execute(
|
|
||||||
delete(ProductionRun).where(ProductionRun.id.in_(runs))
|
|
||||||
)).rowcount
|
|
||||||
await session.execute(
|
|
||||||
delete(StationRecipeAssignment).where(StationRecipeAssignment.recipe_id == ricetta.id)
|
|
||||||
)
|
|
||||||
await session.execute(
|
|
||||||
delete(RecipeVersionAudit).where(RecipeVersionAudit.recipe_id == ricetta.id)
|
|
||||||
)
|
|
||||||
# Versioni, task e quote se ne vanno con la cascade ORM della ricetta.
|
|
||||||
await session.delete(ricetta)
|
|
||||||
await session.flush()
|
|
||||||
return conti
|
|
||||||
|
|
||||||
|
|
||||||
async def _nuova_ricetta(session, owner, stazione, *, code, name, description,
|
|
||||||
interval, requires_serial, note_versione):
|
|
||||||
"""Ricetta + versione 1 corrente + assegnazione alla stazione."""
|
|
||||||
ricetta = Recipe(
|
|
||||||
code=code, name=name, description=description, created_by=owner.id, active=True,
|
|
||||||
measurement_interval_minutes=interval, requires_lot=True,
|
|
||||||
requires_serial=requires_serial, allow_manual_input=True,
|
|
||||||
)
|
|
||||||
session.add(ricetta)
|
|
||||||
await session.flush()
|
|
||||||
versione = RecipeVersion(
|
|
||||||
recipe_id=ricetta.id, version_number=1, is_current=True,
|
|
||||||
created_by=owner.id, change_notes=note_versione,
|
|
||||||
)
|
|
||||||
session.add(versione)
|
|
||||||
session.add(StationRecipeAssignment(
|
|
||||||
station_id=stazione.id, recipe_id=ricetta.id, assigned_by=owner.id,
|
|
||||||
))
|
|
||||||
await session.flush()
|
|
||||||
return ricetta, versione
|
|
||||||
|
|
||||||
|
|
||||||
async def _task(session, versione, ordine, titolo, tipo, direttiva=None,
|
|
||||||
descrizione=None, quote=()):
|
|
||||||
task = RecipeTask(
|
|
||||||
version_id=versione.id, order_index=ordine, title=titolo, task_type=tipo,
|
|
||||||
directive=direttiva, description=descrizione,
|
|
||||||
)
|
|
||||||
session.add(task)
|
|
||||||
await session.flush()
|
|
||||||
sottotask = []
|
|
||||||
for q in quote:
|
|
||||||
st = RecipeSubtask(task_id=task.id, **q)
|
|
||||||
session.add(st)
|
|
||||||
sottotask.append(st)
|
|
||||||
await session.flush()
|
|
||||||
return task, sottotask
|
|
||||||
|
|
||||||
|
|
||||||
async def _ricetta_meccanica(session, owner, stazione, *, code, name, description,
|
|
||||||
interval, requires_serial, profilo, titoli_task):
|
|
||||||
"""Ricetta con una nota iniziale e un task di misura per ogni gruppo di quote del profilo."""
|
|
||||||
ricetta, versione = await _nuova_ricetta(
|
|
||||||
session, owner, stazione, code=code, name=name, description=description,
|
|
||||||
interval=interval, requires_serial=requires_serial,
|
|
||||||
note_versione="Ricetta dimostrativa per la fiera",
|
|
||||||
)
|
|
||||||
await _task(session, versione, 0, "Preparazione del pezzo", "note",
|
|
||||||
direttiva="Prima di misurare",
|
|
||||||
descrizione="Pulire il pezzo e controllare che sia a **temperatura ambiente**.")
|
|
||||||
sottotask_per_quota = {}
|
|
||||||
for indice, titolo in enumerate(titoli_task):
|
|
||||||
quote = [q for (t, q, *_r) in profilo if t == indice]
|
|
||||||
_t, sottotask = await _task(
|
|
||||||
session, versione, indice + 1, titolo, "measure",
|
|
||||||
direttiva="Misurare le quote in sequenza", quote=quote,
|
|
||||||
)
|
|
||||||
for st in sottotask:
|
|
||||||
sottotask_per_quota[st.marker_number] = st
|
|
||||||
return ricetta, versione, sottotask_per_quota
|
|
||||||
|
|
||||||
|
|
||||||
def _valore(rng, quota, scost, sigma, deriva, frazione_tempo):
|
|
||||||
"""Un valore: normale attorno a nominale + scostamento (+ deriva nell'ultima settimana)."""
|
|
||||||
tol = quota["utl"] - quota["nominal"]
|
|
||||||
media = quota["nominal"] + tol * (scost + deriva * max(0.0, (frazione_tempo - 0.75) / 0.25))
|
|
||||||
return round(rng.gauss(media, tol * sigma), 6)
|
|
||||||
|
|
||||||
|
|
||||||
async def _storico(session, rng, fine, owner_ricetta, stazione, utenti, ricetta, versione,
|
|
||||||
sottotask, profilo, pezzi_giorno, prefisso_lotto, prefisso_seriale):
|
|
||||||
"""Una produzione chiusa per giorno lavorativo, con i suoi pezzi, eventi e misure."""
|
|
||||||
giorni = _giorni_lavorativi(fine)
|
|
||||||
operatore, capoturno = utenti["operatore"], utenti["capoturno"]
|
|
||||||
task_ids = {} # marker -> id del task di misura
|
|
||||||
for marker, st in sottotask.items():
|
|
||||||
task_ids[marker] = st.task_id
|
|
||||||
ordine_task = sorted(set(task_ids.values()))
|
|
||||||
ultimo_task = ordine_task[-1]
|
|
||||||
contatore = 0
|
|
||||||
totale = 0
|
|
||||||
for indice_giorno, giorno in enumerate(giorni):
|
|
||||||
frazione = indice_giorno / max(1, len(giorni) - 1) # 0 = primo giorno, 1 = ieri
|
|
||||||
n = rng.randint(*pezzi_giorno)
|
|
||||||
slot = _slot_giornalieri(rng, n)
|
|
||||||
lotto = f"{prefisso_lotto}{giorno:%y%m%d}"
|
|
||||||
inizio, fine_run = _ora_utc(giorno, slot[0] - 5), _ora_utc(giorno, slot[-1] + 10)
|
|
||||||
run = ProductionRun(
|
|
||||||
station_id=stazione.id, recipe_id=ricetta.id, version_id=versione.id,
|
|
||||||
operator_id=operatore.id, lot_number=lotto, status="closed",
|
|
||||||
measurement_interval_minutes=ricetta.measurement_interval_minutes,
|
|
||||||
cycle_count=n, started_at=inizio, closed_at=fine_run, closed_by=capoturno.id,
|
|
||||||
active_station_id=None,
|
|
||||||
)
|
|
||||||
session.add(run)
|
|
||||||
await session.flush()
|
|
||||||
session.add(ProductionEvent(run_id=run.id, event_type="start", user_id=operatore.id,
|
|
||||||
created_at=inizio))
|
|
||||||
for numero, minuti in enumerate(slot, start=1):
|
|
||||||
contatore += 1
|
|
||||||
seriale = f"{prefisso_seriale}{contatore:05d}"
|
|
||||||
istante = _ora_utc(giorno, minuti)
|
|
||||||
for marker in sorted(sottotask):
|
|
||||||
st = sottotask[marker]
|
|
||||||
_t, quota, scost, sigma, deriva = next(p for p in profilo if p[1]["marker_number"] == marker)
|
|
||||||
valore = _valore(rng, quota, scost, sigma, deriva, frazione)
|
|
||||||
esito, scarto = calculate_pass_fail(valore, st)
|
|
||||||
m = Measurement(
|
|
||||||
subtask_id=st.id, version_id=versione.id, measured_by=operatore.id,
|
|
||||||
value=valore, pass_fail=esito, deviation=scarto, lot_number=lotto,
|
|
||||||
serial_number=seriale,
|
|
||||||
input_method="usb_caliper" if rng.random() < 0.75 else "manual",
|
|
||||||
input_duration_ms=rng.randint(2500, 9000),
|
|
||||||
production_run_id=run.id,
|
|
||||||
# un fuori tolleranza passa solo con l'ok del capoturno (come in produzione)
|
|
||||||
supervisor_id=capoturno.id if esito == "fail" else None,
|
|
||||||
authorised_at=istante + timedelta(minutes=1) if esito == "fail" else None,
|
|
||||||
measured_at=istante + timedelta(seconds=20 * marker),
|
|
||||||
)
|
|
||||||
session.add(m)
|
|
||||||
totale += 1
|
|
||||||
# un task_measured per ogni task di misura non finale, poi il ciclo completato
|
|
||||||
for tid in ordine_task:
|
|
||||||
tipo = "cycle_completed" if tid == ultimo_task else "task_measured"
|
|
||||||
session.add(ProductionEvent(
|
|
||||||
run_id=run.id, event_type=tipo, user_id=operatore.id, task_id=tid,
|
|
||||||
created_at=istante + timedelta(minutes=1),
|
|
||||||
))
|
|
||||||
session.add(ProductionEvent(run_id=run.id, event_type="close", user_id=capoturno.id,
|
|
||||||
supervisor_id=capoturno.id, created_at=fine_run))
|
|
||||||
await session.flush()
|
|
||||||
return contatore, totale
|
|
||||||
|
|
||||||
|
|
||||||
async def _ricetta_dxf(session, owner, stazione):
|
|
||||||
"""Demo visione: un task dxf_compare, senza grafo (si crea in fiera dal pannello DXF)."""
|
|
||||||
ricetta, versione = await _nuova_ricetta(
|
|
||||||
session, owner, stazione, code="FIERA-DXF", name="Profilo DXF (visione)",
|
|
||||||
description="Confronto di un profilo con il disegno DXF tramite la camera. "
|
|
||||||
"La ricetta di visione si crea in fiera dal pannello «Crea ricetta da DXF».",
|
|
||||||
interval=None, requires_serial=False, note_versione="Ricetta dimostrativa per la fiera",
|
|
||||||
)
|
|
||||||
await _task(session, versione, 0, "Preparazione del pezzo", "note",
|
|
||||||
direttiva="Prima di misurare",
|
|
||||||
descrizione="Appoggiare il pezzo sul piano della camera, **in piano**.")
|
|
||||||
await _task(
|
|
||||||
session, versione, 1, "Confronto profilo con DXF", "dxf_compare",
|
|
||||||
direttiva="Acquisire e confrontare",
|
|
||||||
descrizione="Lo scostamento massimo del profilo rispetto al DXF deve restare entro la tolleranza.",
|
|
||||||
quote=[_quota(1, "Scostamento massimo profilo", 0.0, 0.15)],
|
|
||||||
)
|
|
||||||
return ricetta
|
|
||||||
|
|
||||||
|
|
||||||
async def seed(replace: bool, dry_run: bool, fine: date) -> int:
|
|
||||||
"""Esegue il seed in una sola transazione; con dry_run annulla tutto alla fine."""
|
|
||||||
rng = random.Random(SEED)
|
|
||||||
async with async_session_factory() as session:
|
|
||||||
utenti = await _upsert_utenti(session)
|
|
||||||
owner = utenti["adriano"]
|
|
||||||
stazione = await _upsert_stazione(session, owner)
|
|
||||||
|
|
||||||
esistenti = list((await session.execute(
|
|
||||||
select(Recipe).where(Recipe.code.in_(CODES))
|
|
||||||
)).scalars().all())
|
|
||||||
if esistenti and not replace:
|
|
||||||
print(f"Ricette FIERA gia' presenti ({', '.join(r.code for r in esistenti)}): "
|
|
||||||
"non duplico. Usa --replace per rifarle.")
|
|
||||||
else:
|
|
||||||
if esistenti:
|
|
||||||
conti = await _conta_e_cancella(session, esistenti)
|
|
||||||
print(f"--replace: rimosse {', '.join(r.code for r in esistenti)} "
|
|
||||||
f"({conti['misure']} misure, {conti['produzioni']} produzioni, "
|
|
||||||
f"{conti['eventi']} eventi)")
|
|
||||||
|
|
||||||
r1, v1, sub1 = await _ricetta_meccanica(
|
|
||||||
session, owner, stazione, code="FIERA-FLANGIA", name="Flangia Ø80",
|
|
||||||
description="Flangia tornita in alluminio: diametri, spessore, interasse fori "
|
|
||||||
"e profondita' della lamatura. Lotto e seriale obbligatori.",
|
|
||||||
interval=30, requires_serial=True, profilo=FLANGIA,
|
|
||||||
titoli_task=["Misure diametrali", "Spessore e posizione fori"],
|
|
||||||
)
|
|
||||||
n1, m1 = await _storico(session, rng, fine, owner, stazione, utenti, r1, v1, sub1,
|
|
||||||
FLANGIA, (10, 14), "LF", "FL")
|
|
||||||
r2, v2, sub2 = await _ricetta_meccanica(
|
|
||||||
session, owner, stazione, code="FIERA-ALBERO", name="Albero rettificato",
|
|
||||||
description="Albero in acciaio rettificato: sedi dei cuscinetti, lunghezza e "
|
|
||||||
"rugosita'. Tolleranze strette.",
|
|
||||||
interval=30, requires_serial=False, profilo=ALBERO,
|
|
||||||
titoli_task=["Diametri rettificati", "Lunghezza e rugosita'"],
|
|
||||||
)
|
|
||||||
n2, m2 = await _storico(session, rng, fine, owner, stazione, utenti, r2, v2, sub2,
|
|
||||||
ALBERO, (8, 11), "LA", "AL")
|
|
||||||
await _ricetta_dxf(session, owner, stazione)
|
|
||||||
print(f"FIERA-FLANGIA: {n1} pezzi, {m1} misure; FIERA-ALBERO: {n2} pezzi, {m2} misure")
|
|
||||||
|
|
||||||
if dry_run:
|
|
||||||
await session.rollback()
|
|
||||||
print("--dry-run: nessuna modifica salvata.")
|
|
||||||
else:
|
|
||||||
await session.commit()
|
|
||||||
print(f"Seed fiera completato (stazione {stazione.code}, utenti: "
|
|
||||||
f"{', '.join(u[0] for u in UTENTI)}).")
|
|
||||||
return 0
|
|
||||||
|
|
||||||
|
|
||||||
def _factory_per_file(db: Path):
|
|
||||||
"""Sessioni async su un file SQLite qualsiasi, con le stesse PRAGMA dell'app (FK attive)."""
|
|
||||||
engine = create_async_engine(f"sqlite+aiosqlite:///{db.resolve().as_posix()}", echo=False)
|
|
||||||
|
|
||||||
@event.listens_for(engine.sync_engine, "connect")
|
|
||||||
def _pragma(dbapi_connection, _record) -> None:
|
|
||||||
cur = dbapi_connection.cursor()
|
|
||||||
cur.execute("PRAGMA foreign_keys=ON")
|
|
||||||
cur.execute("PRAGMA busy_timeout=5000")
|
|
||||||
cur.close()
|
|
||||||
|
|
||||||
return engine, async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
||||||
|
|
||||||
|
|
||||||
async def _main_async(args) -> int:
|
|
||||||
global async_session_factory
|
|
||||||
engine, async_session_factory = _factory_per_file(args.db)
|
|
||||||
try:
|
|
||||||
fine = date.fromisoformat(args.fine) if args.fine else date.today() - timedelta(days=1)
|
|
||||||
return await seed(args.replace, args.dry_run, fine)
|
|
||||||
finally:
|
|
||||||
await engine.dispose()
|
|
||||||
|
|
||||||
|
|
||||||
def main() -> int:
|
|
||||||
parser = argparse.ArgumentParser(description=__doc__,
|
|
||||||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
||||||
parser.add_argument("--db", required=True, type=Path, help="file SQLite con lo schema gia' creato")
|
|
||||||
parser.add_argument("--replace", action="store_true",
|
|
||||||
help="rimuove i dati FIERA-* (ricette, misure, produzioni) e li rifa'")
|
|
||||||
parser.add_argument("--dry-run", action="store_true", help="esegue tutto e annulla")
|
|
||||||
parser.add_argument("--fine", default=None, help="ultimo giorno dello storico AAAA-MM-GG (default: ieri)")
|
|
||||||
args = parser.parse_args()
|
|
||||||
if not args.db.is_file():
|
|
||||||
print(f"File non trovato: {args.db}")
|
|
||||||
return 2
|
|
||||||
return asyncio.run(_main_async(args))
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
raise SystemExit(main())
|
|
||||||
@@ -1,194 +0,0 @@
|
|||||||
"""Test di scripts/seed_fiera.py (issue #85): il seed del PC della fiera.
|
|
||||||
|
|
||||||
Lo script non e' un modulo del pacchetto: si carica da file con importlib, come
|
|
||||||
test_seed_collaudo. La parte "dato" gira sul DB in memoria dei test; l'ultimo test
|
|
||||||
percorre invece la strada vera (file SQLite con `alembic upgrade head` + --db).
|
|
||||||
"""
|
|
||||||
import importlib.util
|
|
||||||
import sqlite3
|
|
||||||
from argparse import Namespace
|
|
||||||
from datetime import date
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
from alembic import command
|
|
||||||
from alembic.config import Config
|
|
||||||
from sqlalchemy import func, select
|
|
||||||
|
|
||||||
from src.backend.config import Settings, settings
|
|
||||||
from src.backend.models.orm.measurement import Measurement
|
|
||||||
from src.backend.models.orm.recipe import Recipe
|
|
||||||
from src.backend.models.orm.station import Station, StationRecipeAssignment
|
|
||||||
from src.backend.models.orm.user import User
|
|
||||||
from src.backend.services.auth_service import authenticate_user
|
|
||||||
from src.backend.tests.conftest import TestSessionFactory, test_engine
|
|
||||||
|
|
||||||
_SCRIPT_PATH = Path(__file__).resolve().parents[3] / "scripts" / "seed_fiera.py"
|
|
||||||
_ALEMBIC_INI = Path(__file__).resolve().parents[1] / "migrations" / "alembic.ini"
|
|
||||||
FINE = date(2026, 10, 22) # data fissa: il test non dipende dal giorno in cui gira
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def seed_fiera(monkeypatch):
|
|
||||||
spec = importlib.util.spec_from_file_location("seed_fiera", _SCRIPT_PATH)
|
|
||||||
module = importlib.util.module_from_spec(spec)
|
|
||||||
spec.loader.exec_module(module)
|
|
||||||
monkeypatch.setattr(module, "async_session_factory", TestSessionFactory)
|
|
||||||
return module
|
|
||||||
|
|
||||||
|
|
||||||
async def _utente(db_session, username) -> User:
|
|
||||||
return (await db_session.execute(
|
|
||||||
select(User).where(User.username == username)
|
|
||||||
)).scalar_one()
|
|
||||||
|
|
||||||
|
|
||||||
class TestSeedFiera:
|
|
||||||
async def test_utenti_ruoli_e_login(self, seed_fiera, db_session):
|
|
||||||
assert await seed_fiera.seed(False, False, FINE) == 0
|
|
||||||
|
|
||||||
attesi = {
|
|
||||||
"adriano": ("123", True),
|
|
||||||
"maker": ("123", False),
|
|
||||||
"capoturno": ("123", False),
|
|
||||||
"operatore": ("123", False),
|
|
||||||
}
|
|
||||||
for username, (password, admin) in attesi.items():
|
|
||||||
utente = await authenticate_user(db_session, username, password)
|
|
||||||
assert utente is not None, username
|
|
||||||
assert utente.is_admin is admin
|
|
||||||
assert (await _utente(db_session, "maker")).roles == ["Maker"]
|
|
||||||
assert "Supervisor" in (await _utente(db_session, "capoturno")).roles
|
|
||||||
assert (await _utente(db_session, "operatore")).roles == ["MeasurementTec"]
|
|
||||||
assert {"Maker", "Metrologist"} <= set((await _utente(db_session, "adriano")).roles)
|
|
||||||
|
|
||||||
async def test_setup_risulta_completato(self, seed_fiera, client, monkeypatch):
|
|
||||||
import src.backend.api.routers.setup as setup_mod
|
|
||||||
monkeypatch.setattr(settings, "setup_password", "test-setup-password")
|
|
||||||
monkeypatch.setattr(setup_mod, "engine", test_engine)
|
|
||||||
monkeypatch.setattr(setup_mod, "async_session_factory", TestSessionFactory)
|
|
||||||
|
|
||||||
# prima del seed il frontend vede il setup da fare, dopo no
|
|
||||||
assert (await client.get("/api/setup/status")).json() == {"initialised": False}
|
|
||||||
await seed_fiera.seed(False, False, FINE)
|
|
||||||
assert (await client.get("/api/setup/status")).json() == {"initialised": True}
|
|
||||||
|
|
||||||
async def test_stazione_e_ricette(self, seed_fiera, db_session):
|
|
||||||
await seed_fiera.seed(False, False, FINE)
|
|
||||||
|
|
||||||
stazione = (await db_session.execute(
|
|
||||||
select(Station).where(Station.code == "ST-DEFAULT")
|
|
||||||
)).scalar_one()
|
|
||||||
assert stazione.camera_url == "http://127.0.0.1:8765"
|
|
||||||
assert stazione.camera_source == "camera:601020"
|
|
||||||
|
|
||||||
assegnate = (await db_session.execute(
|
|
||||||
select(Recipe.code)
|
|
||||||
.join(StationRecipeAssignment, StationRecipeAssignment.recipe_id == Recipe.id)
|
|
||||||
.where(StationRecipeAssignment.station_id == stazione.id)
|
|
||||||
)).scalars().all()
|
|
||||||
assert sorted(assegnate) == ["FIERA-ALBERO", "FIERA-DXF", "FIERA-FLANGIA"]
|
|
||||||
|
|
||||||
# la ricetta DXF ha un task dxf_compare e nessun grafo di visione
|
|
||||||
from src.backend.models.orm.task import RecipeTask
|
|
||||||
dxf = (await db_session.execute(
|
|
||||||
select(RecipeTask).where(RecipeTask.task_type == "dxf_compare")
|
|
||||||
)).scalars().all()
|
|
||||||
assert len(dxf) == 1 and dxf[0].vision_json is None
|
|
||||||
|
|
||||||
async def test_misure_nel_numero_atteso(self, seed_fiera, db_session):
|
|
||||||
await seed_fiera.seed(False, False, FINE)
|
|
||||||
|
|
||||||
# ogni pezzo ha una misura per quota: pezzi 150-300 per ricetta
|
|
||||||
for codice, quote in (("FIERA-FLANGIA", 5), ("FIERA-ALBERO", 4)):
|
|
||||||
misure = (await db_session.execute(
|
|
||||||
select(func.count(Measurement.id))
|
|
||||||
.where(Measurement.version_id.in_(
|
|
||||||
select(seed_fiera.RecipeVersion.id)
|
|
||||||
.join(Recipe, Recipe.id == seed_fiera.RecipeVersion.recipe_id)
|
|
||||||
.where(Recipe.code == codice)
|
|
||||||
))
|
|
||||||
)).scalar_one()
|
|
||||||
assert misure % quote == 0
|
|
||||||
assert 150 <= misure // quote <= 300, (codice, misure // quote)
|
|
||||||
|
|
||||||
# lun-ven soltanto, e le misure compaiono nelle produzioni
|
|
||||||
giorni = (await db_session.execute(
|
|
||||||
select(seed_fiera.ProductionRun.started_at)
|
|
||||||
)).scalars().all()
|
|
||||||
assert giorni and all(g.weekday() < 5 for g in giorni)
|
|
||||||
senza_run = (await db_session.execute(
|
|
||||||
select(func.count(Measurement.id)).where(Measurement.production_run_id.is_(None))
|
|
||||||
)).scalar_one()
|
|
||||||
assert senza_run == 0
|
|
||||||
|
|
||||||
async def test_statistiche_numeriche(self, seed_fiera, client, db_session):
|
|
||||||
await seed_fiera.seed(False, False, FINE)
|
|
||||||
adriano = await _utente(db_session, "adriano")
|
|
||||||
from src.backend.services.auth_service import issue_api_key
|
|
||||||
chiave = await issue_api_key(db_session, adriano)
|
|
||||||
intestazioni = {"X-API-Key": chiave}
|
|
||||||
|
|
||||||
flangia = (await db_session.execute(
|
|
||||||
select(Recipe).where(Recipe.code == "FIERA-FLANGIA")
|
|
||||||
)).scalar_one()
|
|
||||||
quote = (await client.get(
|
|
||||||
"/api/statistics/subtasks", headers=intestazioni,
|
|
||||||
params={"recipe_id": flangia.id},
|
|
||||||
)).json()
|
|
||||||
assert len(quote) == 5
|
|
||||||
cpk = []
|
|
||||||
for quota in quote:
|
|
||||||
risposta = await client.get(
|
|
||||||
"/api/statistics/capability", headers=intestazioni,
|
|
||||||
params={"recipe_id": flangia.id, "subtask_id": quota["id"]},
|
|
||||||
)
|
|
||||||
assert risposta.status_code == 200
|
|
||||||
d = risposta.json()
|
|
||||||
for campo in ("mean", "std_dev", "cp", "cpk"):
|
|
||||||
assert isinstance(d[campo], (int, float)), (quota["id"], campo)
|
|
||||||
cpk.append(d["cpk"])
|
|
||||||
# Cpk diversi per quota: una capace, una al limite o sotto
|
|
||||||
assert max(cpk) > 1.4
|
|
||||||
assert min(cpk) < 1.1
|
|
||||||
|
|
||||||
async def test_secondo_giro_non_duplica_e_replace_rifa(self, seed_fiera, db_session):
|
|
||||||
async def conta():
|
|
||||||
return [
|
|
||||||
(await db_session.execute(select(func.count(t.id)))).scalar_one()
|
|
||||||
for t in (Recipe, Measurement, User)
|
|
||||||
]
|
|
||||||
|
|
||||||
await seed_fiera.seed(False, False, FINE)
|
|
||||||
primo = await conta()
|
|
||||||
await seed_fiera.seed(False, False, FINE)
|
|
||||||
assert await conta() == primo # utenti aggiornati, ricette e misure non duplicate
|
|
||||||
|
|
||||||
await seed_fiera.seed(True, False, FINE)
|
|
||||||
assert await conta() == primo # stessi numeri: il seme e' fisso
|
|
||||||
|
|
||||||
async def test_dry_run_non_salva(self, seed_fiera, db_session):
|
|
||||||
await seed_fiera.seed(False, True, FINE)
|
|
||||||
assert (await db_session.execute(select(func.count(User.id)))).scalar_one() == 0
|
|
||||||
assert (await db_session.execute(select(func.count(Measurement.id)))).scalar_one() == 0
|
|
||||||
|
|
||||||
|
|
||||||
async def test_file_sqlite_con_alembic(seed_fiera, tmp_path, monkeypatch):
|
|
||||||
"""La strada vera: schema da `alembic upgrade head`, poi il seed con --db."""
|
|
||||||
db = tmp_path / "tmf.db"
|
|
||||||
url = f"sqlite+aiosqlite:///{db.as_posix()}"
|
|
||||||
monkeypatch.setattr(Settings, "database_url", property(lambda self: url))
|
|
||||||
# alembic fa asyncio.run(): fuori dal loop di pytest-asyncio, in un thread
|
|
||||||
import asyncio
|
|
||||||
await asyncio.to_thread(command.upgrade, Config(str(_ALEMBIC_INI)), "head")
|
|
||||||
|
|
||||||
argomenti = Namespace(db=db, replace=False, dry_run=False, fine=FINE.isoformat())
|
|
||||||
assert await seed_fiera._main_async(argomenti) == 0
|
|
||||||
|
|
||||||
con = sqlite3.connect(db)
|
|
||||||
try:
|
|
||||||
assert con.execute("SELECT COUNT(*) FROM users WHERE is_admin=1").fetchone()[0] == 1
|
|
||||||
assert con.execute("SELECT COUNT(*) FROM recipes").fetchone()[0] == 3
|
|
||||||
assert con.execute("SELECT COUNT(*) FROM measurements").fetchone()[0] > 1000
|
|
||||||
finally:
|
|
||||||
con.close()
|
|
||||||
@@ -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,
|
||||||
|
}
|
||||||
@@ -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)
|
||||||
@@ -17,6 +17,7 @@ import numpy as np
|
|||||||
from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
|
from fastapi import FastAPI, File, Form, HTTPException, UploadFile, status
|
||||||
from PIL import Image, UnidentifiedImageError
|
from PIL import Image, UnidentifiedImageError
|
||||||
|
|
||||||
|
from src.vision.costruttori import CostruttoreError, build_dxf_compare # #83
|
||||||
from src.vision.runner import (
|
from src.vision.runner import (
|
||||||
GraphBuildError,
|
GraphBuildError,
|
||||||
RunOutcome,
|
RunOutcome,
|
||||||
@@ -223,3 +224,67 @@ async def validate(
|
|||||||
validate_graph, parsed_graph, resolved_dir
|
validate_graph, parsed_graph, resolved_dir
|
||||||
)
|
)
|
||||||
return {"ok": ok, "errors": errors, "outputs": outputs}
|
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
|
||||||
Reference in New Issue
Block a user