182 lines
11 KiB
Python
182 lines
11 KiB
Python
"""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 / '도면추출'
|
|
|
|
|
|
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()
|