Fusion - the Fusion 360 Bridge

@@ -2,11 +2,12 @@ import csv from pathlib import Path import tempfile+import time import adsk.core import adsk.fusion  from . import bom_views-from .assembly_state import metadata+from .assembly_state import metadata, native_properties from .core import identity, items, occurrence_suppression, ref, require, target from .exporting import publish_staged from .configurations import provenance, snapshot as configuration_snapshot, source_link@@ -39,7 +40,11 @@ def wire_length_mm(comp):     return round(best * 10, 1) if best is not None else stock_length_mm(comp)  -def generate(design, value, args):+def generate(design, value, args, started=None, budget_seconds=None):+    """budget_seconds, counted from started: past it, part numbers and descriptions not read yet are left+    for the next run (each is a cloud read of about 0.5 s on a cloud assembly: 280 of them took 282 s)."""+    deadline = started + budget_seconds if started is not None and budget_seconds else None+    pending = []     configuration = configuration_snapshot(design, strict=True)     table = configuration["table"]     columns = {column["id"]: column for column in table["columns"]} if table else {}@@ -59,6 +64,18 @@ def generate(design, value, args):     physical = {"mass": 0.0, "volume": 0.0, "bodyInstances": 0}     accuracy = adsk.fusion.CalculationAccuracy.HighCalculationAccuracy     budget = {"visited": 0}+    # what the walks read about a definition, read once per definition instead of once per instance+    # (a 4,579-occurrence assembly has 1,249 definitions): every instance of a definition weighs the same+    counts, physicals, metas, kits, refs = {}, {}, {}, {}, {}+    def body_count(comp, key):+        if key not in counts:+            counts[key] = comp.bRepBodies.count+        return counts[key]+    def direct_physical(comp, key):+        if key not in physicals:+            props = [b.getPhysicalProperties(accuracy) for b in items(comp.bRepBodies)]+            physicals[key] = (sum(p.mass for p in props), sum(p.volume * 1000 for p in props), len(props))+        return physicals[key]     def visit(comp, depth, ancestry):         budget["visited"] += 1         require(depth <= 100 and budget["visited"] <= 20000, "bom_budget", "Assembly exceeds combined purchasing/physical traversal budget.")@@ -69,12 +86,10 @@ def generate(design, value, args):     # once per physical instance, including root bodies and repeated children.     def physical_walk(comp, occurrence=None, depth=0, ancestry=()):         key = visit(comp, depth, ancestry)-        bodies = items(occurrence.bRepBodies if occurrence else comp.bRepBodies)-        for body in bodies:-            props = body.getPhysicalProperties(accuracy)-            physical["mass"] += props.mass-            physical["volume"] += props.volume * 1000-            physical["bodyInstances"] += 1+        mass, volume, bodies = direct_physical(comp, key)+        physical["mass"] += mass+        physical["volume"] += volume+        physical["bodyInstances"] += bodies         children = items(occurrence.childOccurrences if occurrence else comp.occurrences)         for child in children:             if suppression(child) is None:@@ -87,11 +102,19 @@ def generate(design, value, args):         if reason:             excluded.append({"path": path, "reason": reason})             return-        if occurrence and bom_views.is_library(comp.name, comp.bRepBodies.count) and not args.get("includeLibraries", False):+        if occurrence and bom_views.is_library(comp.name, body_count(comp, key)) and not args.get("includeLibraries", False):             # a palette of parts kept to copy from: counting it adds one of everything it holds             excluded.append({"path": path, "reason": "library_group"})             return-        data = metadata(comp, args.get("nativeProperties", True))+        if key not in metas:+            native = args.get("nativeProperties", True)+            if native and deadline is not None and time.perf_counter() > deadline and native_properties(comp, read=False) is None:+                native = False+                pending.append(key)+            metas[key] = metadata(comp, native)+            if key in pending:+                metas[key]["nativePending"] = True+        data = dict(metas[key])         if data.get("excludeFromBom", False):             excluded.append({"path": path, "reason": "excludeFromBom"})             return@@ -104,26 +127,29 @@ def generate(design, value, args):         # someone said otherwise: without this a 3,000-occurrence cell listed 90 resistors and every screw         # inside a machine's vendor model         explicit = "asPurchased" in data or data.get("makeBuy") == "make"-        kit_reason = None if explicit or not occurrence or not children or not args.get("inferKits", True) else \-            bom_views.inferred_kit(comp.name, [child.component.name for child in children])+        if not (explicit or not occurrence or not children or not args.get("inferKits", True)) and key not in kits:+            kits[key] = bom_views.inferred_kit(comp.name, [child.component.name for child in children])+        kit_reason = None if explicit or not occurrence or not children or not args.get("inferKits", True) else kits[key]         if kit_reason:             data = {**data, "kitReason": kit_reason}         purchased = (bool(data.get("asPurchased", False)) or bool(kit_reason)) and not args.get("explodePurchased", False)+        if key not in refs:+            refs[key] = ref(comp, value)         row = {"path": path, "parentPath": parent_path, "depth": depth, "componentId": key, "metadata": data,-               "bodyCount": comp.bRepBodies.count, "childCount": len(children), "asPurchased": purchased,-               "componentRef": ref(comp, value), "occurrenceRef": ref(occurrence, value) if occurrence else None}+               "bodyCount": body_count(comp, key), "childCount": len(children), "asPurchased": purchased,+               "componentRef": dict(refs[key]), "occurrenceRef": ref(occurrence, value) if occurrence else None}         instances.append(row)-        if comp.bRepBodies.count or purchased:+        if body_count(comp, key) or purchased:             if key not in parts:                 props = comp.getPhysicalProperties(accuracy) if purchased else None-                direct_properties = [] if purchased else [b.getPhysicalProperties(accuracy) for b in items(comp.bRepBodies)]+                direct_mass, direct_volume, _ = (0.0, 0.0, 0) if purchased else direct_physical(comp, key)                 kind = None if purchased else bom_views.category(data)                 parts[key] = {"componentId": key, **data, "quantity": 0, "occurrencePaths": [],                               "scope": "component_subtree" if purchased else "direct_component_bodies",-                              "bodyCount": comp.bRepBodies.count,+                              "bodyCount": body_count(comp, key),                               "lengthMm": stock_length_mm(comp) if kind in bom_views.LINEAR else wire_length_mm(comp) if kind == "wire" else None,-                              "unitMass": props.mass if props else sum(p.mass for p in direct_properties),-                              "unitVolume": props.volume * 1000 if props else sum(p.volume * 1000 for p in direct_properties)}+                              "unitMass": props.mass if props else direct_mass,+                              "unitVolume": props.volume * 1000 if props else direct_volume}             parts[key]["quantity"] += 1             parts[key]["occurrencePaths"].append(path)         if not purchased:@@ -153,20 +179,28 @@ def generate(design, value, args):             "physical": {**physical, "scope": "all unsuppressed direct-body instances; BOM exclusions do not remove physical mass",                          "materialBasis": "Fusion assigned/default material; supplier provenance does not certify modeled material density"},             "units": {"mass": "kg", "volume": "mm3"}, "calculationAccuracy": "high",-            "identityScope": "running add-in session; supplier metadata persists in native archives"}+            "identityScope": "running add-in session; supplier metadata persists in native archives",+            **({"nativePending": len(pending),+                "nativeHint": "%d part numbers and descriptions were left unread (each is a cloud read of about 0.5 s): run bom "+                              "again and it continues from what this run read, or pass budgetSeconds to wait longer, or "+                              "nativeProperties:false to skip them." % len(pending)} if pending else {})}   def bom(app, args):+    started = time.perf_counter()     _, design, value = target(app, args)-    return generate(design, value, args)+    return generate(design, value, args, started, args.get("budgetSeconds", 40))   def bom_export(app, args):+    started = time.perf_counter()     _, design, value = target(app, args, True, model_write=False)     destination = Path(args["filename"])     require(destination.is_absolute() and destination.suffix.lower() == ".csv" and not destination.is_symlink() and not destination.is_dir(), "invalid_destination", "BOM destination must be an absolute regular .csv path.")     require(args.get("overwrite", False) or not destination.exists(), "destination_exists", "Pass overwrite:true to replace an existing CSV.")-    result = generate(design, value, args)+    # journaled, but its first answer still has ab's 60 s, and Fusion stays busy while it reads: the same+    # budget as bom, and a CSV written past it says how many part numbers it left blank (export again)+    result = generate(design, value, args, started, args.get("budgetSeconds", 40))     destination.parent.mkdir(parents=True, exist_ok=True)     published = {}     try:@@ -216,6 +250,7 @@ def bom_export(app, args):             raise         published["cleanupWarning"] = str(exc)     return {"target": value, "partCount": result["partCount"], "mode": args.get("mode", "flat"), "rowCount": len(rows),+            **{k: result[k] for k in ("nativePending", "nativeHint") if k in result},             "totals": result["totals"], "warningCount": len(result["warnings"]),             "excludedCount": len(result["excluded"]),             "calculationAccuracy": result["calculationAccuracy"], "configuration": result["configuration"], **published}

