"""Generate the twelve packaged samples through the existing drawing pipeline. Read-only inputs outside this project; every output/cache/log stays below data/output. Missing or ambiguous measured surfaces remain failures, never synthetic surfaces. """ from __future__ import annotations import argparse from datetime import datetime import hashlib import json from pathlib import Path import re import subprocess import sys import traceback ROOT = Path(__file__).resolve().parent OUTPUT = ROOT / 'data/output/samples_20261102' PACKAGE = ROOT.parent / '도면추출_연동패키지_20261102' def save(path, value): path.parent.mkdir(parents=True, exist_ok=True) temp = path.with_suffix(path.suffix + '.tmp') temp.write_text(json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False), encoding='utf-8') temp.replace(path) def discover(source): jobs = [] references=ROOT/'data/input/samples_20261102/recovered/jobs.json' overrides=json.loads(references.read_text(encoding='utf-8')) if references.is_file() else {} for workbook in sorted((PACKAGE / 'samples').glob('샘플*.xlsx')): match = re.match(r'샘플(\d+)_(.+)_교각입력데이터', workbook.stem) if not match: continue number, bridge = int(match[1]), match[2] folders = [p for p in source.iterdir() if p.is_dir() and f'(sample{number})' in p.name] drawings = [p for folder in folders for p in folder.iterdir() if p.is_file() and p.suffix.lower() in ('.dwg', '.dxf') and '종평면도' in p.name] jobs.append(dict(sample=number, bridge=bridge, workbook=str(workbook), drawing=str(drawings[0]) if len(drawings) == 1 else None, candidates=[str(p) for p in drawings], output=str(OUTPUT / f'{number:02d}_{bridge}'))) if str(number) in overrides: jobs[-1]['profile_drawing']=jobs[-1]['drawing'] jobs[-1].update(overrides[str(number)]) return sorted(jobs, key=lambda item: item['sample']) def generate(job): from pierzainer_ground.drawing_job import build_from_files, build_drawing, DrawingJob from pierzainer_ground.project_coordinates import workbook_piers from pierzainer_ground.map_export import validate_embedded_image out = Path(job['output']); out.mkdir(parents=True, exist_ok=True) result = {**job, 'status': 'failed'} with (out / 'generation.log').open('w', encoding='utf-8') as log: def message(text): print(text, flush=True); log.write(str(text) + '\n'); log.flush() try: if job.get('recovered_source'): import shutil source=Path(job['drawing']);shared=ROOT/'data/cache/recovered_sources' cache=out/'cache';cache.mkdir(exist_ok=True) for suffix in ('.dxf','.dxf.source.json','.geometry-safe.dxf','.geometry-safe.repair.json'): saved=shared/(source.stem+suffix) if saved.is_file():shutil.copyfile(saved,cache/saved.name) # Inspection confirmed that the generic model-space candidate also # includes non-terrain CAD elevations in these two exact sources. # Reuse established layer readers; never shift coordinates or infer Z. digest=hashlib.sha256(Path(job['drawing']).read_bytes()).hexdigest() selections={ 'ee1d703b738e28c9ccb824707f2f2b6f87203499dd9efe34fea3b620c04264fa': dict(layers='CA002,CA0021,CAA001,CAA002',terrain_block='',z_min=0.,z_max=200., reason='기존 운곡교 LayerMap의 실제 표고점 INSERT/글자 및 등고선만 재사용'), 'e5cf5799e8265b28fca915be6ca1a8d97e0798f725a16f9d6d475daf9c031439': dict(layers='XR_현황-수정$0$Point',terrain_block='*Model_Space',z_min=-500.,z_max=9000., reason='원본의 비영 표고 측량 POINT만 사용. 일반 CAD 높이선 제외'), 'a5ade71241ad36364ba8f1b099ea6f5d221f28b30d8873bbbcec44a0a647942f': dict(layers='CAA001,CAA002',terrain_block='',z_min=0.,z_max=200.,terrain_crs='EPSG:5181', reason='감북교 원본의 2D POLYLINE 헤더 등고선 사용. 원좌표 N=447km / Excel N=547km 차이는 EPSG:5181→5186 규약 후보로 기록(원본 CRS 표기 미확인)')} if job.get('terrain_parameters'): defaults=json.loads((ROOT/'data/input/map_defaults.json').read_text(encoding='utf-8')) settings=DrawingJob(drawing=job['drawing'],piers=job['workbook'],output=str(out), crs=defaults['crs'],terrain_crs=defaults['crs'],allow_download=True,**job['terrain_parameters']) settings.auto_detection={'crs_status':defaults['status'],'crs_assumed':True, 'project_default_applied':True,'arbitrary_offset_applied':False, 'recovered_source':job['recovered_source'], 'original_profile_drawing':job.get('profile_drawing'), 'method':'recovered_original_CAD_survey_geometry'} message('누락된 원본 CAD 지형 복구: '+job['recovered_source']['source']) data=build_drawing(settings,message) elif digest in selections: selection=dict(selections[digest]);reason=selection.pop('reason') defaults=json.loads((ROOT/'data/input/map_defaults.json').read_text(encoding='utf-8')) terrain_crs=selection.pop('terrain_crs',defaults['crs']) settings=DrawingJob(drawing=job['drawing'],piers=job['workbook'],output=str(out), crs=defaults['crs'],terrain_crs=terrain_crs,allow_download=True,**selection) settings.auto_detection={'crs_status':defaults['status'],'crs_assumed':True, 'project_default_applied':True,'arbitrary_offset_applied':False, 'source_sha256':digest,'layer_selection_reason':reason, 'method':'inspected_original_terrain_layers'} if terrain_crs!=defaults['crs']: settings.auto_detection.update(crs_status='inferred_false_northing_convention_not_source_verified', project_default_applied=False, terrain_crs_assumed=True, coordinate_reference_reason=reason) message(reason) data=build_drawing(settings,message) else: data = build_from_files(job['drawing'], job['workbook'], message, out) controls = {p['id']:p['axis'] for p in workbook_piers(job['workbook'])} errors = [] for pier in data['piers']: a, b = controls[pier['id']] error = ((pier['X']-(a[0]+b[0])/2)**2+(pier['Y']-(a[1]+b[1])/2)**2)**.5 errors.append(error) # Existing schema serializes N/E to 0.001 m independently. if error > 0.000708: raise ValueError('IP1/IP3 midpoint verification failed') grid = data['terrain_grid'] if len(grid['z']) != grid['nx'] * grid['ny'] or not any(z is not None for z in grid['z']): raise ValueError('Invalid terrain grid') for layer in data.get('visualization', {}).get('layers', []): validate_embedded_image(layer) filename = 'ground_model_3d_with_maps.json' if data['map_status']['status'] == 'mapped' else 'ground_model.json' restored = json.loads((out / filename).read_text(encoding='utf-8')) assert restored['schema'] == 2 and len(restored['piers']) == len(controls) result.update(status='generated', json=str(out / filename), pier_count=len(data['piers']), valid_grid_points=sum(z is not None for z in grid['z']), covered_piers=sum(p['ground']['used'] is not None for p in data['piers']), midpoint_error_max_m=max(errors, default=0), map_status=data['map_status'], crs=data['crs'], crs_assumed=data['crs_assumed'], terrain_block=data['surface_provenance']['terrain_block'], sha256=hashlib.sha256((out / filename).read_bytes()).hexdigest()) except Exception as exc: result['error'] = str(exc) log.write(traceback.format_exc()); message('FAILED: ' + str(exc)) save(out / 'result.json', result) return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--source', type=Path, default=Path.home()/'Desktop/도면추출') parser.add_argument('--sample', type=int) parser.add_argument('--worker', type=int) args = parser.parse_args() OUTPUT.mkdir(parents=True, exist_ok=True) if args.worker is not None: jobs = json.loads((OUTPUT/'inputs.json').read_text(encoding='utf-8')) generate(next(job for job in jobs if job['sample'] == args.worker)) return jobs = discover(args.source) if len(jobs) != 12: raise ValueError(f'Expected twelve workbook samples, found {len(jobs)}') save(OUTPUT/'inputs.json', jobs) for job in jobs: if args.sample and job['sample'] != args.sample: continue out = Path(job['output']); out.mkdir(parents=True, exist_ok=True) if not job['drawing']: save(out/'result.json', {**job, 'status':'missing_drawing', 'error':'대응하는 종평면도 원본이 없거나 여러 개입니다. 임의 지형 생성 안 함.'}) continue print(f"START {job['sample']:02d} {job['bridge']}", flush=True) import os environment = dict(os.environ, PYTHONIOENCODING='utf-8') completed = subprocess.run([sys.executable, '-B', '-u', str(Path(__file__).resolve()), '--worker', str(job['sample'])], cwd=ROOT, env=environment) if completed.returncode: save(out/'result.json', {**job, 'status':'process_failed', 'exit_code':completed.returncode}) results = [json.loads((Path(job['output'])/'result.json').read_text(encoding='utf-8')) for job in jobs if (Path(job['output'])/'result.json').exists()] save(OUTPUT/'summary.json', {'updated':datetime.now().isoformat(), 'results':results}) for result in results: print(result['sample'], result['bridge'], result['status'], result.get('map_status', {}).get('status', ''), flush=True) if __name__ == '__main__': main()