NX Journal로 부품별 질량·밀도 CSV 내보내기｜누락값과 단위 검증

# -*- coding: utf-8 -*-
"""NX Open Python Journal - assembly occurrence mass audit (read-only design).

Test target: NX 2206/2212/2306+ API surfaces; NOT run in NX in this environment.
Does not create expressions/features, modify bodies, assign materials, or save parts.
Reports measured NX physical-material mass (kg), volume (mm^3), derived density
(kg/m^3) only for solids with an explicitly assigned physical material.

Run within NX: Tools > Journal > Play (UI labels may differ by release).
Output: $HOME/nx_mass_audit_YYYYMMDD_HHMMSS.csv
IMPORTANT: trial on a backup/non-confidential assembly first.
"""
import csv
import datetime
import math
import os

import NXOpen

ACCURACY = 0.99
OUTPUT_DIR = os.path.expanduser("~")
FIELDS = [
    "component_path", "part_number", "part_units", "occurrence_count",
    "body_count", "valid_body_count", "material_names", "volume_mm3",
    "density_kg_m3", "mass_kg", "partial_mass_kg", "status", "reason"
]


def safe_text(value):
    """Prevent CSV formula execution when opened in spreadsheet programs."""
    s = str(value if value is not None else "").replace("\x00", "")
    if s and s[0] in ("=", "+", "-", "@", "\t", "\r", "\n"):
        return "'" + s
    return s


def default_row(path, part_name="", unit_name=""):
    return dict(component_path=path, part_number=part_name,
                part_units=unit_name, occurrence_count="1", body_count="0",
                valid_body_count="0", material_names="", volume_mm3="",
                density_kg_m3="", mass_kg="", partial_mass_kg="",
                status="", reason="")


def number(value):
    return "{:.9g}".format(value) if value is not None and math.isfinite(value) else ""


def get_units(part):
    uc = part.UnitCollection
    return [uc.FindObject(name) for name in (
        "SquareMilliMeter", "CubicMilliMeter", "Kilogram", "MilliMeter", "Newton")]


def get_part_units(part):
    try:
        return str(part.PartUnits)
    except Exception:
        return "unknown (measured output: mm3/kg)"


def read_bodies(part):
    return [body for body in part.Bodies if body.IsSolidBody]


def read_physical_material(part, body):
    """No fallback density: if unsupported/unassigned, report explicitly."""
    try:
        mat = part.MaterialManager.PhysicalMaterials.AskMaterialOfObject(body)
    except AttributeError as exc:
        return None, "MATERIAL_API_UNAVAILABLE: " + str(exc)
    except Exception as exc:
        return None, "MATERIAL_QUERY_FAILED: " + str(exc)
    if mat is None:
        return None, "MISSING_MATERIAL"
    try:
        return str(mat.Name), None
    except Exception:
        return "assigned_material", None


def measure_one(part, body, units):
    """Temporary measurement object; no CreateFeature and no saved part changes."""
    measurement = None
    try:
        measurement = part.MeasureManager.NewMassProperties(units, ACCURACY, [body])
        return float(measurement.Volume), float(measurement.Mass)
    finally:
        if measurement is not None:
            measurement.Dispose()