@@ -55,6 +55,7 @@ def concrete(entity): # many minutes. The indexes only narrow the candidates; equality is still native object equality. _identity_buckets = {} _identity_tokens = {}+_identity_token_of = {}  # identity key -> the token read when the entity was registered MAX_IDENTITIES = 100000  @@ -85,23 +86,40 @@ def identity(entity):     if not _entities and (_identity_buckets or _identity_tokens):         _identity_buckets.clear()  # the registry was reset         _identity_tokens.clear()-    bucket, token = _identity_bucket(entity), _identity_token(entity)-    for candidates in (_identity_buckets.get(bucket, ()), _identity_tokens.get(token, ()) if token else ()):-        for obj, key in candidates:-            if obj.isValid and same(obj, entity):-                filed = _identity_buckets.setdefault(bucket, [])-                if not any(k == key for _, k in filed):-                    filed.append((obj, key))-                return key+        _identity_token_of.clear()+    bucket = _identity_bucket(entity)+    for obj, key in _identity_buckets.get(bucket, ()):+        if obj.isValid and same(obj, entity):+            return key+    # only a miss (new, or renamed since) reads the token: an occurrence's entityToken costs about 2 ms,+    # which was a quarter of every revision digest on a 4,579-occurrence assembly+    token = _identity_token(entity)+    for obj, key in _identity_tokens.get(token, ()) if token else ():+        if obj.isValid and same(obj, entity):+            _identity_buckets.setdefault(bucket, []).append((obj, key))+            _identity_token_of[key] = token[1]+            return key     require(len(_entities) < MAX_IDENTITIES, "identity_budget", "Session entity budget exhausted.")     key = str(uuid.uuid4())     _entities.append((entity, key))     _identity_buckets.setdefault(bucket, []).append((entity, key))     if token:         _identity_tokens.setdefault(token, []).append((entity, key))+        _identity_token_of[key] = token[1]     return key  +def entity_token(entity):+    """The token a reference carries. Reading an occurrence's costs about 5 ms, which was over half of+    inspect on a 4,579-occurrence assembly, so an occurrence registered this session reuses the token+    read when it was registered: Fusion resolves every token it gave an entity to that same entity."""+    if entity.objectType.endswith("::Occurrence"):+        token = _identity_token_of.get(identity(entity))+        if token:+            return token+    return entity.entityToken++ def document_identity(doc):     require(threading.current_thread() is threading.main_thread(), "wrong_thread", "Document identity requires the Fusion main thread.")     require(doc is not None and doc.isValid, "wrong_document", "Document identity requires a live document.")@@ -157,6 +175,25 @@ def reference_version(component):         return {"error": str(error)[:120]}  +def components_by_occurrence(design, all_occurrences, size):+    """Every component of the design, found through the occurrences the digest reads anyway: Fusion keeps+    a component only while it has an occurrence (the root aside). allComponents.item() costs about 5 ms a+    component (6 s of a 1,249-component digest); occurrence.component about 0.05 ms. If the counts ever+    disagree, the slow full list is used."""+    root = design.rootComponent+    found, seen = [root], {identity(root)}+    for occurrence in all_occurrences:+        comp = occurrence.component+        key = identity(comp)+        if key not in seen:+            seen.add(key)+            found.append(comp)+    if len(found) == size:+        return found+    collection = design.allComponents+    return [collection.item(index) for index in range(size)]++ def revision(design):     """Initialize all mesh collections, then capture the unchanged revision graph. @@ -189,7 +226,8 @@ def revision(design):         collection = design.allComponents         size = collection.count         require(type(size) is int and 0 <= size <= MAX_ENTITIES, "inspection_budget", "Component collection exceeds inspection budget.")-        component_handles = [collection.item(index) for index in range(size)]+        all_occurrences = items(design.rootComponent.allOccurrences)+        component_handles = components_by_occurrence(design, all_occurrences, size)         mesh_collections = []  # Retain all native collection handles through the full digest.         # A component inserted by reference belongs to its own document's Design: it is read-only         # here, so it enters the revision by its source version instead of its native internals.@@ -223,7 +261,7 @@ def revision(design):                                [[identity(s), str(s.revisionId)] for s in items(comp.sketches)],                                component_state(comp), [feature_data(f, order=i) for i, f in enumerate(items(comp.features))] if hasattr(comp, "features") else [], construction_state(comp)])         occurrences = [[identity(o), identity(o.component), list(o.transform2.asArray()),-                        occurrence_suppression(o), occurrence_state(o)] for o in items(design.rootComponent.allOccurrences)]+                        occurrence_suppression(o), occurrence_state(o)] for o in all_occurrences]         params = [[p.name, p.expression, p.unit] for p in items(design.allParameters)]         result = digest([components, occurrences, params, design.timeline.count, design.timeline.markerPosition,                          configuration_snapshot(design)])@@ -337,8 +375,8 @@ def ref(entity, target_value, occurrence=None):     native = entity if is_occurrence else (getattr(entity, "nativeObject", None) or entity)     context = None if is_occurrence else (occurrence or getattr(entity, "assemblyContext", None))     result = {"documentId": target_value["documentId"], "revision": target_value["revision"],-            "token": native.entityToken, "objectType": native.objectType,-            "occurrenceToken": context.entityToken if context else None,+            "token": entity_token(native), "objectType": native.objectType,+            "occurrenceToken": entity_token(context) if context else None,             "occurrencePath": context.fullPathName if context else (entity.fullPathName if is_occurrence else None)}     if native.objectType == "adsk::fusion::Component" and same(native, native.parentDesign.rootComponent):         # Fusion can emit a root token that findEntityByToken cannot resolve.@@ -353,10 +391,44 @@ def ref(entity, target_value, occurrence=None):     return result  +def occurrence_at(design, path):+    """The root-context occurrence at a full path, found level by level with a few reads instead of by+    reading every occurrence's path (about 0.3 s per lookup on a 4,579-occurrence assembly). A name can+    itself hold '+', so each level takes the shortest name that exists. None when nothing is there."""+    level, parts = design.rootComponent.occurrences, path.split("+") if path else []+    try:+        while parts:+            for size in range(1, len(parts) + 1):+                occurrence = level.itemByName("+".join(parts[:size]))+                if occurrence is not None:+                    break+            else:+                return None+            parts = parts[size:]+            if not parts:+                return occurrence+            level = occurrence.childOccurrences+    except (AttributeError, RuntimeError):+        return None  # callers fall back to reading every occurrence+    return None+++def in_root_context(design, occurrence):+    """Whether an occurrence is one of the root context's (rootComponent.allOccurrences), by its path+    first; only a miss reads them all."""+    found = occurrence_at(design, occurrence.fullPathName)+    if found is not None and same(found, occurrence):+        return True+    return any(same(occurrence, item) for item in items(design.rootComponent.allOccurrences))++ def occurrence_context(design, entity, path):     entity = concrete(entity)     require(entity.objectType.endswith("::Occurrence"), "invalid_context", "Context must resolve to an occurrence.")     native = getattr(entity, "nativeObject", None) or entity+    found = occurrence_at(design, path)+    if found is not None and found.fullPathName == path and same(getattr(found, "nativeObject", None) or found, native):+        return found     candidates = [o for o in items(design.rootComponent.allOccurrences)                   if o.fullPathName == path and same(getattr(o, "nativeObject", None) or o, native)]     if candidates:@@ -516,19 +588,24 @@ def parameter_data(design, parameter):     return row  -def recovery_snapshot(design):-    """Semantic recovery witness; native revision IDs legitimately change on undo."""+def recovery_snapshot(design, health_state=None):+    """Semantic recovery witness; native revision IDs legitimately change on undo. health_state is the+    health just read on this same unchanged model, when the caller has it (each walk reads every timeline+    item, about 23 ms apiece on a big assembly)."""     from .readback import construction_state, feature_data, sketch_data, topology_witness     from .assembly_state import component_state, occurrence_state     from .configurations import snapshot as configuration_snapshot     configuration = configuration_snapshot(design, strict=True)+    all_occurrences = items(design.rootComponent.allOccurrences)     return {"configuration": configuration,             "parameters": [[p.name, p.expression, p.unit] for p in items(design.allParameters)],             "components": [[identity(c), c.name, [topology_witness(b) for b in items(c.bRepBodies)],                             [sketch_data(s) for s in items(c.sketches)],-                            [feature_data(f, order=i) for i, f in enumerate(items(c.features))], component_state(c), construction_state(c)] for c in items(design.allComponents)],-            "occurrences": [[identity(o), list(o.transform2.asArray()), occurrence_state(o)] for o in items(design.rootComponent.allOccurrences)],-            "timeline": design.timeline.count, "timelineMarker": design.timeline.markerPosition, "health": health(design)}+                            [feature_data(f, order=i) for i, f in enumerate(items(c.features))], component_state(c), construction_state(c)]+                           for c in components_by_occurrence(design, all_occurrences, design.allComponents.count)],+            "occurrences": [[identity(o), list(o.transform2.asArray()), occurrence_state(o)] for o in all_occurrences],+            "timeline": design.timeline.count, "timelineMarker": design.timeline.markerPosition,+            "health": health_state if health_state is not None else health(design)}   def number(value, name, positive=False):

