diff --git a/.gitignore b/.gitignore index ce5e0e7..6892384 100644 --- a/.gitignore +++ b/.gitignore @@ -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/ diff --git a/README.md b/README.md index 1cfa865..a605148 100644 --- a/README.md +++ b/README.md @@ -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 | diff --git a/creoparams/cli.py b/creoparams/cli.py index 4202930..55820b1 100644 --- a/creoparams/cli.py +++ b/creoparams/cli.py @@ -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) diff --git a/creoparams/drawing.py b/creoparams/drawing.py new file mode 100644 index 0000000..7e2018f --- /dev/null +++ b/creoparams/drawing.py @@ -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 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, + } diff --git a/creoparams/extract.py b/creoparams/extract.py index 940d49a..296ceb0 100644 --- a/creoparams/extract.py +++ b/creoparams/extract.py @@ -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.""" diff --git a/creoparams/lzw.py b/creoparams/lzw.py new file mode 100644 index 0000000..acfa3bc --- /dev/null +++ b/creoparams/lzw.py @@ -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 diff --git a/creoparams/preview.py b/creoparams/preview.py new file mode 100644 index 0000000..03f6aa6 --- /dev/null +++ b/creoparams/preview.py @@ -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))) diff --git a/creoparams/ugc.py b/creoparams/ugc.py index 1df663c..a4d34b4 100644 --- a/creoparams/ugc.py +++ b/creoparams/ugc.py @@ -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()) diff --git a/test_regression.py b/test_regression.py index 42a840f..6f45676 100644 --- a/test_regression.py +++ b/test_regression.py @@ -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()