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) <noreply@anthropic.com>
This commit is contained in:
2026-08-06 11:45:39 +02:00
co-authored by Claude Opus 5
commit bc84407a6f
9 changed files with 1058 additions and 0 deletions
+46
View File
@@ -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
+138
View File
@@ -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 (78 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 |
+7
View File
@@ -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"
+13
View File
@@ -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)
+201
View File
@@ -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 = "<binario non decodificato>"
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
+161
View File
@@ -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<stem>.+?)\.(?P<ext>prt|asm|drw|frm|lay)(?:\.(?P<ver>\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())
+195
View File
@@ -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 <NOME> 00 <slot> 88 00 e3 <tipo> <valore> ...
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
+185
View File
@@ -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())
+112
View File
@@ -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()