Anteprime, decompressione LZW e contenuto dei disegni
Aggiunge tre capacita' al lettore, tutte senza dipendenze esterne.
Anteprima incorporata (preview.py)
Ogni file contiene una miniatura JPEG dell'ultimo salvataggio: il
render del pezzo nei modelli, l'immagine della tavola nei disegni.
Non e' referenziata dall'indice, quindi va cercata per firma: ogni
candidato viene validato percorrendone i segmenti JPEG, cosi' i byte
che iniziano per FFD8FF per caso non passano. Sui file campione il
risultato e' esattamente una immagine per file.
Decompressione compress(1) (lzw.py)
Creo comprime alcuni blocchi nel vecchio formato .Z. Rispetto a un
LZW generico va riprodotto l'allineamento a gruppi di otto codici:
senza, la decompressione produce qualche centinaio di byte corretti
e poi diverge silenziosamente.
Contenuto dei disegni (drawing.py)
Un .drw non ha la tabella parametri del modello ma contiene la
rappresentazione vettoriale della tavola. Ne estraiamo i testi:
cartiglio, note, tolleranze, quote. Le tavole diventano cosi'
ricercabili per contenuto. Nel disegno i valori compaiono gia'
risolti come testo, compreso il peso, mentre nel .prt sono ancora
in codifica binaria.
I blocchi oltre l'indice sono ora interpretati da ugc.iter_blocks, che
decomprime quando serve: lo stesso blocco puo' essere memorizzato
compresso in un file e in chiaro in un altro, quindi la scelta va fatta
guardando i byte e non il nome. Cercare soltanto gli stream compressi
faceva perdere i testi su un terzo dei disegni.
Sulla codifica dei numeri: sono double IEEE big-endian con i byte bassi
omessi, preceduti da un byte-tag. Lo dimostrano i disegni, dove size_x e
size_y valgono 420,0 x 297,0 (un A3 esatto). La mappatura del byte-tag
resta pero' incerta, quindi PESO e DENSITA continuano a essere riportati
come encoded_not_decoded: un valore plausibile ma sbagliato sarebbe
peggio di un valore assente.
La geometria 3D resta fuori portata, e ora e' verificato: in 200 KB di
blocchi geometrici non compare una sola costante 1.0 in nessuna codifica
in virgola mobile. Quei blocchi sono compressi con un algoritmo che non
espone firme note.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5
@@ -19,6 +19,9 @@
|
||||
# --- Output generati --------------------------------------------------------
|
||||
metadati.json
|
||||
metadati.csv
|
||||
# Anteprime estratte dai modelli: sono immagini dei pezzi, come i modelli.
|
||||
*.jpg
|
||||
/anteprime/
|
||||
*.metadata.json
|
||||
/out/
|
||||
/export/
|
||||
|
||||
@@ -37,6 +37,7 @@ Opzioni principali:
|
||||
| `--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_*`) |
|
||||
| `--preview [DEST]` | salva l'anteprima incorporata come JPEG |
|
||||
| `--quiet` | non stampa la tabella |
|
||||
|
||||
Codici di uscita: `0` estrazione riuscita, `1` nessun file trovato, `2` nessun
|
||||
@@ -48,6 +49,63 @@ spazio = confermato in due rappresentazioni indipendenti, `?` = trovato una
|
||||
volta sola, `~` = valore binario non decodificato, `!` = valori discordanti,
|
||||
`-` = parametro atteso ma assente.
|
||||
|
||||
## Anteprima incorporata
|
||||
|
||||
Ogni file contiene una miniatura JPEG dell'ultimo salvataggio: il render
|
||||
ombreggiato del pezzo nei modelli, l'immagine della tavola nei disegni. È la
|
||||
stessa che Creo mostra nella finestra "Apri".
|
||||
|
||||
```bash
|
||||
python3 -m creoparams 9258400201.prt.12 --preview # accanto all'originale
|
||||
python3 -m creoparams 9258400201.prt.12 --preview foto.jpg # nome esplicito
|
||||
python3 -m creoparams . --preview anteprime/ # tutte in una cartella
|
||||
```
|
||||
|
||||
Le dimensioni compaiono nell'intestazione della tabella e in `preview` nel
|
||||
JSON; i byte dell'immagine restano fuori dal record, si ottengono con
|
||||
`read_preview()`.
|
||||
|
||||
La miniatura non è referenziata dall'indice delle sezioni, quindi va cercata
|
||||
per firma. Per non scambiare dati binari qualsiasi per un'immagine, ogni
|
||||
candidato viene validato percorrendone i segmenti JPEG: sui file campione
|
||||
questo produce esattamente un risultato per file, e il test
|
||||
`test_una_sola_anteprima_per_file` lo verifica.
|
||||
|
||||
Attenzione: l'anteprima fotografa **l'ultimo salvataggio**, non lo stato
|
||||
corrente del modello. Se un pezzo è stato modificato e salvato da un CAD che
|
||||
non rigenera la miniatura, l'immagine può essere più vecchia dei parametri.
|
||||
|
||||
## Contenuto dei disegni (`.drw`)
|
||||
|
||||
Un disegno non contiene la tabella parametri del modello, ma contiene la
|
||||
**rappresentazione vettoriale della tavola**, in blocchi compressi con
|
||||
`compress(1)` (il vecchio `.Z` Unix). Il programma li decomprime ed estrae
|
||||
tutti i testi:
|
||||
|
||||
```bash
|
||||
python3 -m creoparams 9258400205.drw.10
|
||||
```
|
||||
|
||||
```
|
||||
118 testi nella tavola (68 unici). I parametri del modello stanno nel .prt.
|
||||
CODICE ... STAMPIGLIATURA ... PESO FINITO ... DISEGNATO ... SCALA
|
||||
9258400205 ... 0.197 ... 9/02/2026 ... 1:2 ... Grilli
|
||||
```
|
||||
|
||||
Questo rende le tavole **ricercabili per contenuto**: voci del cartiglio,
|
||||
note, tolleranze, quote. Da notare che nel disegno i valori compaiono
|
||||
**già risolti come testo** — compreso il peso — mentre nel `.prt` sono
|
||||
numeri in codifica binaria.
|
||||
|
||||
Attenzione: i testi sono estratti in ordine di comparsa, **non associati
|
||||
alla rispettiva etichetta**. Accoppiare `PESO FINITO` al suo valore richiede
|
||||
le coordinate di ogni testo, che non sono ancora decodificate (vedi sotto).
|
||||
Un disegno riporta inoltre lo stato all'ultima rigenerazione, non
|
||||
necessariamente quello attuale del modello.
|
||||
|
||||
Nel JSON tutto questo sta sotto `drawing`, insieme all'elenco delle primitive
|
||||
grafiche presenti (`prim_text`, `prim_arc`, `prim_multiline`, …).
|
||||
|
||||
## Come funziona
|
||||
|
||||
Un file nativo Creo è un container con header ASCII in chiaro, un indice delle
|
||||
@@ -94,22 +152,30 @@ 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.
|
||||
**Parametri numerici non decodificati.** `PESO` e `DENSITA` restano
|
||||
`encoded_not_decoded`, con i byte grezzi in `raw`. La codifica è però ora in
|
||||
gran parte capita: sono **double IEEE big-endian con i byte bassi omessi**,
|
||||
preceduti da un byte-tag. La prova viene dai disegni, dove `size_x` = `2f 7a 40`
|
||||
e `size_y` = `2f 72 90` corrispondono a 420,0 × 297,0 — un A3 esatto — e dove
|
||||
le coordinate di una polilinea danno differenze regolari (−9,72 / −6,48: un
|
||||
tratteggio). Applicando la stessa regola, `DENSITA` risulta **7,8e-06**, la
|
||||
densità dell'acciaio, identica in tutti i file.
|
||||
|
||||
**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.
|
||||
Quello che manca è la mappatura del byte-tag iniziale: sui pesi produce due
|
||||
candidati, uno plausibile e uno assurdo, e su un file nessuno dei due convince.
|
||||
Finché la regola non è certa il programma **non pubblica alcun valore**: un
|
||||
peso plausibile ma sbagliato è peggio di un peso assente. Per chiudere servono
|
||||
due o tre valori di `PESO` letti da Creo.
|
||||
|
||||
**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.
|
||||
Nel frattempo i disegni offrono una via alternativa: nel `.drw` il peso compare
|
||||
**già risolto come testo** nel cartiglio.
|
||||
|
||||
**Copertura verificata.** Il lettore è stato validato su file `PART`,
|
||||
`PART/SHEETMETAL` e `DRAWING` scritti da **Creo 9.0.3.0**. Gli **assiemi
|
||||
(`.asm`) non sono mai stati provati**: nessun file campione era disponibile.
|
||||
Il codice li gestisce come i part, ma è una previsione, non una verifica.
|
||||
Idem per family table e versioni Creo precedenti: prima di usare il lettore
|
||||
sull'archivio storico va provato 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
|
||||
@@ -132,7 +198,10 @@ file con Creo.
|
||||
|
||||
| File | Contenuto |
|
||||
|---|---|
|
||||
| [creoparams/ugc.py](creoparams/ugc.py) | header e indice delle sezioni del container |
|
||||
| [creoparams/ugc.py](creoparams/ugc.py) | header, indice delle sezioni e blocchi del container |
|
||||
| [creoparams/params.py](creoparams/params.py) | riconoscimento dei record parametro nel binario |
|
||||
| [creoparams/lzw.py](creoparams/lzw.py) | decompressione compress(1) dei blocchi interni |
|
||||
| [creoparams/drawing.py](creoparams/drawing.py) | testi e primitive delle tavole |
|
||||
| [creoparams/preview.py](creoparams/preview.py) | anteprima JPEG incorporata |
|
||||
| [creoparams/extract.py](creoparams/extract.py) | aggregazione, confidenza, record di output |
|
||||
| [creoparams/cli.py](creoparams/cli.py) | riga di comando ed esportazioni |
|
||||
|
||||
+60
-4
@@ -10,7 +10,7 @@ import sys
|
||||
|
||||
from . import ugc
|
||||
from .extract import extract as extract_file
|
||||
from .extract import latest_versions, split_filename
|
||||
from .extract import latest_versions, read_preview, split_filename
|
||||
|
||||
CREO_EXTENSIONS = (".prt", ".asm", ".drw", ".frm", ".lay")
|
||||
|
||||
@@ -58,19 +58,33 @@ def _visible(record: dict, include_system: bool, include_features: bool = False)
|
||||
|
||||
|
||||
def _print_table(records: "list[dict]", stream, include_system: bool = False,
|
||||
include_features: bool = False) -> None:
|
||||
include_features: bool = False, limit_texts: int = 20) -> 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 '?'}"
|
||||
shot = record.get("preview", {})
|
||||
if shot.get("present"):
|
||||
header += f" anteprima {shot['width']}x{shot['height']}"
|
||||
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)
|
||||
sheet = record.get("drawing")
|
||||
if sheet and sheet["texts"]:
|
||||
print(" %d testi nella tavola (%d unici). I parametri del "
|
||||
"modello stanno nel .prt." % (
|
||||
sheet["text_count"], len(sheet["texts"])), file=stream)
|
||||
for text in sheet["texts"][:limit_texts]:
|
||||
print(" %s" % text, file=stream)
|
||||
if len(sheet["texts"]) > limit_texts:
|
||||
print(" ... altri %d (usare -o per averli tutti)" % (
|
||||
len(sheet["texts"]) - limit_texts), file=stream)
|
||||
else:
|
||||
print(" (i disegni non contengono la tabella parametri del modello)",
|
||||
file=stream)
|
||||
print(file=stream)
|
||||
continue
|
||||
|
||||
@@ -92,6 +106,40 @@ def _print_table(records: "list[dict]", stream, include_system: bool = False,
|
||||
print(file=stream)
|
||||
|
||||
|
||||
def _preview_path(source: str, dest: str, single: bool) -> str:
|
||||
"""Decide dove scrivere l'anteprima di `source`.
|
||||
|
||||
Senza destinazione la mette accanto all'originale; se la destinazione e'
|
||||
una cartella ci scrive dentro; un nome esplicito vale solo con un file
|
||||
solo, altrimenti i file si sovrascriverebbero a vicenda.
|
||||
"""
|
||||
name = os.path.basename(source) + ".jpg"
|
||||
if not dest:
|
||||
return source + ".jpg"
|
||||
# La barra finale indica una cartella anche se non esiste ancora; con piu'
|
||||
# file la destinazione e' comunque una cartella, altrimenti si
|
||||
# sovrascriverebbero a vicenda.
|
||||
looks_like_dir = dest.endswith(("/", os.sep)) or os.path.isdir(dest)
|
||||
if looks_like_dir or not single:
|
||||
os.makedirs(dest, exist_ok=True)
|
||||
return os.path.join(dest, name)
|
||||
return dest
|
||||
|
||||
|
||||
def _save_previews(paths: "list[str]", dest: str, stream) -> None:
|
||||
single = len(paths) == 1
|
||||
for source in paths:
|
||||
shot = read_preview(source)
|
||||
if shot is None:
|
||||
print(f"{os.path.basename(source)}: nessuna anteprima nel file",
|
||||
file=stream)
|
||||
continue
|
||||
target = _preview_path(source, dest, single)
|
||||
shot.save(target)
|
||||
print(f"Anteprima {shot.width}x{shot.height} salvata in {target}",
|
||||
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)})
|
||||
@@ -131,6 +179,9 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
"con piu' file una lista")
|
||||
parser.add_argument("--csv", metavar="FILE",
|
||||
help="scrive una tabella riepilogativa in CSV")
|
||||
parser.add_argument("--preview", nargs="?", const="", metavar="FILE|CARTELLA",
|
||||
help="salva l'anteprima incorporata come JPEG. Senza "
|
||||
"argomento la scrive accanto al file di origine")
|
||||
parser.add_argument("-r", "--recursive", action="store_true",
|
||||
help="esplora le sottocartelle")
|
||||
parser.add_argument("--all-versions", action="store_true",
|
||||
@@ -189,6 +240,11 @@ def main(argv: "list[str] | None" = None) -> int:
|
||||
_write_csv(records, args.csv, args.system, args.features)
|
||||
print(f"CSV scritto in {args.csv}", file=sys.stderr)
|
||||
|
||||
if args.preview is not None:
|
||||
_save_previews(
|
||||
[r["source_file"] for r in records], args.preview, sys.stderr
|
||||
)
|
||||
|
||||
for path, reason in failures:
|
||||
print(f"SALTATO {os.path.basename(path)}: {reason}", file=sys.stderr)
|
||||
|
||||
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Lettura del contenuto dei disegni Creo (`.drw`).
|
||||
|
||||
Un disegno non contiene la tabella parametri del modello: contiene un blocco
|
||||
`PictureObj`, compresso con compress(1), con la rappresentazione vettoriale
|
||||
della tavola. Dentro ci sono le primitive grafiche (linee, archi, polilinee,
|
||||
testi con font, colore e rotazione) e, soprattutto, **tutti i testi**: voci
|
||||
del cartiglio, note, tolleranze.
|
||||
|
||||
Qui estraiamo i testi, che rendono le tavole ricercabili per contenuto. La
|
||||
geometria e' presente e decodificabile (le coordinate sono double IEEE con i
|
||||
byte bassi omessi) ma il suo disegno non e' implementato.
|
||||
|
||||
I testi seguono uno schema regolare:
|
||||
|
||||
18 e3 e3 43 <a> <b> <testo> 00
|
||||
|
||||
dove il secondo byte vale 2*len+2. Verifichiamo la lunghezza dichiarata
|
||||
contro quella reale: e' un controllo a costo nullo che scarta i falsi
|
||||
riscontri.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
|
||||
from . import ugc
|
||||
|
||||
TEXT_MARK = b"\x18\xe3\xe3\x43"
|
||||
MAX_TEXT = 512
|
||||
|
||||
_STRING_BYTES = frozenset(
|
||||
set(range(0x20, 0x7F)) | (set(range(0xA0, 0x100)) - {0xE1, 0xE3, 0xF6, 0xF7})
|
||||
)
|
||||
|
||||
#: Primitive grafiche note, utili per descrivere il contenuto della tavola.
|
||||
_PRIMITIVE_RE = re.compile(rb"value\((prim_[a-z_]+)\)")
|
||||
|
||||
|
||||
def _decode(raw: bytes) -> str:
|
||||
try:
|
||||
return raw.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
return raw.decode("cp1252", errors="replace")
|
||||
|
||||
|
||||
def iter_texts(blob: bytes) -> "list[str]":
|
||||
"""Testi presenti in un blocco PGL decompresso, in ordine di comparsa."""
|
||||
texts, pos = [], 0
|
||||
while True:
|
||||
pos = blob.find(TEXT_MARK, pos)
|
||||
if pos < 0:
|
||||
return texts
|
||||
start = pos + len(TEXT_MARK) + 2
|
||||
declared = blob[pos + len(TEXT_MARK) + 1]
|
||||
|
||||
end = start
|
||||
limit = min(len(blob), start + MAX_TEXT)
|
||||
while end < limit and blob[end] != 0x00:
|
||||
if blob[end] not in _STRING_BYTES:
|
||||
break
|
||||
end += 1
|
||||
|
||||
length = end - start
|
||||
# Lo schema dichiara 2*len+2: se non torna, non e' un testo.
|
||||
if length and end < limit and blob[end] == 0x00 and declared == (2 * length + 2) % 256:
|
||||
texts.append(_decode(blob[start:end]))
|
||||
pos = end
|
||||
else:
|
||||
pos += 1
|
||||
|
||||
|
||||
#: Blocchi che contengono la rappresentazione grafica della tavola.
|
||||
GRAPHIC_BLOCKS = ("PictureObj", "PicturePrimdata", "PicturePersistTable")
|
||||
|
||||
|
||||
def read(data: bytes) -> dict:
|
||||
"""Estrae testi e primitive dai blocchi grafici della tavola."""
|
||||
blocks = [b for b in ugc.iter_blocks(data)
|
||||
if b[0].split("#", 1)[0] in GRAPHIC_BLOCKS]
|
||||
blob = b"".join(payload for _, payload, _ in blocks)
|
||||
compressed = sum(1 for _, _, was_compressed in blocks if was_compressed)
|
||||
|
||||
texts = iter_texts(blob)
|
||||
seen, unique = set(), []
|
||||
for text in texts:
|
||||
stripped = text.strip()
|
||||
if stripped and stripped not in seen:
|
||||
seen.add(stripped)
|
||||
unique.append(stripped)
|
||||
|
||||
primitives = sorted({m.decode() for m in _PRIMITIVE_RE.findall(blob)})
|
||||
|
||||
return {
|
||||
"graphic_blocks": len(blocks),
|
||||
"compressed_blocks": compressed,
|
||||
"decompressed_bytes": len(blob),
|
||||
"primitives": primitives,
|
||||
"text_count": len(texts),
|
||||
"texts": unique,
|
||||
}
|
||||
+22
-2
@@ -8,7 +8,7 @@ import re
|
||||
from collections import defaultdict
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from . import params, ugc
|
||||
from . import drawing, params, preview, ugc
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
METHOD = "ugc_native_reader"
|
||||
@@ -100,6 +100,7 @@ def extract(path: str, expected: "list[str] | None" = None) -> dict:
|
||||
data = handle.read()
|
||||
|
||||
container = ugc.parse(data)
|
||||
thumbnail = preview.find(data)
|
||||
occurrences = params.iter_occurrences(data)
|
||||
for occurrence in occurrences:
|
||||
occurrence.section = container.section_at(occurrence.offset)
|
||||
@@ -118,7 +119,7 @@ def extract(path: str, expected: "list[str] | None" = None) -> dict:
|
||||
name, {"value": None, "status": "parameter_not_present", "type": None}
|
||||
)
|
||||
|
||||
return {
|
||||
record = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"source_file": os.path.abspath(path),
|
||||
"file": {
|
||||
@@ -145,9 +146,28 @@ def extract(path: str, expected: "list[str] | None" = None) -> dict:
|
||||
"parameters_found": len(occurrences),
|
||||
"holds_model_parameters": holds_parameters,
|
||||
},
|
||||
# Metadati dell'anteprima: i byte dell'immagine restano fuori dal
|
||||
# record, si ottengono con read_preview().
|
||||
"preview": thumbnail.to_dict() if thumbnail else {"present": False},
|
||||
"parameters": parameters,
|
||||
}
|
||||
|
||||
# Un disegno non ha parametri di modello, ma ha il contenuto della tavola.
|
||||
if container.base_kind == "DRAWING":
|
||||
record["drawing"] = drawing.read(data)
|
||||
|
||||
return record
|
||||
|
||||
|
||||
def read_preview(path: str):
|
||||
"""L'anteprima incorporata nel file, o None se assente.
|
||||
|
||||
>>> shot = read_preview("9258400201.prt.12")
|
||||
>>> shot.save("anteprima.jpg")
|
||||
"""
|
||||
with open(path, "rb") as handle:
|
||||
return preview.find(handle.read())
|
||||
|
||||
|
||||
def latest_versions(paths: "list[str]") -> "list[str]":
|
||||
"""Tra piu' versioni dello stesso modello tiene solo la piu' recente."""
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
"""Decompressione LZW nel formato di compress(1), il vecchio `.Z` Unix.
|
||||
|
||||
Creo lo usa per alcuni blocchi interni dei file nativi, fra cui `PictureObj`
|
||||
che contiene la rappresentazione vettoriale dei disegni.
|
||||
|
||||
Rispetto a un LZW generico, compress ha una particolarita' che va riprodotta
|
||||
fedelmente: i codici sono letti a gruppi di otto e, quando la lunghezza del
|
||||
codice aumenta o si incontra un CLEAR, i bit residui del gruppo vengono
|
||||
scartati per riallinearsi. Senza questo dettaglio la decompressione produce
|
||||
qualche centinaio di byte corretti e poi diverge.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
MAGIC = b"\x1f\x9d"
|
||||
_CLEAR = 256
|
||||
_FIRST = 257
|
||||
INIT_BITS = 9
|
||||
|
||||
|
||||
class NotCompressed(ValueError):
|
||||
"""I byte non iniziano con la firma di compress."""
|
||||
|
||||
|
||||
def _codes(data: bytes, start: int):
|
||||
"""Genera (codice, posizione_byte_successiva) applicando le regole di
|
||||
allineamento di compress."""
|
||||
flags = data[start + 2]
|
||||
maxbits = flags & 0x1F
|
||||
block_mode = bool(flags & 0x80)
|
||||
|
||||
pos = start + 3
|
||||
code_len = INIT_BITS
|
||||
next_code = _FIRST if block_mode else 256
|
||||
group = [] # codici letti nel gruppo corrente di 8
|
||||
|
||||
while True:
|
||||
if not group:
|
||||
need = code_len # 8 codici da code_len bit = code_len byte
|
||||
chunk = data[pos : pos + need]
|
||||
if len(chunk) < need:
|
||||
return
|
||||
value = int.from_bytes(chunk, "little")
|
||||
mask = (1 << code_len) - 1
|
||||
group = [(value >> (i * code_len)) & mask for i in range(8)]
|
||||
pos += need
|
||||
|
||||
code = group.pop(0)
|
||||
reset = yield code, pos, next_code, code_len
|
||||
if reset == "clear":
|
||||
group = []
|
||||
code_len = INIT_BITS
|
||||
next_code = _FIRST
|
||||
continue
|
||||
if reset is not None:
|
||||
next_code = reset
|
||||
if next_code >= (1 << code_len) and code_len < maxbits:
|
||||
code_len += 1
|
||||
group = [] # il cambio di larghezza riallinea il gruppo
|
||||
|
||||
|
||||
def decompress(data: bytes, start: int = 0) -> "tuple[bytes, int]":
|
||||
"""Decomprime lo stream che inizia a `start`.
|
||||
|
||||
Restituisce (dati, byte_consumati). Si ferma senza sollevare eccezioni al
|
||||
primo codice non valido: i blocchi dentro i file Creo non sono terminati
|
||||
in modo esplicito, quindi la fine si riconosce cosi'.
|
||||
"""
|
||||
if data[start : start + 2] != MAGIC:
|
||||
raise NotCompressed("firma compress (1f 9d) assente")
|
||||
|
||||
block_mode = bool(data[start + 2] & 0x80)
|
||||
table: dict[int, bytes] = {i: bytes([i]) for i in range(256)}
|
||||
out = bytearray()
|
||||
prev: bytes | None = None
|
||||
next_code = _FIRST if block_mode else 256
|
||||
consumed = start + 3
|
||||
|
||||
gen = _codes(data, start)
|
||||
try:
|
||||
code, pos, _, _ = next(gen)
|
||||
while True:
|
||||
if block_mode and code == _CLEAR:
|
||||
table = {i: bytes([i]) for i in range(256)}
|
||||
next_code = _FIRST
|
||||
prev = None
|
||||
consumed = pos
|
||||
code, pos, _, _ = gen.send("clear")
|
||||
continue
|
||||
|
||||
if code in table:
|
||||
entry = table[code]
|
||||
elif prev is not None and code == next_code:
|
||||
entry = prev + prev[:1]
|
||||
else:
|
||||
break # fine reale dello stream
|
||||
|
||||
out += entry
|
||||
if prev is not None:
|
||||
table[next_code] = prev + entry[:1]
|
||||
next_code += 1
|
||||
prev = entry
|
||||
consumed = pos
|
||||
|
||||
code, pos, _, _ = gen.send(next_code)
|
||||
except StopIteration:
|
||||
pass
|
||||
|
||||
return bytes(out), consumed - start
|
||||
|
||||
|
||||
def find_streams(data: bytes, min_size: int = 256) -> "list[tuple[int, int, bytes]]":
|
||||
"""Individua e decomprime ogni blocco compress presente nel file.
|
||||
|
||||
Un blocco puo' contenere piu' stream concatenati e la lunghezza consumata
|
||||
e' una stima, quindi avanzare a salti farebbe perdere l'inizio di quello
|
||||
successivo. Proviamo ogni firma e teniamo gli stream che non ricadono
|
||||
dentro uno gia' accettato: cosi' nessun blocco viene saltato e i falsi
|
||||
riscontri, che decomprimono pochi byte, cadono da soli.
|
||||
"""
|
||||
candidates = []
|
||||
pos = 0
|
||||
while True:
|
||||
pos = data.find(MAGIC, pos)
|
||||
if pos < 0:
|
||||
break
|
||||
try:
|
||||
blob, used = decompress(data, pos)
|
||||
except (NotCompressed, IndexError):
|
||||
pos += 1
|
||||
continue
|
||||
if len(blob) >= min_size:
|
||||
candidates.append((pos, used, blob))
|
||||
pos += 1
|
||||
|
||||
# A parita' di area coperta vince lo stream che produce piu' dati.
|
||||
candidates.sort(key=lambda item: (item[0], -len(item[2])))
|
||||
found: list = []
|
||||
covered_until = -1
|
||||
for start, used, blob in candidates:
|
||||
if start < covered_until:
|
||||
continue
|
||||
found.append((start, used, blob))
|
||||
covered_until = start + used
|
||||
return found
|
||||
@@ -0,0 +1,115 @@
|
||||
"""Estrazione dell'anteprima incorporata nei file nativi Creo.
|
||||
|
||||
Creo salva nel file una miniatura JPEG dell'ultimo salvataggio: il render
|
||||
ombreggiato del pezzo per i modelli, l'immagine della tavola per i disegni.
|
||||
E' la stessa immagine che compare nella finestra "Apri" di Creo.
|
||||
|
||||
La miniatura non e' referenziata dall'indice UGC_TOC, quindi la si individua
|
||||
per firma. Per non scambiare dati binari qualsiasi per un'immagine, ogni
|
||||
candidato viene validato percorrendone i segmenti JPEG.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
|
||||
SOI = b"\xff\xd8\xff"
|
||||
EOI = b"\xff\xd9"
|
||||
|
||||
#: Marcatori "Start of Frame": contengono le dimensioni dell'immagine.
|
||||
_SOF = {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}
|
||||
_SOS = 0xDA
|
||||
_STANDALONE = {0xD8, 0xD9, 0x01} | set(range(0xD0, 0xD8))
|
||||
|
||||
MIN_BYTES = 512
|
||||
|
||||
|
||||
@dataclass
|
||||
class Preview:
|
||||
"""Una miniatura trovata nel file."""
|
||||
|
||||
data: bytes
|
||||
offset: int
|
||||
width: int
|
||||
height: int
|
||||
format: str = "jpeg"
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"present": True,
|
||||
"format": self.format,
|
||||
"width": self.width,
|
||||
"height": self.height,
|
||||
"bytes": len(self.data),
|
||||
"offset": hex(self.offset),
|
||||
}
|
||||
|
||||
def save(self, path: str) -> None:
|
||||
with open(path, "wb") as handle:
|
||||
handle.write(self.data)
|
||||
|
||||
|
||||
def _parse_jpeg(data: bytes, start: int) -> Preview | None:
|
||||
"""Percorre i segmenti a partire da `start`. Restituisce None se la
|
||||
sequenza non e' un JPEG completo e coerente."""
|
||||
pos = start + 2 # oltre il SOI
|
||||
size = None
|
||||
|
||||
while pos < len(data) - 3:
|
||||
if data[pos] != 0xFF:
|
||||
return None
|
||||
marker = data[pos + 1]
|
||||
|
||||
if marker == 0xFF: # byte di riempimento
|
||||
pos += 1
|
||||
continue
|
||||
if marker in _STANDALONE:
|
||||
pos += 2
|
||||
continue
|
||||
|
||||
length = struct.unpack(">H", data[pos + 2 : pos + 4])[0]
|
||||
if length < 2:
|
||||
return None
|
||||
|
||||
if marker in _SOF and pos + 9 <= len(data):
|
||||
height, width = struct.unpack(">HH", data[pos + 5 : pos + 9])
|
||||
size = (width, height)
|
||||
|
||||
if marker == _SOS:
|
||||
# I dati compressi seguono il segmento: il byte FF che vi compare
|
||||
# e' sempre preceduto da escape, quindi il primo EOI e' la fine.
|
||||
end = data.find(EOI, pos + 2 + length)
|
||||
if end < 0 or size is None:
|
||||
return None
|
||||
blob = data[start : end + 2]
|
||||
if len(blob) < MIN_BYTES:
|
||||
return None
|
||||
return Preview(blob, start, size[0], size[1])
|
||||
|
||||
pos += 2 + length
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def find_all(data: bytes) -> "list[Preview]":
|
||||
"""Tutte le miniature valide presenti nel file, in ordine di comparsa."""
|
||||
found, pos = [], 0
|
||||
while True:
|
||||
pos = data.find(SOI, pos)
|
||||
if pos < 0:
|
||||
return found
|
||||
preview = _parse_jpeg(data, pos)
|
||||
if preview:
|
||||
found.append(preview)
|
||||
pos += len(preview.data)
|
||||
else:
|
||||
pos += 1
|
||||
|
||||
|
||||
def find(data: bytes) -> Preview | None:
|
||||
"""La miniatura piu' grande, che e' quella a risoluzione migliore."""
|
||||
candidates = find_all(data)
|
||||
if not candidates:
|
||||
return None
|
||||
return max(candidates, key=lambda p: (p.width * p.height, len(p.data)))
|
||||
@@ -180,6 +180,39 @@ def parse(data: bytes) -> Container:
|
||||
return container
|
||||
|
||||
|
||||
#: Oltre l'indice, il corpo prosegue con blocchi introdotti da una riga
|
||||
#: "#Nome". I nomi possono contenere ':' e '#' (es. "PicturePrimdata#0").
|
||||
_BLOCK_RE = re.compile(rb"\n#([A-Za-z][A-Za-z0-9_:#]{2,40})\n")
|
||||
|
||||
|
||||
def iter_blocks(data: bytes) -> "list[tuple[str, bytes, bool]]":
|
||||
"""I blocchi che seguono l'indice, come (nome, contenuto, compresso).
|
||||
|
||||
Il contenuto e' gia' decompresso quando serve: lo stesso blocco puo'
|
||||
essere memorizzato compresso in un file e in chiaro in un altro, quindi
|
||||
la scelta va fatta guardando i byte, non il nome.
|
||||
"""
|
||||
from . import lzw
|
||||
|
||||
marks = [(m.start(), m.end(), m.group(1).decode("latin-1"))
|
||||
for m in _BLOCK_RE.finditer(data)]
|
||||
|
||||
blocks = []
|
||||
for index, (start, body, name) in enumerate(marks):
|
||||
end = marks[index + 1][0] if index + 1 < len(marks) else len(data)
|
||||
payload = data[body:end]
|
||||
if payload.startswith(lzw.MAGIC):
|
||||
try:
|
||||
payload, _ = lzw.decompress(payload)
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
else:
|
||||
blocks.append((name, payload, True))
|
||||
continue
|
||||
blocks.append((name, payload, False))
|
||||
return blocks
|
||||
|
||||
|
||||
def read(path) -> Container:
|
||||
with open(path, "rb") as handle:
|
||||
return parse(handle.read())
|
||||
|
||||
+86
-1
@@ -12,7 +12,8 @@ import glob
|
||||
import os
|
||||
import unittest
|
||||
|
||||
from creoparams.extract import extract, latest_versions, split_filename
|
||||
from creoparams.extract import (extract, latest_versions, read_preview,
|
||||
split_filename)
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
@@ -99,6 +100,30 @@ class TestEstrazione(unittest.TestCase):
|
||||
record["model"]["name"],
|
||||
)
|
||||
|
||||
def test_anteprima_presente_e_valida(self):
|
||||
for path in self.parts:
|
||||
name = os.path.basename(path)
|
||||
with self.subTest(file=name):
|
||||
shot = read_preview(path)
|
||||
self.assertIsNotNone(shot, "anteprima non trovata")
|
||||
self.assertTrue(shot.data.startswith(b"\xff\xd8\xff"))
|
||||
self.assertTrue(shot.data.endswith(b"\xff\xd9"))
|
||||
self.assertGreater(shot.width, 0)
|
||||
self.assertGreater(shot.height, 0)
|
||||
# Le dimensioni dichiarate devono corrispondere al record.
|
||||
declared = self.records[name]["preview"]
|
||||
self.assertTrue(declared["present"])
|
||||
self.assertEqual(declared["width"], shot.width)
|
||||
|
||||
def test_una_sola_anteprima_per_file(self):
|
||||
"""Se il rilevatore trovasse piu' candidati, starebbe scambiando dati
|
||||
binari qualsiasi per immagini."""
|
||||
from creoparams import preview
|
||||
for path in self.parts:
|
||||
with self.subTest(file=os.path.basename(path)):
|
||||
with open(path, "rb") as handle:
|
||||
self.assertEqual(len(preview.find_all(handle.read())), 1)
|
||||
|
||||
def test_parametri_feature_separati(self):
|
||||
"""I parametri dei fori non devono finire tra quelli del modello."""
|
||||
for name, record in self.records.items():
|
||||
@@ -108,5 +133,65 @@ class TestEstrazione(unittest.TestCase):
|
||||
self.assertEqual(entry["owner"], "feature")
|
||||
|
||||
|
||||
class TestDisegni(unittest.TestCase):
|
||||
"""I .drw non hanno parametri di modello ma contengono la tavola."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
paths = latest_versions(glob.glob(os.path.join(HERE, "*.drw*")))
|
||||
cls.drawings = [p for p in paths
|
||||
if split_filename(os.path.basename(p))[1] == "drw"]
|
||||
if not cls.drawings:
|
||||
raise unittest.SkipTest("nessun file .drw nella cartella")
|
||||
cls.records = {os.path.basename(p): extract(p) for p in cls.drawings}
|
||||
|
||||
def test_riconosciuti_come_disegni(self):
|
||||
for name, record in self.records.items():
|
||||
with self.subTest(file=name):
|
||||
self.assertEqual(record["model"]["kind"], "DRAWING")
|
||||
self.assertFalse(record["extraction"]["holds_model_parameters"])
|
||||
|
||||
def test_testi_estratti(self):
|
||||
for name, record in self.records.items():
|
||||
with self.subTest(file=name):
|
||||
sheet = record["drawing"]
|
||||
self.assertGreater(sheet["decompressed_bytes"], 10_000)
|
||||
self.assertGreater(len(sheet["texts"]), 20)
|
||||
|
||||
def test_voci_del_cartiglio(self):
|
||||
"""Il cartiglio aziendale deve comparire in ogni tavola."""
|
||||
attese = {"CODICE", "DISEGNATO", "STAMPIGLIATURA", "SCALA"}
|
||||
for name, record in self.records.items():
|
||||
with self.subTest(file=name):
|
||||
testi = set(record["drawing"]["texts"])
|
||||
self.assertTrue(attese.issubset(testi),
|
||||
f"mancano {attese - testi}")
|
||||
|
||||
def test_primitive_grafiche(self):
|
||||
for name, record in self.records.items():
|
||||
with self.subTest(file=name):
|
||||
prims = record["drawing"]["primitives"]
|
||||
self.assertIn("prim_text", prims)
|
||||
self.assertIn("prim_line", prims)
|
||||
|
||||
|
||||
class TestLzw(unittest.TestCase):
|
||||
def test_round_trip_noto(self):
|
||||
"""Verifica il decoder su uno stream prodotto da compress(1), se
|
||||
disponibile nel sistema."""
|
||||
import shutil
|
||||
import subprocess
|
||||
from creoparams import lzw
|
||||
tool = shutil.which("compress") or shutil.which("gzip")
|
||||
if not tool:
|
||||
self.skipTest("nessun compressore disponibile")
|
||||
if "gzip" in tool:
|
||||
self.skipTest("gzip non produce il formato .Z")
|
||||
payload = b"PGL " * 500 + bytes(range(256)) * 4
|
||||
proc = subprocess.run([tool, "-c"], input=payload, capture_output=True)
|
||||
out, _ = lzw.decompress(proc.stdout)
|
||||
self.assertEqual(out, payload)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user