@@ -4,7 +4,7 @@ import re import adsk.core import adsk.fusion -from .core import (MechanicalError, active, body_data, current_target, health,+from .core import (MechanicalError, components_by_occurrence, active, body_data, current_target, health,                    identity, items, number, parameter_data, recovery_snapshot,                    ref, require, require_healthy, resolve, target, UNITS) from .exporting import export_design@@ -24,11 +24,19 @@ def inspect(app, args):     # part numbers and descriptions are cloud reads (~0.6 s each per component on a cloud document):     # read them for a small design, or when asked; a big one gets them from bom or inspect_entity     native = args.get("componentProperties", design.allComponents.count <= 200)-    for comp in items(design.allComponents):+    # body metrics (physical properties and a bounding box, about 10 ms a body) and timeline health+    # (about 22 ms a timeline item) are read for a small design or when asked: on a 4,579-occurrence+    # assembly they were 40 s of a 103 s inspect; inspect_entity and the health op read them for one thing+    metrics = args.get("bodyMetrics", design.allComponents.count <= 200)+    with_health = args.get("health", design.timeline.count <= 200)+    # allComponents.item() costs about 5 ms a component, and inspect walked the collection five times+    all_occurrences = items(design.rootComponent.allOccurrences)+    all_components = components_by_occurrence(design, all_occurrences, design.allComponents.count)+    for comp in all_components:         components.append({"id": identity(comp), **metadata(comp, native), "ref": ref(comp, value),                            "bodyCount": comp.bRepBodies.count, "sketchCount": comp.sketches.count})         for body in items(comp.bRepBodies):-            row = body_data(body)+            row = body_data(body) if metrics else {"name": body.name, "space": "component"}             row.update({"ref": ref(body, value), "componentId": identity(comp)})             bodies.append(row)         for sketch in items(comp.sketches):@@ -44,7 +52,7 @@ def inspect(app, args):         from .readback import plane_data         for plane in [comp.xYConstructionPlane, comp.xZConstructionPlane, comp.yZConstructionPlane] + items(comp.constructionPlanes):             planes.append({**plane_data(plane, value), "componentId": identity(comp)})-    for occ in items(design.rootComponent.allOccurrences):+    for occ in all_occurrences:         matrix = list(occ.transform2.asArray())         # Matrix4x4.asArray is row-major; translation lies at 3,7,11.         for i in (3, 7, 11):@@ -56,17 +64,20 @@ def inspect(app, args):     from .joint_state import origin_data, link_data     from .configurations import snapshot as configuration_snapshot     configuration = configuration_snapshot(design)-    return {"target": value, "componentPropertiesRead": bool(native), "document": {"name": doc.name, "creationId": doc.creationId,+    return {"target": value, "componentPropertiesRead": bool(native), "bodyMetricsRead": bool(metrics),+            "document": {"name": doc.name, "creationId": doc.creationId,             "saved": doc.isSaved}, "designIntent": str(getattr(design, "designIntent", "unavailable")),             "units": UNITS, "components": components, "bodies": bodies, "occurrences": occurrences,             "sketches": sketches, "features": features, "constructionAxes": axes, "constructionPlanes": planes,-            "joints": [joint_data(j, value) for comp in items(design.allComponents)+            "joints": [joint_data(j, value) for comp in all_components                        for name in ("joints", "asBuiltJoints") for j in items(getattr(comp, name))],-            "rigidGroups": [rigid_group_data(g, value) for comp in items(design.allComponents) for g in items(comp.rigidGroups)],-            "jointOrigins": [origin_data(o, value) for comp in items(design.allComponents) for o in items(comp.jointOrigins)],-            "motionLinks": [link_data(o, value) for comp in items(design.allComponents) for o in items(comp.motionLinks)] if hasattr(design.rootComponent, "motionLinks") else [],+            "rigidGroups": [rigid_group_data(g, value) for comp in all_components for g in items(comp.rigidGroups)],+            "jointOrigins": [origin_data(o, value) for comp in all_components for o in items(comp.jointOrigins)],+            "motionLinks": [link_data(o, value) for comp in all_components for o in items(comp.motionLinks)] if hasattr(design.rootComponent, "motionLinks") else [],             "parameters": [parameter_data(design, p) for p in items(design.allParameters)],-            "health": health(design), "complete": configuration["complete"], "configuration": configuration}+            "health": health(design) if with_health else {"skipped": True, "timelineCount": design.timeline.count,+                                                        "hint": "Big timeline: run op health, or pass health:true."},+            "complete": configuration["complete"], "configuration": configuration}   def inspect_entity(app, args):@@ -179,13 +190,13 @@ def create_box(design, dimensions):  def parameter_set(app, args):     _, design, before_target = target(app, args, True)-    require_healthy(design, False)+    healthy = require_healthy(design, False)     name, expression = args.get("name"), args.get("expression")     require(isinstance(name, str) and isinstance(expression, str) and expression.strip(), "invalid_input", "name and expression are required strings.")     parameter = design.allParameters.itemByName(name)     require(parameter is not None, "parameter_missing", "No parameter named " + name)     require(design.unitsManager.isValidExpression(expression, parameter.unit), "invalid_expression", "Expression is invalid for the parameter's unit.")-    before, old = recovery_snapshot(design), parameter.expression+    before, old = recovery_snapshot(design, healthy), parameter.expression     try:         parameter.expression = expression         state = require_healthy(design)@@ -205,13 +216,13 @@ def parameter_set(app, args):  def parameter_create(app, args):     _, design, _ = target(app, args, True)-    require_healthy(design, False)+    healthy = require_healthy(design, False)     name, expression, unit = args.get("name"), args.get("expression"), args.get("unit")     require(isinstance(name, str) and re.fullmatch(r"[A-Za-z][A-Za-z0-9_]{0,63}", name), "invalid_input", "Use a short parameter identifier.")     require(unit in ("mm", "deg", "kg", ""), "invalid_input", "unit must be mm, deg, kg or empty for unitless.")     require(isinstance(expression, str) and design.unitsManager.isValidExpression(expression, unit), "invalid_expression", "Expression must match the declared unit.")     require(design.allParameters.itemByName(name) is None, "parameter_exists", "Parameter already exists.")-    before = recovery_snapshot(design)+    before = recovery_snapshot(design, healthy)     parameter = None     try:         parameter = design.userParameters.add(name, adsk.core.ValueInput.createByString(expression), unit, args.get("comment", ""))

@@ -0,0 +1,307 @@+"""Hot paths of the revision digest, offline: what each native read costs is the point, so these count reads."""+import importlib+import time+import unittest.mock+from types import SimpleNamespace as NS+import unittest++from test_mechanical import adsk++core = importlib.import_module('mechanical_test_commands.mechanical.core')+readback = importlib.import_module('mechanical_test_commands.mechanical.readback')+regular_joints = importlib.import_module('mechanical_test_commands.mechanical.regular_joints')+bom = importlib.import_module('mechanical_test_commands.mechanical.bill_of_materials')+++class Counted:+    """An occurrence whose entityToken read is counted (it costs about 2 ms in Fusion)."""+    objectType = "adsk::fusion::Occurrence"++    def __init__(self, path):+        self.fullPathName, self.reads, self.isValid = path, 0, True++    @property+    def entityToken(self):+        self.reads += 1+        return "token-" + self.fullPathName+++class Identity(unittest.TestCase):+    def test_a_known_entity_is_found_without_reading_its_token(self):+        occ = Counted("Frame:1+Rail:3")+        first = core.identity(occ)+        reads = occ.reads+        for _ in range(5):+            self.assertEqual(core.identity(occ), first)+        self.assertEqual(occ.reads, reads)++    def test_a_renamed_entity_is_still_found_by_its_token(self):+        occ = Counted("Frame:1+Rail:4")+        key = core.identity(occ)+        occ.fullPathName = "Frame:1+Rail renamed:4"   # new bucket; same token string? no: token follows the object+        occ.__class__ = type("Renamed", (Counted,), {"entityToken": property(lambda self: "token-Frame:1+Rail:4")})+        self.assertEqual(core.identity(occ), key)++    def test_a_reference_reuses_the_token_read_at_registration(self):+        occ = Counted("Frame:1+Rail:5")+        core.identity(occ)+        reads = occ.reads+        for _ in range(3):+            self.assertEqual(core.entity_token(occ), "token-Frame:1+Rail:5")+        self.assertEqual(occ.reads, reads)++    def test_an_unregistered_occurrence_reads_its_token_once(self):+        occ = Counted("Frame:1+Rail:6")+        self.assertEqual(core.entity_token(occ), "token-Frame:1+Rail:6")+        self.assertEqual(core.entity_token(occ), "token-Frame:1+Rail:6")+        self.assertEqual(occ.reads, 1)++    def test_other_entities_read_their_own_token(self):+        body = NS(objectType="adsk::fusion::BRepBody", entityToken="body-token")+        self.assertEqual(core.entity_token(body), "body-token")+++class Body:+    """A body whose witness reads are counted (physical properties cost milliseconds in Fusion)."""+    objectType = "adsk::fusion::BRepBody"++    def __init__(self, name, component):+        self.name, self.parentComponent, self.isValid, self.revisionId = name, component, True, "b1"+        self.material, self.reads = NS(id="steel"), 0+        empty = NS(count=0, item=lambda i: None)+        self.faces, self.edges, self.isSolid = empty, empty, True+        self.boundingBox = NS(minPoint=NS(x=0, y=0, z=0), maxPoint=NS(x=1, y=1, z=1))++    @property+    def physicalProperties(self):+        self.reads += 1+        return NS(volume=1, area=6, mass=7.8, centerOfMass=NS(x=0.5, y=0.5, z=0.5))+++class Witnesses(unittest.TestCase):+    def test_an_unchanged_body_is_not_read_again(self):+        body = Body("Plate", NS(name="Bracket"))+        first = readback.topology_witness(body)+        self.assertEqual(readback.topology_witness(body), first)+        self.assertEqual(body.reads, 1)++    def test_a_changed_body_material_or_name_is_read_again(self):+        body = Body("Plate", NS(name="Bracket 2"))+        readback.topology_witness(body)+        body.revisionId = "b2"+        readback.topology_witness(body)+        body.material = NS(id="aluminum")+        readback.topology_witness(body)+        self.assertEqual(body.reads, 3)++    def test_a_cached_witness_is_a_copy(self):+        body = Body("Plate", NS(name="Bracket 3"))+        readback.topology_witness(body)["name"] = "changed by a caller"+        self.assertEqual(readback.topology_witness(body)["name"], "Plate")+++class Feature:+    objectType = "adsk::fusion::BaseFeature"++    def __init__(self, name, component):+        self.name, self.parentComponent, self.isValid, self.index_reads = name, component, True, 0+        feature = self++        class Timeline:+            isSuppressed, isValid = False, True++            @property+            def index(self):+                feature.index_reads += 1+                raise RuntimeError("2 : InternalValidationError : res >= 0")+        self.timelineObject = Timeline()+++class UnindexedFeatures(unittest.TestCase):+    def test_a_refused_timeline_index_is_asked_once_per_component_revision(self):+        comp = NS(name="Vendor model", revisionId="r1")+        feature = Feature("Base Feature1", comp)+        for _ in range(4):+            data = readback.feature_data(feature, order=0)+            self.assertEqual((data["index"], data["indexUnavailable"]), (None, "native_timeline_position_unavailable"))+        self.assertEqual(feature.index_reads, 1)+        comp.revisionId = "r2"   # the component changed: ask again+        readback.feature_data(feature, order=0)+        self.assertEqual(feature.index_reads, 2)++++class Components(unittest.TestCase):+    def comp(self, name):+        return NS(objectType="adsk::fusion::Component", name=name, isValid=True)++    def design(self, comps, occurrences):+        reads = []+        collection = NS(count=len(comps), item=lambda i: reads.append(i) or comps[i])+        return NS(rootComponent=comps[0], allComponents=collection), reads++    def test_components_come_from_occurrences_without_reading_the_slow_collection(self):+        root, a, b = self.comp("root"), self.comp("A"), self.comp("B")+        occurrences = [NS(component=a), NS(component=b), NS(component=a)]+        design, reads = self.design([root, a, b], occurrences)+        found = core.components_by_occurrence(design, occurrences, 3)+        self.assertEqual([c.name for c in found], ["root", "A", "B"])+        self.assertEqual(reads, [])++    def test_a_count_mismatch_falls_back_to_the_full_list(self):+        root, a, orphan = self.comp("root"), self.comp("A"), self.comp("Orphan")+        design, reads = self.design([root, a, orphan], [NS(component=a)])+        found = core.components_by_occurrence(design, [NS(component=a)], 3)+        self.assertEqual([c.name for c in found], ["root", "A", "Orphan"])+        self.assertEqual(reads, [0, 1, 2])+++class Level:+    """Occurrences under one parent, found by name the way Fusion's itemByName does."""+    def __init__(self, occurrences=()):+        self.values = list(occurrences)++    @property+    def count(self):+        return len(self.values)++    def item(self, index):+        return self.values[index]++    def itemByName(self, name):+        return next((o for o in self.values if o.name == name), None)+++class Occ:+    objectType = "adsk::fusion::Occurrence"+    compared = 0++    def __init__(self, name, children=(), parent=None):+        self.name, self.isValid, self.parent = name, True, parent+        self.childOccurrences = Level(children)+        for child in children:+            child.parent = self++    @property+    def fullPathName(self):+        return (self.parent.fullPathName + "+" if self.parent else "") + self.name++    def __eq__(self, other):+        Occ.compared += 1+        return self is other++    __hash__ = object.__hash__+++class Paths(unittest.TestCase):+    def tree(self):+        c = Occ("Nut:1")+        b = Occ("Bolt:1", [c])+        a = Occ("Frame:1", [b])+        plus = Occ("M3+washer:1", [Occ("Ring:2")])+        every = [a, b, c, plus, plus.childOccurrences.item(0)]++        class Everything(Level):+            reads = 0++            def item(self, index):+                Everything.reads += 1+                return self.values[index]+        all_occurrences = Everything(every)+        design = NS(rootComponent=NS(occurrences=Level([a, plus]), allOccurrences=all_occurrences))+        return design, a, b, c, plus, all_occurrences++    def test_an_occurrence_is_found_by_walking_its_path(self):+        design, a, b, c, plus, _ = self.tree()+        self.assertIs(core.occurrence_at(design, "Frame:1+Bolt:1+Nut:1"), c)+        self.assertIs(core.occurrence_at(design, "Frame:1"), a)+        self.assertIsNone(core.occurrence_at(design, "Frame:1+Washer:1"))+        self.assertIsNone(core.occurrence_at(design, ""))++    def test_a_name_holding_a_plus_is_still_found(self):+        design, a, b, c, plus, _ = self.tree()+        self.assertIs(core.occurrence_at(design, "M3+washer:1+Ring:2"), plus.childOccurrences.item(0))++    def test_resolving_a_context_does_not_read_every_occurrence(self):+        design, a, b, c, plus, every = self.tree()+        every.__class__.reads = 0+        self.assertIs(core.occurrence_context(design, c, "Frame:1+Bolt:1+Nut:1"), c)+        self.assertTrue(core.in_root_context(design, b))+        self.assertEqual(every.__class__.reads, 0)++    def test_an_occurrence_outside_the_root_context_still_reads_them_all(self):+        design, a, b, c, plus, every = self.tree()+        local = Occ("Nut:1")   # a definition-local occurrence: its path is relative to its parent+        every.__class__.reads = 0+        self.assertFalse(core.in_root_context(design, local))+        self.assertEqual(every.__class__.reads, every.count)+++class JointPoses(unittest.TestCase):+    def test_pose_check_compares_each_occurrence_once(self):+        occurrences = [Occ("Part:%d" % i) for i in range(200)]+        for o in occurrences:+            o.transform2 = NS(asArray=lambda: [1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1, 0, 0, 0, 0, 1])+            o.isGrounded, o.isGroundToParent = False, False+        design = NS(rootComponent=NS(allOccurrences=Level(occurrences)))+        before = regular_joints.poses(design)+        Occ.compared = 0+        result = regular_joints.pose_result(design, before)+        self.assertEqual(result["changedOccurrences"], [])+        self.assertLessEqual(Occ.compared, len(occurrences))+++class Part:+    """A component whose part number is a counted cloud read (about 0.5 s each in Fusion)."""+    objectType = "adsk::fusion::Component"++    def __init__(self, name):+        self.name, self.isValid, self.revisionId, self.reads = name, True, "r1", 0+        self.bRepBodies, self.occurrences, self.attributes, self.material = Level(), Level(), Level(), None++    @property+    def partNumber(self):+        self.reads += 1+        return "PN-" + self.name++    description = "a part"+++class BomBudget(unittest.TestCase):+    def design(self, tag):+        parts = [Part("%s %d" % (tag, i)) for i in range(3)]+        occurrences = [NS(objectType="adsk::fusion::Occurrence", component=p, fullPathName=p.name + ":1", isLightBulbOn=True,+                          timelineObject=NS(isSuppressed=False), childOccurrences=Level(), bRepBodies=Level(), isConfiguration=False)+                       for p in parts]+        root = Part(tag + " root")+        root.occurrences, root.allOccurrences = Level(occurrences), Level(occurrences)+        design = NS(isConfiguredDesign=False, isConfiguration=False, configurationRowId="", rootComponent=root)+        return design, [root] + parts++    def generate(self, design, started):+        with unittest.mock.patch.object(bom, "ref", return_value={}):+            return bom.generate(design, {"documentId": "D", "revision": "R"}, {}, started, 40)++    def test_past_the_budget_part_numbers_are_left_for_the_next_run(self):+        design, comps = self.design("Late")+        result = self.generate(design, time.perf_counter() - 100)+        self.assertEqual(result["nativePending"], 4)+        self.assertEqual(sum(c.reads for c in comps), 0)+        self.assertIn("run bom again", result["nativeHint"])++    def test_inside_the_budget_every_part_number_is_read(self):+        design, comps = self.design("Early")+        result = self.generate(design, time.perf_counter())+        self.assertNotIn("nativePending", result)+        self.assertEqual([r["metadata"]["partNumber"] for r in result["instances"]], ["PN-" + c.name for c in comps])++    def test_what_an_earlier_run_read_is_used_past_the_budget(self):+        design, comps = self.design("Again")+        self.generate(design, time.perf_counter())+        result = self.generate(design, time.perf_counter() - 100)+        self.assertNotIn("nativePending", result)+        self.assertEqual(sum(c.reads for c in comps), 4)+++if __name__ == "__main__":+    unittest.main()