"""Publish explicitly authorized IFC-rhythm body ties, preserving other objects.""" import argparse import copy import hashlib import json import math import os from pathlib import Path import re import sys import tempfile import time project_root = Path(__file__).resolve().parents[1] backend_root = project_root/'src'/'backend' sys.path.insert(0, str(backend_root if backend_root.is_dir() else project_root)) from body_source_tie import ordinary_spine, build_reference_review from cad_rebar_recipe import DrawingSource def read_reference_rule(path, cycle_entity, tie_entities): raw = path.read_bytes() entities = {} for line in raw.decode().splitlines(): match = re.fullmatch(r'#(\d+)\s*=\s*(\w+)\((.*)\);', line.strip()) if match: entities[int(match[1])] = (match[2], match[3]) def children(number, target): kind, value = entities[number] if kind == target: return [number] if kind not in ('IFCREINFORCINGBAR', 'IFCPRODUCTDEFINITIONSHAPE', 'IFCSHAPEREPRESENTATION', 'IFCSWEPTDISKSOLID', 'IFCCOMPOSITECURVE', 'IFCCOMPOSITECURVESEGMENT'): return [] return [child for ref in re.findall(r'#(\d+)', value) for child in children(int(ref), target)] def first(number): line = children(number, 'IFCPOLYLINE')[0] point = int(re.findall(r'#(\d+)', entities[line][1])[0]) return list(map(float, entities[point][1].strip('()').split(','))) solids = children(cycle_entity, 'IFCSWEPTDISKSOLID') origin, second = first(solids[0]), first(solids[1]) delta = [second[i]-origin[i] for i in range(2)] length = math.hypot(*delta) axis = [v/length for v in delta] def station(number): p = first(number) return sum((p[i]-origin[i])*axis[i] for i in range(2)) _, pitch = ordinary_spine([station(s) for s in solids]) runs = [ordinary_spine([station(s) for s in children(n, 'IFCSWEPTDISKSOLID')]) for n in tie_entities] if len(runs) != 2 or abs(runs[0][1]-runs[1][1]) > 1e-5: raise ValueError('two matching IFC tie phases required') repeat = runs[0][1] phase = abs(runs[1][0][0]-runs[0][0][0]) % repeat ratio, offset = repeat/pitch, phase/pitch if abs(ratio-round(ratio)) > 1e-3 or abs(offset-round(offset)) > 1e-3: raise ValueError('IFC rhythm does not close on its own cycle grid') return {'authority': 'reference_ifc', 'source_sha256': hashlib.sha256(raw).hexdigest(), 'cycle_entity': cycle_entity, 'tie_entities': tie_entities, 'measured_cycle_pitch_m': pitch, 'measured_tie_pitch_m': repeat, 'measured_phase_m': phase, 'repeat_cycle_steps': round(ratio), 'alternate_cycle_steps': round(offset), 'scope': 'rhythm only; IFC absolute coordinates, lengths and wall fabrication are not copied'} def main(): p = argparse.ArgumentParser(description=__doc__) p.add_argument('job', type=Path) p.add_argument('cad', type=Path) p.add_argument('ifc', type=Path) p.add_argument('--extra-cad', type=Path, action='append', default=[]) p.add_argument('--reference-cycle', type=int, required=True) p.add_argument('--reference-ties', type=int, nargs=2, required=True) p.add_argument('--confirm-ifc-rhythm', action='store_true') p.add_argument('--apply', action='store_true') p.add_argument('--replace-body-ties', action='store_true', help='Replace only prior source-body-tie groups after preservation checks') p.add_argument('--update-review-pose', action='store_true', help='Allow audited corrections to this review-only assembly; preserve all other bars') p.add_argument('--output', type=Path) args = p.parse_args() if not args.confirm_ifc_rhythm: raise ValueError('explicit IFC-rhythm review confirmation required') raw = args.job.read_bytes() job = json.loads(raw) if job.get('status') not in ('completed', 'awaiting_feedback', 'failed'): raise ValueError('refusing to change an active job') rule = read_reference_rule(args.ifc, args.reference_cycle, args.reference_ties) source = DrawingSource(args.cad.read_bytes()) sources = {source.sha256: source} for path in args.extra_cad: extra = DrawingSource(path.read_bytes()) sources[extra.sha256] = extra failures = [] groups = build_reference_review(job['result'], sources, rule, authorized=True, failures=failures) if len({g['id'] for g in groups}) != len(groups): raise ValueError('ambiguous duplicate body tie fabrication/host identity') result = copy.deepcopy(job['result']) existing = {g.get('id') for g in result['rebar_groups'] if g.get('id')} if any(g['id'] in existing for g in groups) and not args.replace_body_ties: raise ValueError('review already exists; explicit comparison required before replacement') replacing = ({g['id'] for g in result['rebar_groups'] if g.get('source_body_tie_binding')} if args.replace_body_ties else set()) previous = [g for g in result['rebar_groups'] if g.get('id') in replacing] if any(not g.get('source_body_tie_binding') for g in previous): raise ValueError('replacement is not an existing source body tie') # Every existing ordinary path must survive the fresh-source regeneration. def path_key(path): return tuple(tuple(round(v, 6) for v in p) for p in path) for old in previous: new = next((g for g in groups if g['id'] == old['id']),None) if args.update_review_pose: continue if new is None: raise ValueError('prior body tie missing from new source generation') old_paths = {path_key(p) for p in old['instance_paths_m']} new_paths = {path_key(p) for p in new['instance_paths_m']} if not old_paths <= new_paths: raise ValueError('fresh source regeneration changes previous tie paths: '+old['id']) retained = [g for g in result['rebar_groups'] if g.get('id') not in replacing] result['rebar_groups'] = list(retained) result['rebar_groups'].extend(groups) result['drawingEvidence']['body_tie_reference_rule'] = {**rule, 'explicit_request': True} result['drawingEvidence']['body_tie_reference_review'] = { 'approved': False, 'group_ids': [g['id'] for g in groups], 'reference_authority': 'IFC rhythm authorized by user; not current drawing spacing', 'existing_concrete_and_rebar_preserved': True} result['drawingEvidence'].setdefault('preview_geometry', {})['rebar_groups'] = copy.deepcopy(result['rebar_groups']) if result['concrete_components'] != job['result']['concrete_components']: raise ValueError('concrete changed') if result['rebar_groups'][:len(retained)] != [g for g in job['result']['rebar_groups'] if g.get('id') not in replacing]: raise ValueError('existing bars changed') summary = {'groups': len(groups), 'bars': sum(g['count'] for g in groups), 'reference_rule': rule, 'failures': failures, 'rows': [{'block': g['rebar_block'], 'mark': g['mark'], 'placed': g['count'], 'schedule': g['fabrication_measurement']['schedule_count'], 'skew': sum(s['domain']=='skew' for s in g['source_body_tie_binding']['stations']), 'outside_samples': g['placement_geometry_check']['tube_outside_samples']} for g in groups], 'existing_concrete_and_rebar_preserved': True, 'applied': args.apply} print(json.dumps(summary, ensure_ascii=False), flush=True) if args.output: with args.output.open('x') as stream: json.dump({'summary': summary, 'result': result}, stream, ensure_ascii=False, allow_nan=False) if not args.apply: return if args.job.read_bytes() != raw: raise ValueError('concurrent job update') backup = args.job.with_name(args.job.stem+f'.before-body-tie-review-{time.time_ns()}.json') with backup.open('xb') as stream: stream.write(raw) job.update(result=result, partialResult=copy.deepcopy(result), geometryRevision=time.time_ns()) with tempfile.NamedTemporaryFile(mode='w', dir=args.job.parent, delete=False) as stream: json.dump(job, stream, ensure_ascii=False, allow_nan=False) temporary = stream.name if args.job.read_bytes() != raw: raise ValueError('concurrent job update before publish') # Preserve symlink-backed runtime storage. os.replace(temporary, args.job.resolve()) print(json.dumps({'saved': True, 'backup': str(backup)}), flush=True) if __name__ == '__main__': main()