From bc84407a6fd6aef61cc75317e72b101ddd940fbf Mon Sep 17 00:00:00 2001 From: Davide Grilli Date: Thu, 6 Aug 2026 11:45:39 +0200 Subject: [PATCH] Lettore di metadati per file nativi Creo Parametric Prototipo CLI in Python che estrae parametri e proprieta' dai file .prt senza avviare Creo e senza SDK commerciali. Solo libreria standard. I file nativi Creo sono container con header ASCII e un indice delle sezioni (UGC_TOC); la maggior parte delle sezioni non e' compressa e la tabella dei parametri e' leggibile direttamente. Ogni parametro compare in due rappresentazioni indipendenti (tabella estesa in LargeText e copia neutra in NeuPrtSld). Il lettore le confronta entrambe, e ne ricava: - confidence: 0.99 quando le due copie concordano, valori inferiori se il dato compare una volta sola o se divergono (status "conflict"); - owner: distingue i parametri del modello da quelli delle feature, che vivono solo nella copia neutra. I parametri numerici (PESO, DENSITA) usano una codifica a lunghezza variabile non ancora risolta: vengono riportati come "encoded_not_decoded" conservando i byte grezzi, senza inventare valori. Validato su file PART e PART/SHEETMETAL scritti da Creo 9.0.3.0. Assiemi, family table e versioni precedenti non sono ancora stati verificati. Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 46 ++++++++++ README.md | 138 ++++++++++++++++++++++++++++ creoparams/__init__.py | 7 ++ creoparams/__main__.py | 13 +++ creoparams/cli.py | 201 +++++++++++++++++++++++++++++++++++++++++ creoparams/extract.py | 161 +++++++++++++++++++++++++++++++++ creoparams/params.py | 195 +++++++++++++++++++++++++++++++++++++++ creoparams/ugc.py | 185 +++++++++++++++++++++++++++++++++++++ test_regression.py | 112 +++++++++++++++++++++++ 9 files changed, 1058 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 creoparams/__init__.py create mode 100644 creoparams/__main__.py create mode 100644 creoparams/cli.py create mode 100644 creoparams/extract.py create mode 100644 creoparams/params.py create mode 100644 creoparams/ugc.py create mode 100644 test_regression.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ce5e0e7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,46 @@ +# --- Modelli Creo ----------------------------------------------------------- +# I file nativi sono disegni aziendali: non finiscono nel repository. +# I test di regressione girano sui file presenti in locale e si saltano da +# soli se la cartella e' vuota. +*.prt +*.prt.* +*.asm +*.asm.* +*.drw +*.drw.* +*.frm +*.frm.* +*.lay +*.lay.* +*.neu +*.stp +*.step + +# --- Output generati -------------------------------------------------------- +metadati.json +metadati.csv +*.metadata.json +/out/ +/export/ + +# --- Python ----------------------------------------------------------------- +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.venv/ +venv/ +env/ +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +htmlcov/ + +# --- Editor e sistema ------------------------------------------------------- +.vscode/ +.idea/ +*.swp +.DS_Store +Thumbs.db diff --git a/README.md b/README.md new file mode 100644 index 0000000..1cfa865 --- /dev/null +++ b/README.md @@ -0,0 +1,138 @@ +# creoparams + +Estrae parametri e proprietà dai file nativi Creo Parametric (`.prt`, `.asm`, +`.drw`) **senza avviare Creo** e senza SDK commerciali. Solo Python 3, nessuna +dipendenza esterna. + +## Uso + +Un file alla volta — è il caso d'uso principale: + +```bash +python3 -m creoparams 9258400201.prt.12 # tabella a video +python3 -m creoparams 9258400201.prt.12 --quiet -o - # JSON su stdout +python3 -m creoparams 9258400201.prt.12 -o scheda.json +``` + +Con un solo file l'output JSON è un **oggetto**; con più file una **lista**. +`-o -` scrive su stdout, quindi il comando si può mettere in pipe. + +Un file indicato esplicitamente viene sempre elaborato così com'è: la scelta +automatica dell'ultima versione riguarda solo l'esplorazione di una cartella. + +In lotto, quando serve: + +```bash +python3 -m creoparams . --csv metadati.csv # cartella corrente +python3 -m creoparams /rete/archivio -r --csv indice.csv +``` + +Opzioni principali: + +| Opzione | Effetto | +|---|---| +| `-o`, `--json FILE` | scrive il JSON su file, oppure su stdout con `-` | +| `--csv FILE` | tabella riepilogativa CSV (Excel, separatore `;`) | +| `-r`, `--recursive` | esplora le sottocartelle | +| `--all-versions` | elabora tutte le versioni (`.prt.1`, `.prt.2`, …) invece della sola più recente | +| `--features` | include i parametri delle feature (fori, lavorazioni) oltre a quelli di modello | +| `--system` | include i parametri generati da Creo (`PTC_*`, `SMT_*`) | +| `--quiet` | non stampa la tabella | + +Codici di uscita: `0` estrazione riuscita, `1` nessun file trovato, `2` nessun +file elaborabile (file mancante o non nativo Creo). I messaggi diagnostici +vanno su stderr, così stdout resta pulito per le pipe. + +Nella tabella a video, il marcatore a inizio riga indica l'affidabilità: +spazio = confermato in due rappresentazioni indipendenti, `?` = trovato una +volta sola, `~` = valore binario non decodificato, `!` = valori discordanti, +`-` = parametro atteso ma assente. + +## Come funziona + +Un file nativo Creo è un container con header ASCII in chiaro, un indice delle +sezioni (`UGC_TOC`) e un corpo binario. La maggior parte delle sezioni **non è +compressa**, e la tabella dei parametri è leggibile direttamente. + +Ogni parametro è presente in due rappresentazioni indipendenti: + +- la **tabella estesa** (sezione `LargeText`), che contiene i parametri di modello; +- la **copia neutra** (sezione `NeuPrtSld`), che contiene anche quelli delle feature. + +Il lettore percorre entrambe e confronta i risultati. Da qui derivano due +informazioni che finiscono nell'output: + +- `owner`: `model` se il parametro compare nella tabella estesa, `feature` se + vive solo nella copia neutra (è un parametro di un foro, non del pezzo); +- `confidence`: `0.99` se le due rappresentazioni concordano, valori più bassi + se il dato compare una volta sola o se le copie divergono (`status: conflict`). + +## Formato di output + +```json +{ + "schema_version": 1, + "file": { "name": "9258400201.prt.12", "sha256": "994dd3a8…", "size": 450797 }, + "model": { "name": "9258400201", "kind": "PART", "creo_version": "9.0.3.0" }, + "extraction": { "method": "ugc_native_reader", "holds_model_parameters": true }, + "parameters": { + "DENOMINAZIONE": { + "type": "string", "owner": "model", "scope": "user", + "value": "Carter lato destro", "status": "found", "confidence": 0.99, + "sections": ["LargeText", "NeuPrtSld"] + }, + "PESO": { + "type": "real", "status": "encoded_not_decoded", + "raw": "90f8382356ca2d", "driven_by": "value(d_val)" + } + } +} +``` + +L'`sha256` permette di verificare in seguito se un record estratto corrisponde +ancora al file da cui proviene. + +## Limiti noti + +**Parametri numerici non decodificati.** `PESO` e `DENSITA` usano una codifica +binaria a lunghezza variabile (7–8 byte osservati) che non è un double IEEE +lineare. Non vengono inventati valori: lo stato è `encoded_not_decoded` e i byte +grezzi sono conservati in `raw`, così da poter essere interpretati in seguito +senza rileggere l'archivio. Per decifrare la codifica servono alcuni valori di +riferimento letti da Creo. + +**Disegni.** I `.drw` vengono riconosciuti e ne viene letto l'header, ma non +contengono la tabella parametri del modello: espongono note e cartiglio, con una +struttura diversa non ancora implementata. Il campo `holds_model_parameters` +lo dichiara esplicitamente. + +**Copertura verificata.** Il lettore è stato validato su file `PART` e +`PART/SHEETMETAL` scritti da **Creo 9.0.3.0**. Non è stato provato su assiemi, +family table, né su file di versioni Creo precedenti. Prima di usarlo +sull'archivio storico, va verificato su un campione di file più vecchi. + +**Family table.** Le istanze non hanno un file proprio: vivono nel generic. Su +un generic con family table il lettore restituirebbe i valori del generic, non +quelli dell'istanza. Nei file campione la sezione `FamilyInf` è vuota, quindi +il caso non è mai stato esercitato. + +## Test + +```bash +python3 -m unittest test_regression -v +``` + +I test girano sui `.prt` presenti nella cartella e verificano che i parametri +aziendali obbligatori siano letti con confidenza alta. `KNOWN` in +`test_regression.py` contiene i valori verificati a mano: **è la rete di +sicurezza vera del progetto** e va estesa ogni volta che si confronta un nuovo +file con Creo. + +## Struttura + +| File | Contenuto | +|---|---| +| [creoparams/ugc.py](creoparams/ugc.py) | header e indice delle sezioni del container | +| [creoparams/params.py](creoparams/params.py) | riconoscimento dei record parametro nel binario | +| [creoparams/extract.py](creoparams/extract.py) | aggregazione, confidenza, record di output | +| [creoparams/cli.py](creoparams/cli.py) | riga di comando ed esportazioni | diff --git a/creoparams/__init__.py b/creoparams/__init__.py new file mode 100644 index 0000000..e6d0c8f --- /dev/null +++ b/creoparams/__init__.py @@ -0,0 +1,7 @@ +"""Lettura dei metadati dai file nativi Creo Parametric senza avviare Creo.""" + +from .extract import extract, latest_versions +from .ugc import NotACreoFile + +__all__ = ["extract", "latest_versions", "NotACreoFile"] +__version__ = "0.1.0" diff --git a/creoparams/__main__.py b/creoparams/__main__.py new file mode 100644 index 0000000..032beb3 --- /dev/null +++ b/creoparams/__main__.py @@ -0,0 +1,13 @@ +import os +import sys + +from .cli import main + +try: + sys.exit(main()) +except KeyboardInterrupt: + sys.exit(130) +except BrokenPipeError: + # Uscita normale quando l'output finisce in una pipe chiusa (| head). + os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno()) + sys.exit(0) diff --git a/creoparams/cli.py b/creoparams/cli.py new file mode 100644 index 0000000..4202930 --- /dev/null +++ b/creoparams/cli.py @@ -0,0 +1,201 @@ +"""Interfaccia a riga di comando.""" + +from __future__ import annotations + +import argparse +import csv +import json +import os +import sys + +from . import ugc +from .extract import extract as extract_file +from .extract import latest_versions, split_filename + +CREO_EXTENSIONS = (".prt", ".asm", ".drw", ".frm", ".lay") + +#: Parametri aziendali attesi: se mancano vengono riportati esplicitamente +#: come "parameter_not_present" invece di sparire dall'output. +DEFAULT_EXPECTED = [ + "CODICE", + "DENOMINAZIONE", + "REVISIONE", + "STAMPIGLIATURA", + "MATERIALE", + "DISEGNATORE", + "DATA", +] + + +def _is_creo_file(path: str) -> bool: + stem, ext, _ = split_filename(os.path.basename(path)) + return f".{ext}" in CREO_EXTENSIONS + + +def collect(inputs: "list[str]", recursive: bool) -> "tuple[list[str], list[str]]": + """Separa i file indicati esplicitamente da quelli trovati esplorando una + cartella: solo su questi ultimi ha senso scegliere l'ultima versione.""" + explicit, discovered = [], [] + for item in inputs: + if os.path.isdir(item): + walker = os.walk(item) if recursive else [(item, [], os.listdir(item))] + for root, _, files in walker: + discovered.extend( + os.path.join(root, f) for f in files if _is_creo_file(f) + ) + else: + explicit.append(item) + return sorted(set(explicit)), sorted(set(discovered)) + + +def _visible(record: dict, include_system: bool, include_features: bool = False) -> dict: + return { + name: entry + for name, entry in record["parameters"].items() + if (include_system or entry.get("scope") != "system") + and (include_features or entry.get("owner", "model") == "model") + } + + +def _print_table(records: "list[dict]", stream, include_system: bool = False, + include_features: bool = False) -> None: + for record in records: + model = record["model"] + header = f"{record['file']['name']} [{model['kind']}" + if model["subtype"]: + header += f"/{model['subtype']}" + header += f"] Creo {model['creo_version'] or '?'}" + print(header, file=stream) + print("-" * len(header), file=stream) + + if not record["extraction"]["holds_model_parameters"]: + print(" (i disegni non contengono la tabella parametri del modello)", + file=stream) + print(file=stream) + continue + + for name, entry in _visible(record, include_system, include_features).items(): + status = entry.get("status") + if status == "found": + shown = entry["value"] if entry["value"] != "" else "(vuoto)" + mark = " " if entry.get("confidence", 0) >= 0.95 else "?" + elif status == "encoded_not_decoded": + shown = "" + mark = "~" + elif status == "conflict": + shown = " | ".join(entry.get("candidates", [])) + mark = "!" + else: + shown = "(assente)" + mark = "-" + print(f" {mark} {name:<22} {shown}", file=stream) + print(file=stream) + + +def _write_csv(records: "list[dict]", path: str, include_system: bool = False, + include_features: bool = False) -> None: + names = sorted({n for r in records for n in _visible(r, include_system, include_features)}) + columns = ["file", "model", "kind", "creo_version"] + names + + with open(path, "w", newline="", encoding="utf-8-sig") as handle: + writer = csv.DictWriter(handle, fieldnames=columns, delimiter=";") + writer.writeheader() + for record in records: + row = { + "file": record["file"]["name"], + "model": record["model"]["name"], + "kind": record["model"]["kind"], + "creo_version": record["model"]["creo_version"], + } + for name in names: + entry = record["parameters"].get(name, {}) + row[name] = entry.get("value") if entry.get("status") == "found" else "" + writer.writerow(row) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="creoparams", + description="Estrae parametri e proprieta' dai file nativi Creo " + "(.prt/.asm/.drw) senza avviare Creo Parametric.", + epilog="Esempi:\n" + " creoparams 9258400201.prt.12 un singolo file\n" + " creoparams 9258400201.prt.12 -o - JSON su stdout\n" + " creoparams /rete/archivio -r --csv i.csv intera cartella", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument("inputs", nargs="+", metavar="FILE|CARTELLA") + parser.add_argument("-o", "--json", metavar="FILE", + help="scrive il risultato in JSON; usare '-' per " + "stdout. Con un solo file emette un oggetto, " + "con piu' file una lista") + parser.add_argument("--csv", metavar="FILE", + help="scrive una tabella riepilogativa in CSV") + parser.add_argument("-r", "--recursive", action="store_true", + help="esplora le sottocartelle") + parser.add_argument("--all-versions", action="store_true", + help="elabora tutte le versioni (.prt.1, .prt.2, ...) " + "invece della sola piu' recente") + parser.add_argument("--system", action="store_true", + help="include anche i parametri generati da Creo " + "(PTC_*, SMT_*)") + parser.add_argument("--features", action="store_true", + help="include anche i parametri appartenenti alle " + "feature invece dei soli parametri di modello") + parser.add_argument("--quiet", action="store_true", + help="non stampa la tabella a video") + return parser + + +def main(argv: "list[str] | None" = None) -> int: + args = build_parser().parse_args(argv) + + explicit, discovered = collect(args.inputs, args.recursive) + if not args.all_versions: + # Un file indicato a mano si elabora com'e': la scelta della versione + # piu' recente riguarda solo l'esplorazione di una cartella. + discovered = latest_versions(discovered) + paths = sorted(set(explicit) | set(discovered)) + + if not paths: + print("Nessun file Creo trovato.", file=sys.stderr) + return 1 + + records, failures = [], [] + for path in paths: + try: + records.append(extract_file(path, DEFAULT_EXPECTED)) + except ugc.NotACreoFile as error: + failures.append((path, str(error))) + except OSError as error: + failures.append((path, str(error))) + + if not args.quiet: + _print_table(records, sys.stdout, args.system, args.features) + + if args.json: + # Un solo file produce un oggetto, non una lista di un elemento: + # e' la forma piu' comoda da consumare a valle. + payload = records[0] if len(records) == 1 else records + if args.json == "-": + json.dump(payload, sys.stdout, indent=2, ensure_ascii=False) + sys.stdout.write("\n") + else: + with open(args.json, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, ensure_ascii=False) + print(f"JSON scritto in {args.json}", file=sys.stderr) + + if args.csv: + _write_csv(records, args.csv, args.system, args.features) + print(f"CSV scritto in {args.csv}", file=sys.stderr) + + for path, reason in failures: + print(f"SALTATO {os.path.basename(path)}: {reason}", file=sys.stderr) + + # Su un singolo file andato a buon fine il riepilogo e' solo rumore. + if len(paths) > 1 or failures: + print( + f"{len(records)} file elaborati, {len(failures)} saltati.", + file=sys.stderr, + ) + return 0 if records else 2 diff --git a/creoparams/extract.py b/creoparams/extract.py new file mode 100644 index 0000000..940d49a --- /dev/null +++ b/creoparams/extract.py @@ -0,0 +1,161 @@ +"""Costruzione del record di metadati a partire da un file nativo Creo.""" + +from __future__ import annotations + +import hashlib +import os +import re +from collections import defaultdict +from datetime import datetime, timezone + +from . import params, ugc + +SCHEMA_VERSION = 1 +METHOD = "ugc_native_reader" + +#: Prefissi dei parametri generati da Creo (impostazioni lamiera, materiale di +#: sistema, ...). Restano nell'output ma marcati come "system", per non +#: annegare i parametri aziendali nel rumore. +SYSTEM_PREFIXES = ("PTC_", "SMT_", "PRO_", "__") + +#: Nome file Creo con numero di versione: "9258400201.prt.12". +_NAME_RE = re.compile(r"^(?P.+?)\.(?Pprt|asm|drw|frm|lay)(?:\.(?P\d+))?$", re.I) + + +def split_filename(filename: str) -> tuple[str, str, int | None]: + """"9258400201.prt.12" -> ("9258400201", "prt", 12).""" + match = _NAME_RE.match(filename) + if not match: + stem, _, ext = filename.rpartition(".") + return (stem or filename), ext.lower(), None + version = match.group("ver") + return match.group("stem"), match.group("ext").lower(), int(version) if version else None + + +def _confidence(values: set, occurrences: int, sections: set) -> float: + if len(values) > 1: + return 0.40 + if len(sections) >= 2: + return 0.99 + if occurrences >= 2: + return 0.95 + return 0.80 + + +def _aggregate(occurrences: list) -> dict: + """Raggruppa le comparse per nome e assegna stato e confidenza.""" + grouped = defaultdict(list) + for occurrence in occurrences: + grouped[occurrence.name].append(occurrence) + + result = {} + for name, group in grouped.items(): + kind = group[0].kind + sections = {o.section for o in group if o.section} + # Solo i parametri del modello compaiono nella tabella estesa; quelli + # che vivono unicamente nella copia neutra appartengono alle feature + # (fori, lavorazioni), non al modello. + owner = "model" if any(o.form == "extended" for o in group) else "feature" + + entry = { + "type": kind, + "owner": owner, + "scope": "system" if name.startswith(SYSTEM_PREFIXES) else "user", + "occurrences": len(group), + "sections": sorted(sections), + } + + if kind == "real": + raws = sorted({o.raw.hex() for o in group if o.raw}) + expressions = sorted({o.expression for o in group if o.expression}) + entry.update( + value=None, + status="encoded_not_decoded", + confidence=None, + raw=raws[0] if len(raws) == 1 else raws, + ) + if expressions: + # Il valore non e' costante: lo produce una relazione. + entry["driven_by"] = expressions[0] if len(expressions) == 1 else expressions + else: + values = {o.value for o in group} + entry["confidence"] = _confidence(values, len(group), sections) + if len(values) == 1: + entry.update(value=group[0].value, status="found") + else: + entry.update( + value=None, + status="conflict", + candidates=sorted(values), + ) + result[name] = entry + + return dict(sorted(result.items())) + + +def extract(path: str, expected: "list[str] | None" = None) -> dict: + """Estrae i metadati da un singolo file. Solleva ugc.NotACreoFile se il + file non e' un nativo Creo riconoscibile.""" + with open(path, "rb") as handle: + data = handle.read() + + container = ugc.parse(data) + occurrences = params.iter_occurrences(data) + for occurrence in occurrences: + occurrence.section = container.section_at(occurrence.offset) + + filename = os.path.basename(path) + stem, ext, version = split_filename(filename) + + parameters = _aggregate(occurrences) + + # I disegni non contengono la tabella parametri del modello: segnalarne + # l'assenza come "parametro mancante" sarebbe fuorviante. + holds_parameters = container.base_kind in ("PART", "ASSEMBLY") + if holds_parameters: + for name in expected or []: + parameters.setdefault( + name, {"value": None, "status": "parameter_not_present", "type": None} + ) + + return { + "schema_version": SCHEMA_VERSION, + "source_file": os.path.abspath(path), + "file": { + "name": filename, + "size": len(data), + "sha256": hashlib.sha256(data).hexdigest(), + "modified": datetime.fromtimestamp( + os.path.getmtime(path), timezone.utc + ).isoformat(), + "creo_version_number": version, + }, + "model": { + "name": container.common_name.rsplit(".", 1)[0] or stem, + "file_name": container.common_name or f"{stem}.{ext}", + "kind": container.base_kind, + "subtype": container.subtype, + "creo_version": container.creo_version, + "banner": container.banner, + }, + "extraction": { + "method": METHOD, + "extracted_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "sections": len(container.sections), + "parameters_found": len(occurrences), + "holds_model_parameters": holds_parameters, + }, + "parameters": parameters, + } + + +def latest_versions(paths: "list[str]") -> "list[str]": + """Tra piu' versioni dello stesso modello tiene solo la piu' recente.""" + best: dict = {} + for path in paths: + stem, ext, version = split_filename(os.path.basename(path)) + key = (os.path.dirname(os.path.abspath(path)), stem, ext) + current = best.get(key) + if current is None or (version or -1) > (current[0] or -1): + best[key] = (version, path) + return sorted(path for _, path in best.values()) diff --git a/creoparams/params.py b/creoparams/params.py new file mode 100644 index 0000000..62f36b8 --- /dev/null +++ b/creoparams/params.py @@ -0,0 +1,195 @@ +"""Estrazione della tabella parametri dal corpo di un file nativo Creo. + +I parametri sono memorizzati in chiaro, con una struttura regolare che si +ripete per ogni voce: + + f6 e3 00 88 00 e3 ... + +dove `tipo` e' 0x33 ('3') per le stringhe e 0x32 ('2') per i numerici. +Ogni parametro compare due volte di seguito (valore corrente e valore di +riferimento) e l'intera tabella e' duplicata in due sezioni del file +(tipicamente `LargeText` e `NeuPrtSld`). Usiamo questa ridondanza per +assegnare un livello di confidenza a ogni valore. + +I valori stringa sono leggibili direttamente. I numerici usano una codifica +binaria a lunghezza variabile (7-8 byte osservati) che non e' un double IEEE +lineare e che qui non viene decodificata: ne conserviamo i byte grezzi. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +REC_START = b"\xf6\xe3" +VALUE_LEAD = b"\x88\x00\xe3" +NUM_LEAD = b"\xe3\x32" + +KIND_STRING = 0x33 +KIND_NUMERIC = 0x32 + +#: Marcatore di espressione: il valore non e' costante ma prodotto da una +#: relazione del modello (es. PESO guidato da "value(d_val)"). +EXPR_LEAD = b"\xe0\x02" + +MAX_STRING = 256 +MAX_NUMERIC_SPAN = 48 + +# Byte ammessi in un valore stringa: ASCII stampabile piu' l'intervallo alto +# per le lettere accentate, esclusi i byte usati come token dal formato. +_TOKENS = {0xE1, 0xE3, 0xF6, 0xF7} +_STRING_BYTES = frozenset( + set(range(0x20, 0x7F)) | (set(range(0xA0, 0x100)) - _TOKENS) +) + +_NAME = rb"([A-Za-z_][A-Za-z0-9_.\-]{0,63})\x00" + +#: Forma estesa, nella tabella parametri principale (sezione LargeText). +_RECORD_RE = re.compile( + REC_START + _NAME + + rb"(.)" # byte di slot/identificativo + + re.escape(VALUE_LEAD) + + rb"([\x32\x33])", # tipo del valore + re.DOTALL, +) + +#: Forma compatta, nella rappresentazione neutra (sezione NeuPrtSld). Ripete +#: gli stessi parametri e ci serve come verifica incrociata. Il prefisso che +#: separa i record varia a seconda dei campi accessori presenti (ref_id, +#: unit, designated, ...), quindi ci ancoriamo solo al nome seguito dal tipo. +_NEUTRAL_RE = re.compile( + rb"\xe3" + _NAME + rb"\xe2([\x32\x33])", + re.DOTALL, +) + + +@dataclass +class Occurrence: + """Una singola comparsa di un parametro nel file.""" + + name: str + kind: str # "string" | "real" + value: str | None # None se non decodificato + raw: bytes | None # byte grezzi, per i numerici + offset: int + section: str = "" + expression: str | None = None # relazione che pilota il valore + form: str = "extended" # "extended" (tabella modello) | "neutral" + + @property + def decoded(self) -> bool: + return self.value is not None + + +def _decode_text(raw: bytes) -> str: + """I valori sono scritti in UTF-8; i file piu' vecchi possono usare la + codepage Windows.""" + try: + return raw.decode("utf-8") + except UnicodeDecodeError: + return raw.decode("cp1252", errors="replace") + + +def _read_string(data: bytes, pos: int) -> tuple[str, int] | None: + """Legge un valore stringa terminato da NUL. Ritorna (valore, fine).""" + # Un token f7 xx puo' precedere il valore (flag di stato del parametro). + if data[pos : pos + 1] == b"\xf7": + pos += 2 + + end = pos + limit = min(len(data), pos + MAX_STRING) + while end < limit and data[end] != 0x00: + if data[end] not in _STRING_BYTES: + return None + end += 1 + if end >= limit: + return None # nessun terminatore: non e' un record valido + + return _decode_text(data[pos:end]), end + 1 + + +def _read_numeric(data: bytes, pos: int) -> tuple[bytes, int] | None: + """Isola i byte grezzi di un valore numerico. + + Sfrutta il fatto che il valore e' immediatamente ripetuto dopo un secondo + marcatore `e3 32`: proviamo ogni posizione candidata e teniamo la prima in + cui la ripetizione combacia davvero. + """ + start = pos + # Un token f7 xx puo' precedere il valore. + if data[start : start + 1] == b"\xf7": + start += 2 + + search_end = min(len(data), start + MAX_NUMERIC_SPAN) + cursor = start + while True: + repeat = data.find(NUM_LEAD, cursor, search_end) + if repeat < 0: + return None + + candidate = data[start:repeat] + # Un token di separazione puo' chiudere il primo valore. + if candidate[-2:-1] == b"\xf7": + candidate = candidate[:-2] + + after = repeat + len(NUM_LEAD) + if candidate and data[after : after + len(candidate)] == candidate: + return candidate, after + len(candidate) + + cursor = repeat + 1 + + +def _read_expression(data: bytes, pos: int) -> tuple[str | None, int]: + """Se il valore e' pilotato da una relazione, ne legge il testo.""" + if data[pos : pos + len(EXPR_LEAD)] != EXPR_LEAD: + return None, pos + parsed = _read_string(data, pos + len(EXPR_LEAD)) + if parsed is None: + return None, pos + return parsed + + +def iter_occurrences(data: bytes) -> "list[Occurrence]": + """Scorre il file e restituisce ogni comparsa di parametro trovata. + + Vengono percorse entrambe le rappresentazioni presenti nel file: la + tabella parametri estesa e la copia neutra. La ridondanza serve poi ad + assegnare la confidenza in `extract`. + """ + found: list[Occurrence] = [] + + for regex, kind_group, form in ( + (_RECORD_RE, 3, "extended"), + (_NEUTRAL_RE, 2, "neutral"), + ): + for match in regex.finditer(data): + name = match.group(1).decode("ascii") + kind_byte = match.group(kind_group)[0] + pos = match.end() + + if kind_byte == KIND_STRING: + parsed = _read_string(data, pos) + if parsed is None: + continue + found.append( + Occurrence( + name, "string", parsed[0], None, match.start(), form=form + ) + ) + else: + expression, pos = _read_expression(data, pos) + parsed = _read_numeric(data, pos) + found.append( + Occurrence( + name, + "real", + None, + parsed[0] if parsed else None, + match.start(), + expression=expression, + form=form, + ) + ) + + found.sort(key=lambda o: o.offset) + return found diff --git a/creoparams/ugc.py b/creoparams/ugc.py new file mode 100644 index 0000000..1df663c --- /dev/null +++ b/creoparams/ugc.py @@ -0,0 +1,185 @@ +"""Lettura del container UGC dei file nativi Creo (.prt, .asm, .drw). + +Un file nativo Creo si apre con un header ASCII in chiaro, seguito da un +indice delle sezioni (UGC_TOC) e dal corpo binario. Solo alcune sezioni sono +compresse: la tabella dei parametri non lo e', ed e' quello che sfruttiamo. + +Struttura osservata su Creo 9.0.3.0: + + #UGC:2 PART/SHEETMETAL 2038 1760 805 1 1 15 4000 2022414 000002d7 \\ + #- VERS 0 0 \\ + #- CMNM 00e9258400200.prt \\ + #-END_OF_UGC_HEADER + #Creo TM 9.0 (c) 2026 by PTC Inc. All Rights Reserved. 9.0.3.0 + #UGC_TOC 2 32 81 17###... + ND:0:Model_L05_PX:1 ab4 440 557 2038 a 1 b715 85d3###... + BasicData 2c3b 15767 1575c 2038 9 -1 e3ea 48f0###... + ... +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +HEADER_END = b"#-END_OF_UGC_HEADER" +TOC_MARKER = b"#UGC_TOC" +MAGIC = b"#UGC:" + +# Quanto leggiamo per cercare header e TOC. L'indice sta sempre nei primi KB. +_HEAD_WINDOW = 64 * 1024 + +_BANNER_RE = re.compile(r"^#(Creo|Pro/ENGINEER).*", re.IGNORECASE) +_VERSION_RE = re.compile(r"(\d+(?:\.\d+){1,3})\s*$") + + +@dataclass +class Section: + """Una voce dell'indice UGC_TOC. + + `alt_size` e' il terzo campo numerico. Per le sezioni compresse e' maggiore + di `size` (dimensione decompressa); per quelle non compresse e' leggermente + minore. Non lo interpretiamo oltre: ci serve solo come indizio di + compressione. + """ + + name: str + offset: int + size: int + alt_size: int + extra: tuple = () + + @property + def end(self) -> int: + return self.offset + self.size + + @property + def looks_compressed(self) -> bool: + return self.alt_size > self.size + + def contains(self, pos: int) -> bool: + return self.offset <= pos < self.end + + +@dataclass +class Container: + """Un file nativo Creo aperto e con header/TOC interpretati.""" + + data: bytes + kind: str = "" # PART, PART/SHEETMETAL, DRAWING, ASSEMBLY... + creo_version: str = "" # es. "9.0.3.0" + banner: str = "" # riga di copyright completa + common_name: str = "" # nome originale del modello (campo CMNM) + header: dict = field(default_factory=dict) + sections: list = field(default_factory=list) + + @property + def base_kind(self) -> str: + return self.kind.split("/", 1)[0] + + @property + def subtype(self) -> str: + parts = self.kind.split("/", 1) + return parts[1] if len(parts) > 1 else "" + + def section_at(self, pos: int) -> str: + for section in self.sections: + if section.contains(pos): + return section.name + return "" + + +class NotACreoFile(ValueError): + """Il file non ha un header UGC riconoscibile.""" + + +def _decode_cmnm(raw: str) -> str: + """Il campo CMNM e' prefissato da 3 cifre esadecimali con la lunghezza. + + Esempio: "00e9258400200.prt" -> 0x00e = 14 = len("9258400200.prt"). + Se il prefisso non torna, restituiamo il valore grezzo. + """ + if len(raw) > 3: + try: + declared = int(raw[:3], 16) + except ValueError: + return raw + if declared == len(raw) - 3: + return raw[3:] + return raw + + +def _parse_toc_line(line: str) -> Section | None: + # Le righe del TOC sono paddate a destra con '#'. + line = line.rstrip("#").strip() + if not line: + return None + parts = line.split() + if len(parts) < 4: + return None + try: + offset = int(parts[1], 16) + size = int(parts[2], 16) + alt_size = int(parts[3], 16) + except ValueError: + return None + return Section(parts[0], offset, size, alt_size, tuple(parts[4:])) + + +def parse(data: bytes) -> Container: + """Interpreta header e indice. Non tocca il corpo binario.""" + if not data.startswith(MAGIC): + raise NotACreoFile("header UGC assente (i primi byte non sono '#UGC:')") + + head_end = data.find(HEADER_END) + if head_end < 0: + raise NotACreoFile("marcatore di fine header UGC non trovato") + + container = Container(data=data) + window = data[: max(head_end + _HEAD_WINDOW, head_end)].decode( + "latin-1", errors="replace" + ) + lines = window.split("\n") + + in_toc = False + for index, line in enumerate(lines): + stripped = line.rstrip().rstrip("\\").rstrip() + + if index == 0: + # "#UGC:2 PART/SHEETMETAL 2038 1760 ..." + tokens = stripped.split() + container.kind = tokens[1] if len(tokens) > 1 else "" + continue + + if in_toc: + if stripped.startswith("#"): + break # fine dell'indice, inizia il corpo + section = _parse_toc_line(stripped) + if section: + container.sections.append(section) + # NEXT_TOC_ENTRY non punta a un secondo indice ASCII: il corpo + # prosegue con blocchi marcati inline. Ci fermiamo qui. + if section.name == "NEXT_TOC_ENTRY": + break + continue + + if stripped.startswith("#- "): + key, _, value = stripped[3:].partition(" ") + value = value.strip() + container.header[key] = value + if key == "CMNM": + container.common_name = _decode_cmnm(value) + elif stripped.startswith(TOC_MARKER.decode()): + in_toc = True + elif _BANNER_RE.match(stripped): + container.banner = stripped.lstrip("#").strip() + match = _VERSION_RE.search(container.banner) + if match: + container.creo_version = match.group(1) + + return container + + +def read(path) -> Container: + with open(path, "rb") as handle: + return parse(handle.read()) diff --git a/test_regression.py b/test_regression.py new file mode 100644 index 0000000..42a840f --- /dev/null +++ b/test_regression.py @@ -0,0 +1,112 @@ +"""Test di regressione sui file campione presenti nella cartella. + +Il parser si appoggia alla struttura binaria del formato Creo: qualsiasi +modifica alle espressioni di ricerca va verificata contro file reali. Il test +si limita ai file effettivamente presenti, quindi resta eseguibile anche su +un archivio diverso. + + python3 -m unittest test_regression -v +""" + +import glob +import os +import unittest + +from creoparams.extract import extract, latest_versions, split_filename + +HERE = os.path.dirname(os.path.abspath(__file__)) + +#: Parametri che ci aspettiamo di leggere da ogni .prt aziendale. +REQUIRED = ["CODICE", "DENOMINAZIONE", "REVISIONE", "STAMPIGLIATURA", + "MATERIALE", "DISEGNATORE", "DATA"] + +#: Valori verificati manualmente. Vanno estesi man mano che si confrontano +#: altri file con Creo: e' questa la rete di sicurezza vera. +KNOWN = { + "9258400201.prt": { + "CODICE": "9258400201", + "DENOMINAZIONE": "Carter lato destro", + "MATERIALE": "ACC.INOX LAMIERA SP.4", + "DISEGNATORE": "Grilli", + "DATA": "9/02/2026", + }, + "9258400205.prt": { + "CODICE": "9258400205", + "DENOMINAZIONE": "Tirante carter lateriali", + "MATERIALE": "AISI 304 TONDO D.12", + }, +} + + +def _parts(): + paths = latest_versions(glob.glob(os.path.join(HERE, "*.prt*"))) + return [p for p in paths if split_filename(os.path.basename(p))[1] == "prt"] + + +class TestEstrazione(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.parts = _parts() + if not cls.parts: + raise unittest.SkipTest("nessun file .prt nella cartella") + cls.records = { + os.path.basename(p): extract(p) for p in cls.parts + } + + def test_header_riconosciuto(self): + for name, record in self.records.items(): + with self.subTest(file=name): + self.assertEqual(record["model"]["kind"], "PART") + self.assertTrue(record["model"]["creo_version"]) + self.assertGreater(record["extraction"]["sections"], 0) + + def test_parametri_obbligatori_presenti(self): + for name, record in self.records.items(): + for param in REQUIRED: + with self.subTest(file=name, param=param): + entry = record["parameters"].get(param) + self.assertIsNotNone(entry, f"{param} non trovato") + self.assertEqual(entry["status"], "found") + + def test_confidenza_alta(self): + """Ogni parametro obbligatorio deve essere confermato in due + rappresentazioni indipendenti del file.""" + for name, record in self.records.items(): + for param in REQUIRED: + with self.subTest(file=name, param=param): + self.assertGreaterEqual( + record["parameters"][param]["confidence"], 0.95 + ) + + def test_valori_noti(self): + for stem, expected in KNOWN.items(): + matching = [ + r for r in self.records.values() + if r["model"]["file_name"] == stem + ] + if not matching: + continue + record = matching[0] + for param, value in expected.items(): + with self.subTest(file=stem, param=param): + self.assertEqual(record["parameters"][param]["value"], value) + + def test_codice_coerente_col_nome_file(self): + for name, record in self.records.items(): + with self.subTest(file=name): + self.assertEqual( + record["parameters"]["CODICE"]["value"], + record["model"]["name"], + ) + + def test_parametri_feature_separati(self): + """I parametri dei fori non devono finire tra quelli del modello.""" + for name, record in self.records.items(): + for param, entry in record["parameters"].items(): + if param in ("SCREW_SIZE", "DRILL_DIAMETER", "CSINK_ANGLE"): + with self.subTest(file=name, param=param): + self.assertEqual(entry["owner"], "feature") + + +if __name__ == "__main__": + unittest.main()