ahnyongjooandClaude Sonnet 5.5
964a841de1
- Code\rebar_pipeline, review_pipeline, pier-program을 정션에서 실제 폴더로 전환 - Code\rhino_mcp 및 CLAUDE_rhino배경.md 추가, .mcp.json은 PYTHONPATH로 새 위치 사용 - pier-program 내부 .git 제거(이력은 backup\pier-program_20261028.bundle에 보관) - worktree 가이드 반영(CLAUDE.local.md, HANDOFF.md) - D:\Rhino-MCP 경로 참조를 새 위치로 치환 Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
73 lines
3.4 KiB
Python
73 lines
3.4 KiB
Python
"""Compare baseline/staging classification against unchanged saved models.
|
|
|
|
Read-only: no AI calls, job writes, geometry generation or approval changes.
|
|
Run each checkout in a fresh interpreter to prevent mixed module imports.
|
|
"""
|
|
import argparse
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
|
|
|
|
|
|
def worker(root, job):
|
|
sys.path.insert(0, str(root / 'src/backend'))
|
|
from fresh_rebar_8097 import _compact_nonwing_rebar_taxonomy
|
|
raw = job.read_bytes()
|
|
plan = json.loads(raw)['result']
|
|
before = copy.deepcopy(plan)
|
|
groups = plan.get('rebar_groups') or []
|
|
_compact_nonwing_rebar_taxonomy(plan, groups)
|
|
fields = ('id', 'host_component_ids', 'instance_paths_m', 'diameter_mm')
|
|
physical = lambda rows: [{k: g.get(k) for k in fields} for g in rows]
|
|
assert physical(groups) == physical(before.get('rebar_groups') or []), 'rebar geometry/hosts changed'
|
|
assert plan.get('concrete_components') == before.get('concrete_components'), 'concrete changed'
|
|
assert job.read_bytes() == raw, 'saved job changed'
|
|
return dict(pass_=True, concrete=len(plan.get('concrete_components') or []),
|
|
groups=len(groups), paths=sum(len(g.get('instance_paths_m') or []) for g in groups),
|
|
physical_hash=digest(physical(groups)), concrete_hash=digest(plan.get('concrete_components')),
|
|
saved_job_sha256=hashlib.sha256(raw).hexdigest())
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--baseline', type=Path)
|
|
parser.add_argument('--staging', type=Path)
|
|
parser.add_argument('--manifest', type=Path)
|
|
parser.add_argument('--jobs', type=Path)
|
|
parser.add_argument('--output', type=Path)
|
|
parser.add_argument('--worker-root', type=Path)
|
|
parser.add_argument('--worker-job', type=Path)
|
|
args = parser.parse_args()
|
|
if args.worker_root:
|
|
print(json.dumps(worker(args.worker_root, args.worker_job))); return
|
|
if args.output.exists():
|
|
parser.error('refusing to overwrite a comparison report')
|
|
rows = []
|
|
for item in json.loads(args.manifest.read_text())['models']:
|
|
row = dict(item)
|
|
for label, root in [('baseline', args.baseline), ('staging', args.staging)]:
|
|
result = subprocess.run([sys.executable, __file__, '--worker-root', str(root),
|
|
'--worker-job', str(args.jobs / (item['job_id'] + '.json'))],
|
|
capture_output=True, text=True)
|
|
row[label] = json.loads(result.stdout) if result.returncode == 0 else {
|
|
'pass_': False, 'error': result.stderr[-3000:]}
|
|
row['compatible'] = row['baseline'].get('pass_') and row['staging'] == row['baseline']
|
|
rows.append(row)
|
|
print(json.dumps({'name': item['name'], 'compatible': row['compatible'],
|
|
'baseline': row['baseline'], 'staging': row['staging']}, ensure_ascii=False), flush=True)
|
|
with args.output.open('x') as stream:
|
|
json.dump({'scope': 'saved-model classification compatibility only; not regeneration or new engineering approval',
|
|
'all_compatible': all(r['compatible'] for r in rows), 'results': rows}, stream, ensure_ascii=False, indent=2)
|
|
if not all(r['compatible'] for r in rows):
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == '__main__': main()
|