def audit_part(part, path):
    row = default_row(path, safe_text(part.Name), get_part_units(part))
    try:
        bodies = read_bodies(part)
        row["body_count"] = str(len(bodies))
        if not bodies:
            row["status"] = "NO_SOLID_BODY"
            row["reason"] = "No solid bodies in prototype part"
            return row
        units = get_units(part)
    except Exception as exc:
        row["status"] = "PART_READ_FAILED"
        row["reason"] = safe_text(str(exc))
        return row

    volumes, masses, materials, errors = [], [], [], []
    for i, body in enumerate(bodies):
        name, problem = read_physical_material(part, body)
        if problem:
            errors.append("body{}:{}".format(i + 1, problem))
            continue
        try:
            vol, mass = measure_one(part, body, units)
            if not (math.isfinite(vol) and math.isfinite(mass) and vol > 0 and mass >= 0):
                raise ValueError("Invalid volume or mass")
            volumes.append(vol)
            masses.append(mass)
            materials.append(name)
        except Exception as exc:
            errors.append("body{}:MEASURE_FAILED:{}".format(i + 1, str(exc)))

    row["valid_body_count"] = str(len(volumes))
    row["material_names"] = safe_text("; ".join(sorted(set(materials))))
    if volumes:
        row["partial_mass_kg"] = number(sum(masses))
    if errors:
        row["status"] = "INCOMPLETE"
        row["reason"] = safe_text(" | ".join(errors))
        return row  # no total mass/density for a partially measured part

    volume_sum, mass_sum = sum(volumes), sum(masses)
    row["volume_mm3"] = number(volume_sum)
    row["mass_kg"] = number(mass_sum)
    # effective density is mass-weighted at the occurrence level, NOT a material card value.
    row["density_kg_m3"] = number(mass_sum / (volume_sum * 1e-9))
    row["status"] = "OK_MIXED_MATERIAL" if len(set(materials)) > 1 else "OK"
    row["reason"] = "effective density for multiple materials" if len(set(materials)) > 1 else ""
    return row


def walk_component(comp, path, rows, cache, visited=None):
    # Each occurrence is recorded, but prototype properties can be cached.
    name = safe_text(getattr(comp, "DisplayName", getattr(comp, "Name", "component")))
    here = path + "/" + name if path else name
    try:
        if comp.IsSuppressed:
            row = default_row(here)
            row.update(status="SUPPRESSED", reason="Excluded by current arrangement")
            rows.append(row)
            return
    except Exception as exc:
        row = default_row(here)
        row.update(status="SUPPRESSION_UNKNOWN", reason=safe_text(str(exc)))
        rows.append(row)
        return
    try:
        children = list(comp.GetChildren())
    except Exception as exc:
        row = default_row(here)
        row.update(status="CHILD_QUERY_FAILED", reason=safe_text(str(exc)))
        rows.append(row)
        return
    try:
        proto = comp.Prototype
        part = proto.OwningPart
        if part is None:
            raise ValueError("Prototype owning part is not loaded")
        key = int(part.Tag)
        if key not in cache:
            cache[key] = audit_part(part, "")
        report = dict(cache[key])
        report["component_path"] = here
        # Skip empty subassembly container but continue into its children.
        if not (children and report["status"] == "NO_SOLID_BODY"):
            rows.append(report)
    except Exception as exc:
        row = default_row(here)
        row.update(status="UNLOADED_COMPONENT", reason=safe_text(str(exc)))
        rows.append(row)
    for child in children:
        walk_component(child, here, rows, cache)


def main():
    session = NXOpen.Session.GetSession()
    lw = session.ListingWindow
    lw.Open()
    work = session.Parts.Work
    if work is None:
        lw.WriteLine("[ERROR] Open a part or an assembly first.")
        return
    rows = []
    root = work.ComponentAssembly.RootComponent
    if root is None or not list(root.GetChildren()):
        rows.append(audit_part(work, safe_text(work.Name)))
    else:
        cache = {}
        for child in root.GetChildren():
            walk_component(child, safe_text(work.Name), rows, cache)

    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
    target = os.path.join(OUTPUT_DIR, "nx_mass_audit_" + timestamp + ".csv")
    with open(target, "w", newline="", encoding="utf-8-sig") as f:
        writer = csv.DictWriter(f, FIELDS, extrasaction="ignore")
        writer.writeheader()
        writer.writerows(rows)
    passed = [r for r in rows if r["status"] in ("OK", "OK_MIXED_MATERIAL")]
    lw.WriteLine("[DONE] CSV written: " + target)
    lw.WriteLine("[COUNT] rows={}, fully-measured={}".format(len(rows), len(passed)))
    lw.WriteLine("[INFO] All numerical mass results are NX-measured, not independently verified.")
    lw.WriteLine("[INFO] Failed/suppressed/unloaded rows excluded from total.")
    lw.WriteLine("[TOTAL] kg={:.9g}".format(sum(float(r["mass_kg"]) for r in passed)))


if __name__ == "__main__":
    main()