pdm/creoparams/extract.py

"""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 drawing, params, preview, 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:  # noqa: C901
    """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)
    thumbnail = preview.find(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}
            )

    record = {
        "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,
        },
        # 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)

    # Se accanto al modello c'e' una mesh esportata, ne ricaviamo ingombro e
    # volume: sono le uniche grandezze geometriche ottenibili senza Creo.
    if holds_parameters:
        try:
            geometry = read_mesh(path)
        except (OSError, ValueError):
            geometry = None
        if geometry is not None:
            record["geometry"] = geometry.to_dict()
            record["geometry"]["source_kind"] = "mesh esportata"

    return record


#: Estensioni di mesh cercate accanto al modello, in ordine di preferenza.
MESH_EXTENSIONS = (".stl", ".STL")


def find_mesh(path: str) -> "str | None":
    """Cerca una mesh esportata accanto al modello.

    La geometria dei .prt non e' leggibile senza la libreria di PTC, quindi
    la forma arriva da un file esportato con lo stesso nome: 9258400207.prt.11
    -> 9258400207.stl.
    """
    directory = os.path.dirname(os.path.abspath(path))
    stem, _, _ = split_filename(os.path.basename(path))
    for extension in MESH_EXTENSIONS:
        candidate = os.path.join(directory, stem + extension)
        if os.path.isfile(candidate):
            return candidate
    return None


def read_mesh(path: str):
    """La mesh associata al modello, o None se non e' stata esportata."""
    from . import mesh as mesh_module

    found = find_mesh(path)
    return mesh_module.read_stl(found) if found else None


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."""
    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())