# -*- coding: utf-8 -*- """Measure every in-scope assembly in the open model. Runs INSIDE Revit. Not run directly -- `tools/sweep.py` feeds it to the bridge over and over: python tools/rvt.py exec tools/sweep_in_revit.py Read only. No transaction, no view change, no selection, nothing that would show up in the user's undo stack. WHICH ASSEMBLIES, AND WHO SAYS SO --------------------------------- The DECLARED SCOPE (`vibe.scope`), read from the job knowledge layer, exactly as Job Recon, Takeoff and Ticket QC read it. Until 2026-08-31 this file filtered on a two-letter mark prefix written into the code, which was wrong twice over: the declaration in the job file was not load-bearing here (delete it and this still filtered; change it and this did not follow), and one job's naming was baked into a repository that must not carry one. `run()` already loads the knowledge dict for the measurement itself, so the scope costs nothing to read -- there was never a reason for the literal. WHAT IT SAYS ABOUT WHAT IT LEFT ALONE -------------------------------------- Out of scope is a third state, not an absence, so this writes the scope line `scope.json` next to the other outputs AND returns it from the enumerating call, which is the call `tools/sweep.py` prints. A sweep that silently measured a subset would look exactly like a sweep that measured everything. todo.json every in-scope assembly, its element id and its design scope.json the declared scope, the line, and what was left alone jobs/.job.json panellift input, default stages, nothing picked meta/.json what happened: status, axis, weight gate, audit One call does CHUNK pieces and returns how many are left. A piece with a meta file is done, so this is resumable and safe to re-run: delete the sweep folder to start over -- including after the declaration changes, because `todo.json` is enumerated once and does not re-read the scope. """ import io import json import os import traceback from Autodesk.Revit.DB import ElementId from vibe import designspec, knowledge, liftexport, ticketqc from vibe import scope as scopelib from vibe.compat import text_type CHUNK = 10 def _dump(path, obj): text = json.dumps(obj, indent=1, sort_keys=True, default=repr) if not isinstance(text, text_type): text = text.decode('utf-8') with io.open(path, 'w', encoding='utf-8') as fh: fh.write(text) def _load(path): with io.open(path, 'r', encoding='utf-8') as fh: return json.loads(fh.read()) def _folders(doc): root = designspec.sub_dir(doc, 'sweep') return root, designspec.sub_dir(doc, os.path.join('sweep', 'jobs')), \ designspec.sub_dir(doc, os.path.join('sweep', 'meta')) def enumerate_pieces(doc, know, todo_path, scope_path=None): """Every in-scope assembly with the design it belongs to, written once. Returns (todo, scope_info). What the DECLARED scope left alone is counted and named in `scope_info` and written to `scope.json` -- it is never simply absent, which is the one thing `vibe.scope` exists to stop. """ the_scope = scopelib.of(know) todo, left_alone, unreadable, total = [], [], [], 0 for design, group in ticketqc.assemblies_by_design(doc, know).items(): for asm in group: name = str(asm.Name) total += 1 if the_scope.is_unreadable(name): # No readable prefix: 'I could not tell' never removes work. unreadable.append(name) if not the_scope.includes(name): left_alone.append(name) continue todo.append({'id': ticketqc.eid(asm), 'mark': name, 'design': design}) todo.sort(key=lambda r: (r['design'], r['mark'])) _dump(todo_path, todo) scope_info = { 'declared': the_scope.is_declared, 'label': the_scope.label(), 'declaration': the_scope.declaration_lines(), 'line': the_scope.line(left_alone, total=total, unit='assembly', unreadable_names=unreadable), # The headline is written HERE, at enumeration, and read back # verbatim on every later call. See `_scope_headline`. 'headline': the_scope.short_line(left_alone, total=total, unit='assembly'), 'near_miss': the_scope.near_miss_line(left_alone, unit='assembly'), 'left_alone': the_scope.tally(left_alone), 'left_alone_count': len(left_alone), 'measured_count': len(todo), 'assemblies_seen': total, } if scope_path: _dump(scope_path, scope_info) return todo, scope_info def measure(doc, know, sheets, item, jobs_dir): """One assembly, as far as it gets. Never raises.""" meta = {'mark': item['mark'], 'design': item['design'], 'status': 'unknown'} try: asm = doc.GetElement(ElementId(item['id'])) # Unattended: take the time and locate the inserts properly. job, problems = liftexport.export(doc, asm, know, sheets, item['design'], batch=True) meta['problems'] = problems if job is None: meta['status'] = 'refused' return meta src = job.get('source') or {} panel = job.get('panel') or {} meta['axis'] = src.get('erected_vertical_axis') meta['length_ft'] = panel.get('length_ft') meta['width_ft'] = panel.get('width_ft') meta['thickness_in'] = panel.get('thickness_in') meta['lifters'] = len(panel.get('lifters') or []) meta['strands'] = len( ((panel.get('prestress') or {}).get('strands')) or []) meta['fc_psi'] = panel.get('fc_psi') meta['fci_psi'] = panel.get('fci_psi') ok, message, _numbers = liftexport.weight_check(job, know) meta['weight_ok'] = bool(ok) if not ok: meta['weight_msg'] = message stages, _rows, stage_problems = liftexport.default_stages(job, know) if not stages: meta['status'] = 'no stages' return meta if stage_problems: meta['status'] = 'dunnage misconfigured' meta['stage_problems'] = stage_problems return meta # A stage carried on dunnage has no 'lifters' key at all -- it is # dropped rather than emptied, because panellift refuses a stage that # gives both a support line and an insert. meta['stage_lifters'] = dict( (s['name'], len(s.get('lifters') or [])) for s in stages) meta['stage_span'] = dict( (s['name'], s.get('span')) for s in stages) meta['stage_supports'] = dict( (s['name'], s.get('supports')) for s in stages if s.get('supports')) try: meta['audit'] = repr(liftexport.stage_audit(stages)) except Exception as exc: meta['audit'] = 'audit failed: {0}'.format(exc) # The stages here are the DEFAULTS -- what the grouping rules imply, # with nobody having clicked anything. That is the point of a sweep # and it is also its limit: see docs/handoff.md. missing = [s['name'] for s in stages if not s.get('fc_psi')] if missing: meta['status'] = 'no strength' meta['missing_fc'] = missing return meta job['stages'] = stages _dump(os.path.join(jobs_dir, item['mark'] + '.job.json'), liftexport.strip_private(job)) meta['status'] = 'exported' except Exception: meta['status'] = 'crash' meta['error'] = traceback.format_exc() return meta def _scope_headline(info): """The recorded headline, for the FRONT of every return value. Short and first on purpose. `tools/sweep.py` prints `str(out)[:200]` -- it truncates -- so the full sentence lives in `scope.json` and what a reader sees at the console is the count, which is what tells him there is a file to open. A scope line cut off the end of a progress line would be a scope line nobody read. A sweep folder enumerated before this existed has no `scope.json`, and that is reported rather than filled in: what such a run filtered on is genuinely unknown to this code. """ if not info: return ('SCOPE: NOT RECORDED -- this sweep folder was enumerated ' 'before the scope was read, so what it left out is unknown. ' 'Delete the sweep folder and re-run.') return (info.get('headline') or 'SCOPE: recorded without a headline') \ + ' (scope.json)' def run(doc): know = knowledge.load(doc) root, jobs_dir, meta_dir = _folders(doc) todo_path = os.path.join(root, 'todo.json') scope_path = os.path.join(root, 'scope.json') if not os.path.isfile(todo_path): todo, scope_info = enumerate_pieces(doc, know, todo_path, scope_path) return '{0} | enumerated {1} assemblies into {2}'.format( _scope_headline(scope_info), len(todo), root) # Re-read rather than re-derive: the enumeration is what the scope was # applied to, and that happened on an earlier call. Saying it again from # a knowledge file that may have been edited since would describe a # filter this run did not use. scope_info = _load(scope_path) if os.path.isfile(scope_path) else None todo = _load(todo_path) _summaries, sheets = ticketqc.collect_context(doc) did, remaining = [], 0 for item in todo: meta_path = os.path.join(meta_dir, item['mark'] + '.json') if os.path.isfile(meta_path): continue if len(did) >= CHUNK: remaining += 1 continue meta = measure(doc, know, sheets, item, jobs_dir) _dump(meta_path, meta) did.append('{0}:{1}'.format(item['mark'], meta['status'])) return '{0} | did {1} [{2}] remaining {3}'.format( _scope_headline(scope_info), len(did), ', '.join(did), remaining) result = run(doc) # noqa: F821 -- doc is injected by tools/rvt.py