395 lines
24 KiB
Python
395 lines
24 KiB
Python
"""AI-free drawing -> measured terrain -> portable ground/map JSON."""
|
|
from __future__ import annotations
|
|
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass, asdict, field
|
|
from datetime import date
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import re
|
|
from pathlib import Path
|
|
import threading
|
|
from collections import Counter
|
|
|
|
import numpy as np
|
|
from pyproj import CRS, Transformer
|
|
from scipy.spatial import QhullError
|
|
|
|
from . import map_export
|
|
from .boreholes import read_boreholes
|
|
from .config import GroundConfig
|
|
from .dtm import DTM
|
|
from .dxf_terrain import TerrainData, convert_dwg_to_dxf, read_terrain_dxf, read_source_document
|
|
from .piers import pier_from_dict, read_piers
|
|
from .pipeline import GroundModel, evaluate_pier
|
|
from .project_coordinates import workbook_piers
|
|
from .report import pier_dict
|
|
from .survey_terrain import read_survey_block
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
_MAP_LOCK = threading.RLock()
|
|
|
|
|
|
@dataclass
|
|
class DrawingJob:
|
|
drawing: str
|
|
output: str
|
|
crs: str
|
|
terrain_crs: str = ''
|
|
cad_axes: str = 'YX'
|
|
terrain_block: str = ''
|
|
piers: str = ''
|
|
boreholes: str = ''
|
|
layers: str = ''
|
|
margin: float = 100.0
|
|
step: float = 5.0
|
|
max_edge: float = 100.0
|
|
z_min: float = 0.0
|
|
z_max: float = 200.0
|
|
include_maps: bool = True
|
|
allow_download: bool = False
|
|
use_3d_lines: bool = False
|
|
plots: bool = False
|
|
automatic: bool = False
|
|
auto_detection: dict = field(default_factory=dict)
|
|
|
|
|
|
def inside_project(path):
|
|
result = Path(path).resolve()
|
|
if not result.is_relative_to(ROOT) or result == ROOT:
|
|
raise ValueError('결과는 pierzainer_ground 내부 하위 폴더에 저장하세요.')
|
|
if result.parts[len(ROOT.parts)] in ('.venv', 'tools', '.backup', 'pierzainer_ground'):
|
|
raise ValueError('실행 코드·런타임·백업 폴더에는 결과를 저장할 수 없습니다.')
|
|
return result
|
|
|
|
|
|
def metric_crs(value):
|
|
if not value.strip():
|
|
raise ValueError('EPSG 코드를 지정하세요. 좌표계는 자동으로 추정하지 않습니다.')
|
|
crs = CRS.from_user_input(value if ':' in value else 'EPSG:'+value)
|
|
if not crs.is_projected or any(abs(a.unit_conversion_factor-1)>1e-8 for a in crs.axis_info[:2]):
|
|
raise ValueError('m 단위 평면직각좌표계를 지정하세요 (위경도 직접 입력 불가).')
|
|
return crs.to_string()
|
|
|
|
|
|
def validate_job(job):
|
|
if not job.drawing or not Path(job.drawing).is_file() or Path(job.drawing).suffix.lower() not in ('.dxf','.dwg'):
|
|
raise ValueError('DXF 또는 DWG 종평면도를 선택하세요. PDF/이미지에서 측량 좌표를 복원하지 않습니다.')
|
|
for path in (job.piers, job.boreholes):
|
|
if path and not Path(path).is_file(): raise ValueError(f'입력 파일이 없습니다: {path}')
|
|
if not job.automatic or job.crs:
|
|
job.crs = metric_crs(job.crs)
|
|
job.terrain_crs = metric_crs(job.terrain_crs or job.crs)
|
|
if job.automatic and not job.piers:
|
|
raise ValueError('교각 Excel 파일을 선택하세요.')
|
|
if job.cad_axes not in ('YX','XY'): raise ValueError('CAD 축은 YX 또는 XY로 지정하세요.')
|
|
if not all(math.isfinite(v) for v in (job.margin,job.step,job.max_edge,job.z_min,job.z_max)):
|
|
raise ValueError('거리와 표고 설정은 유한한 수여야 합니다.')
|
|
if job.margin<0 or job.step<=0 or job.max_edge<=0 or job.z_min>=job.z_max:
|
|
raise ValueError('여백≥0, 격자/최대 TIN 변 길이>0, 최소 EL<최대 EL이어야 합니다.')
|
|
out=inside_project(job.output)
|
|
if any(out.is_relative_to(ROOT/p) for p in ('data/input','data/output/ungok','data/output/sanwoe')):
|
|
raise ValueError('새 결과는 기존 원본/운곡교/산외교와 다른 하위 폴더에 저장하세요.')
|
|
return out
|
|
|
|
|
|
def read_drawing(job, work):
|
|
cfg=GroundConfig(cad_axes=job.cad_axes,use_profile_points=False,use_3d_lines=job.use_3d_lines,z_min=job.z_min,z_max=job.z_max,tin_max_edge=job.max_edge)
|
|
source=Path(job.drawing)
|
|
if source.suffix.lower()=='.dwg':
|
|
work.mkdir(parents=True,exist_ok=True)
|
|
source=convert_dwg_to_dxf(source,work)
|
|
document=None
|
|
auto_block=False
|
|
if not job.terrain_block.strip():
|
|
# Some DWGs retain the surveyed surface as an identity XREF block,
|
|
# while modelspace contains only the bridge and drawing annotations.
|
|
# An identity INSERT needs no guessed placement or axis/CRS correction.
|
|
from ezdxf import recover
|
|
document,_=read_source_document(source,work)
|
|
centers=[]
|
|
if job.piers:
|
|
controls=([pier_from_dict(p) for p in workbook_piers(job.piers)]
|
|
if Path(job.piers).suffix.lower()=='.xlsx' else read_piers(job.piers))
|
|
centers=[(p.X,p.Y) for p in controls]
|
|
candidates=[]
|
|
for insert in document.modelspace().query('INSERT'):
|
|
name=insert.dxf.name
|
|
if not any(token in name.lower() for token in ('현황','지형','측량','survey','terrain','topo')):continue
|
|
if any(abs(v)>1e-8 for v in insert.dxf.insert) or abs(insert.dxf.rotation)>1e-8:continue
|
|
if any(abs(insert.dxf.get(k,1)-1)>1e-8 for k in ('xscale','yscale','zscale')):continue
|
|
if np.linalg.norm(np.array(insert.dxf.get('extrusion',(0,0,1)))-(0,0,1))>1e-8:continue
|
|
block=document.blocks.get(name)
|
|
if block is None:continue
|
|
measured=[tuple(e.dxf.location) for e in block if e.dxftype()=='POINT'
|
|
and job.z_min<float(e.dxf.location.z)<=job.z_max
|
|
and (not job.layers.strip() or e.dxf.layer in [s.strip() for s in job.layers.split(',')])]
|
|
if len(measured)<3:continue
|
|
array=np.array(measured)
|
|
east,north=(array[:,0],array[:,1]) if job.cad_axes=='YX' else (array[:,1],array[:,0])
|
|
east,north=Transformer.from_crs(job.terrain_crs,job.crs,always_xy=True).transform(east,north)
|
|
xy=np.column_stack([north,east])
|
|
if not np.isfinite(xy).all():continue
|
|
if centers and not np.all((np.array(centers)>=xy.min(0))&(np.array(centers)<=xy.max(0))):continue
|
|
candidates.append(name)
|
|
candidates=sorted(set(candidates))
|
|
if len(candidates)==1:
|
|
job.terrain_block=candidates[0]; auto_block=True
|
|
elif len(candidates)>1:
|
|
raise ValueError('동일 위치의 측량 블록 후보가 여러 개입니다. 측량 블록명에 하나를 지정하세요: '+', '.join(candidates))
|
|
if job.terrain_block.strip():
|
|
if document is None:document,_=read_source_document(source,work)
|
|
points,counts,contours=read_survey_block(source,{
|
|
'terrain_block':job.terrain_block.strip(),'terrain_crs':job.terrain_crs,
|
|
'crs':job.crs,'cad_axes':job.cad_axes,'contour_step_m':cfg.contour_step,
|
|
'dedup_m':cfg.dedup_xy,'z_min':job.z_min,'z_max':job.z_max,
|
|
'layers':[s.strip() for s in job.layers.split(',') if s.strip()]},doc=document)
|
|
terrain=TerrainData(points,np.full(len(points),'survey'),stats={
|
|
'file':source.name,'points_used':len(points),'removed':0,'contours':contours,
|
|
'source_counts':counts,'source_block':job.terrain_block,
|
|
'block_selection':'identity_survey_reference_matching_pier_extent' if auto_block else 'explicit',
|
|
'source_metadata_repair':getattr(document,'_ground_source_repair',None)})
|
|
else:
|
|
if job.layers.strip():
|
|
layers=tuple(s.strip() for s in job.layers.split(',') if s.strip())
|
|
cfg.layers.spot_block=cfg.layers.spot_text=cfg.layers.contour=layers
|
|
cfg.layers.elevation_layer_keywords=cfg.layers.contour_layer_keywords=layers
|
|
cfg.allowed_terrain_layers=layers
|
|
terrain=read_terrain_dxf(source,cfg,work,document=document)
|
|
if job.terrain_crs!=job.crs:
|
|
tr=Transformer.from_crs(job.terrain_crs,job.crs,always_xy=True)
|
|
if len(terrain.points):
|
|
east,north=tr.transform(terrain.points[:,1],terrain.points[:,0])
|
|
terrain.points=np.column_stack([north,east,terrain.points[:,2]])
|
|
for item in terrain.boreholes:
|
|
east,north=tr.transform(item['Y'],item['X']); item.update(X=north,Y=east)
|
|
terrain.contours=[]; terrain.bridge_lines=[]; terrain.pier_labels=[]
|
|
if len(terrain.points):
|
|
keep=(terrain.points[:,2]>job.z_min)&(terrain.points[:,2]<=job.z_max)
|
|
terrain.points=terrain.points[keep]; terrain.sources=terrain.sources[keep]
|
|
if len(terrain.points)<3 or not np.isfinite(terrain.points).all():
|
|
raise ValueError('실제 표고가 있는 지형점이 부족합니다. 평면 측량 블록·레이어·EL 범위를 지정하세요. 종단 도식 좌표로 지도 지형을 만들지 않습니다.')
|
|
return terrain,cfg
|
|
|
|
|
|
@contextmanager
|
|
def projection(crs):
|
|
"""Reuse the raster exporter with an explicit CRS, restoring legacy defaults."""
|
|
names=('_SURVEY_TO_WEB','_SURVEY_TO_WGS84','_WEB_TO_SURVEY')
|
|
with _MAP_LOCK:
|
|
previous=[getattr(map_export,n) for n in names]
|
|
try:
|
|
map_export._SURVEY_TO_WEB=Transformer.from_crs(crs,3857,always_xy=True)
|
|
map_export._SURVEY_TO_WGS84=Transformer.from_crs(crs,4326,always_xy=True)
|
|
map_export._WEB_TO_SURVEY=Transformer.from_crs(3857,crs,always_xy=True)
|
|
yield
|
|
finally:
|
|
for name,value in zip(names,previous): setattr(map_export,name,value)
|
|
|
|
|
|
def atomic_json(path,data):
|
|
content=json.dumps(data,ensure_ascii=False,indent=2,allow_nan=False)
|
|
temporary=path.with_suffix(path.suffix+'.tmp')
|
|
temporary.write_text(content,encoding='utf-8')
|
|
temporary.replace(path)
|
|
|
|
|
|
def inspect_drawing(path, log):
|
|
"""List drawing layer/block candidates in the GUI; do not infer a CRS."""
|
|
from ezdxf import recover
|
|
source=Path(path)
|
|
if not source.is_file() or source.suffix.lower() not in ('.dxf','.dwg'):
|
|
raise ValueError('DXF/DWG를 먼저 선택하세요.')
|
|
if source.suffix.lower()=='.dwg':
|
|
work=ROOT/'data/cache/drawing_conversion'; work.mkdir(parents=True,exist_ok=True)
|
|
source=convert_dwg_to_dxf(source,work)
|
|
doc,_=read_source_document(source,ROOT/'data/cache/drawing_conversion')
|
|
layers=Counter(e.dxf.layer for e in doc.modelspace())
|
|
log('모델 공간 레이어 (객체 수 상위 25개):\n'+ '\n'.join(f' {name}: {count:,}' for name,count in layers.most_common(25)))
|
|
candidates=[]
|
|
for block in doc.blocks:
|
|
if block.name.startswith('*'):continue
|
|
counts=Counter(e.dxftype() for e in block)
|
|
total=sum(counts[k] for k in ('POINT','LWPOLYLINE','POLYLINE'))
|
|
if total>=3:candidates.append((total,block.name,dict(counts)))
|
|
candidates.sort(reverse=True)
|
|
log('지형 검토 대상 블록 (객체가 많다고 측량 블록으로 확정하지 않음):\n'+'\n'.join(f' {name}: {counts}' for _,name,counts in candidates[:15]))
|
|
log('블록 내부가 실제 E/N 측량 좌표인지와 EPSG는 원도면에서 확인해야 합니다. 종단 도식/배치용 INSERT 좌표는 사용하지 마세요.')
|
|
|
|
|
|
def covering_cached_maps(crs,bounds,paths):
|
|
"""Use only real cached images covering this request in the same CRS."""
|
|
tr=Transformer.from_crs(crs,4326,always_xy=True)
|
|
n0,e0,n1,e1=bounds
|
|
corners=[tr.transform(e,n) for n in (n0,n1) for e in (e0,e1)]
|
|
for path in paths:
|
|
if not path.is_file():continue
|
|
try:
|
|
raw=json.loads(path.read_text(encoding='utf-8'))
|
|
vis=raw.get('visualization',raw)
|
|
if CRS.from_user_input(vis['source_crs'])!=CRS.from_user_input(crs):continue
|
|
layers={layer['id']:layer for layer in vis['layers']}
|
|
if layers['ground']['bounds']!=layers['satellite']['bounds']:continue
|
|
(south,west),(north,east)=layers['ground']['bounds']
|
|
if not all(west<=lon<=east and south<=lat<=north for lon,lat in corners):continue
|
|
for key in ('ground','satellite'):map_export.validate_embedded_image(layers[key])
|
|
vis['layers']=[layers['satellite'],layers['ground']]
|
|
vis.setdefault('view',{}).update(ground_visible=False)
|
|
return vis,path
|
|
except (KeyError,ValueError,OSError,TypeError):continue
|
|
return None,None
|
|
|
|
|
|
def build_drawing(job, log=lambda message:None):
|
|
out=validate_job(job)
|
|
out.mkdir(parents=True,exist_ok=True)
|
|
atomic_json(out/'generation_settings.json',asdict(job))
|
|
log('원본 도면에서 표고점·등고선 읽는 중...')
|
|
if job.automatic:
|
|
from .auto_inputs import read_auto_inputs
|
|
terrain,cfg=read_auto_inputs(job,out/'cache',log)
|
|
else:
|
|
terrain,cfg=read_drawing(job,out/'cache')
|
|
atomic_json(out/'generation_settings.json',asdict(job))
|
|
if terrain.stats.get('block_selection')=='identity_survey_reference_matching_pier_extent':
|
|
log(f"실측 표고점을 가진 원위치 측량 블록 자동 선택: {job.terrain_block} (좌표 이동·축 보정 없음)")
|
|
piers=[]
|
|
if job.piers:
|
|
if Path(job.piers).suffix.lower()=='.xlsx':
|
|
piers=[pier_from_dict(p) for p in workbook_piers(job.piers)]
|
|
if not piers: raise ValueError('교각 Excel IP 시트에서 IP1/IP3 좌표 쌍을 찾지 못했습니다.')
|
|
else: piers=read_piers(job.piers)
|
|
if any(not np.isfinite([p.X,p.Y,*p.axis_dir]).all() for p in piers): raise ValueError('교각 좌표/축이 유효하지 않습니다.')
|
|
if len({p.id for p in piers})!=len(piers): raise ValueError('교각 번호가 중복됩니다.')
|
|
# Text insertion points are not physical pier centers.
|
|
xy=np.array([(p.X,p.Y) for p in piers]) if piers else terrain.points[:,:2]
|
|
lo=xy.min(axis=0)-job.margin; hi=xy.max(axis=0)+job.margin
|
|
counts=np.ceil((hi-lo)/job.step).astype(int)+1
|
|
if np.any(counts<2) or int(counts[0])*int(counts[1])>500000:
|
|
raise ValueError('지형 격자가 너무 큽니다. 교각 좌표를 지정하거나 격자 간격을 늘리세요 (최대 50만 점).')
|
|
local=terrain.points[np.all((terrain.points[:,:2]>=lo-job.max_edge)&(terrain.points[:,:2]<=hi+job.max_edge),axis=1)]
|
|
if len(local)<3: raise ValueError('교각과 지형의 좌표 범위가 겹치지 않습니다. EPSG·CAD 축·측량 블록을 확인하세요.')
|
|
try: dtm=DTM(local,job.max_edge)
|
|
except QhullError as exc: raise ValueError('삼각망을 구성할 수 없습니다. 평면 지형의 표고점/등고선이 필요합니다.') from exc
|
|
bores=[]; issues=[]
|
|
if job.boreholes:
|
|
bores,issues=read_boreholes(job.boreholes,cfg,dxf_positions={b['id']:(b['X'],b['Y']) for b in terrain.boreholes})
|
|
results=[evaluate_pier(p,dtm,bores,cfg) for p in piers]
|
|
if results and not any(r.ground_center is not None for r in results):
|
|
nearest=[float(np.linalg.norm(local[:,:2]-[p.X,p.Y],axis=1).min()) for p in piers]
|
|
raise ValueError('모든 교각에서 유효 지형 삼각형을 찾지 못했습니다. 좌표 불일치 외에 측량 블록 누락·지형 공백·최대 TIN 변 길이 제한도 원인일 수 있습니다. '
|
|
f"읽은 지형 범위 N/E: {local[:,:2].min(0).round(3).tolist()} ~ {local[:,:2].max(0).round(3).tolist()}; "
|
|
f"교각별 최근접 지형점 거리(m): {[round(v,1) for v in nearest]}. 측량 블록/레이어를 확인하세요.")
|
|
nx,ny=map(int,counts); origin=np.round(lo,3)
|
|
gx,gy=np.meshgrid(origin[0]+job.step*np.arange(nx),origin[1]+job.step*np.arange(ny))
|
|
log(f'지반 삼각망/격자 생성 중... 지형 {len(local):,}점 / 교각 {len(piers)}개')
|
|
z=dtm.z_many(np.column_stack([gx.ravel(),gy.ravel()]))
|
|
if not np.isfinite(z).any(): raise ValueError('유효한 지반면이 없습니다. 표고점 간격/최대 TIN 변 길이를 확인하세요.')
|
|
if job.crs:
|
|
tr=Transformer.from_crs(job.crs,4326,always_xy=True)
|
|
lon,lat=tr.transform(gy.ravel(),gx.ravel())
|
|
if not np.isfinite([lon,lat]).all() or np.any(np.abs(lat)>85.05) or np.any(np.abs(lon)>180):
|
|
raise ValueError('지도 위치 변환 결과가 유효하지 않습니다. EPSG와 축을 확인하세요.')
|
|
data={'schema':2,'generated':date.today().isoformat(),'coord':'측량 X(북) · Y(동), m. 치수 mm',
|
|
'crs':job.crs or None,'crs_assumed':not bool(job.crs) or bool(job.auto_detection.get('crs_assumed')),
|
|
'terrain':{**terrain.stats,'points_used':len(local)},'removed_points':terrain.removed,
|
|
'boreholes_dxf':list({b['id']:b for b in [*terrain.boreholes,
|
|
*[{'id':b.id,'X':b.X,'Y':b.Y} for b in bores if b.X is not None and b.Y is not None]]}.values()),
|
|
'boreholes_source':[asdict(b) for b in bores],'borehole_issues':issues,
|
|
'piers':[pier_dict(r) for r in results],
|
|
'terrain_grid':{'origin':origin.tolist(),'step':job.step,'nx':nx,'ny':ny,
|
|
'z':[round(float(v),3) if np.isfinite(v) else None for v in z]},
|
|
'surface_provenance':{'drawing':str(Path(job.drawing).resolve()),
|
|
'sha256':hashlib.sha256(Path(job.drawing).read_bytes()).hexdigest(),
|
|
'method':'measured_height_geometry_TIN','terrain_block':job.terrain_block or None,
|
|
'source_crs':None if job.auto_detection.get('plan_registration',{}).get('status')=='registered' else job.terrain_crs or None,
|
|
'source_coordinate_frame':'drawing_layout_CAD_XY' if job.auto_detection.get('plan_registration',{}).get('status')=='registered' else 'original_survey_CAD',
|
|
'coordinate_registration':job.auto_detection.get('plan_registration'),
|
|
'target_crs':job.crs or None,
|
|
'crs_status':job.auto_detection.get('crs_status','user_specified_not_survey_verified'),
|
|
'auto_detection':job.auto_detection,
|
|
'cad_axes':job.cad_axes,'max_edge_m':job.max_edge,
|
|
'pier_source':job.piers or None,
|
|
'pier_sha256':hashlib.sha256(Path(job.piers).read_bytes()).hexdigest() if job.piers else None,
|
|
'pier_method':'IP1_IP3_midpoint' if job.piers.lower().endswith('.xlsx') else 'explicit_JSON' if job.piers else 'none',
|
|
'notes':['교각 문자 위치를 중심으로 추정하지 않음.','자료 없음/긴 TIN 삼각형은 null.','AI/OCR 호출 없음.']}}
|
|
data['map_status']={'status':'not_requested'}
|
|
if job.include_maps and not job.crs:
|
|
data['map_status']={'status':'unresolved_crs','reason':'원본 도면과 교각 Excel에 지도 좌표계를 확정할 정보가 없습니다. 지반면은 원본 측량 좌표로 생성했습니다.'}
|
|
log('지반면 생성 완료. 원본에 지도 좌표계 표기가 없어 위성/2D 지도는 확인 대기 상태로 기록합니다.')
|
|
if job.include_maps and job.crs:
|
|
log('2D 지도·위성사진 매핑 중... (신규 지역은 인터넷 필요)')
|
|
cache=out/'cache/map_layers.json'; cache.parent.mkdir(exist_ok=True)
|
|
bounds=(float(gx.min()),float(gy.min()),float(gx.max()),float(gy.max()))
|
|
key=hashlib.sha256(json.dumps([job.crs,bounds]).encode()).hexdigest()
|
|
try:
|
|
saved=json.loads(cache.read_text(encoding='utf-8')) if cache.exists() else {}
|
|
with projection(job.crs):
|
|
features=map_export._features_from_json(data)
|
|
model=GroundModel(terrain,dtm,bores,issues,results,cfg)
|
|
features['features'].extend(f for f in map_export._features(model)['features'] if f.get('properties',{}).get('kind')=='footing')
|
|
if saved.get('request_key')==key:
|
|
visualization=saved['visualization']; visualization['features']=features
|
|
log('동일 범위의 실제 지도 캐시 재사용')
|
|
else:
|
|
paths=[cache,*sorted((ROOT/'data/cache').glob('*/map_layers.json'))]
|
|
visualization,cache_source=covering_cached_maps(job.crs,bounds,paths)
|
|
if visualization is not None:
|
|
visualization['features']=features
|
|
log(f'요청 범위를 포함하는 실제 지도 캐시 사용: {cache_source}')
|
|
else:
|
|
if not job.allow_download:
|
|
raise ValueError('이 위치를 포함하는 지도 캐시가 없습니다. 새 지도 취득은 화면에서 「새 지도 다운로드 허용」을 선택하세요. 위치 지역이 OSM/Esri에 전달됩니다.')
|
|
extent=map_export._tile_extent_for_survey_bounds(*bounds)
|
|
visualization=map_export._visualization(extent,features)
|
|
visualization.update(source_crs=job.crs,source_crs_status=job.auto_detection.get('crs_status','user_specified_not_survey_verified'))
|
|
visualization['georeferencing'].update(crs=job.crs,controlPoints=0,pixelResidualMax=None,pixelResidualRms=None)
|
|
visualization['notes']=[n for n in visualization['notes'] if 'EPSG:5186' not in n]
|
|
visualization['notes'].append('공인 기준점으로 위치 정확도를 검증하지 않았습니다.')
|
|
if job.auto_detection.get('project_default_applied'):
|
|
visualization['notes'].append(f'원본에 좌표계 표기가 없어 프로젝트 기본 {job.crs} 적용. 원본에서 확인된 좌표계가 아닙니다.')
|
|
visualization['axis_mapping']={'model_axes':{'X':'northing','Y':'easting'},'map_input_order':['Y','X'],'swap_xy':True}
|
|
for layer in visualization['layers']: layer['source_url_template']=map_export._PROVIDERS[layer['id']]['url']
|
|
for layer in visualization['layers']: map_export.validate_embedded_image(layer)
|
|
data['visualization']=visualization
|
|
data['map_status']={'status':'mapped','crs':job.crs,
|
|
'crs_assumed':data['crs_assumed'],
|
|
'crs_status':job.auto_detection.get('crs_status','user_specified_not_survey_verified')}
|
|
atomic_json(cache,{'request_key':key,'visualization':visualization})
|
|
except Exception as exc:
|
|
data['map_status']={'status':'image_unavailable','reason':str(exc)}
|
|
if not job.automatic:
|
|
atomic_json(out/'ground_model_without_maps.json',data)
|
|
raise RuntimeError('지반면은 ground_model_without_maps.json에 저장했습니다. 지도 취득 실패로 기존 지도 포함 JSON은 유지합니다: '+str(exc)) from exc
|
|
log('지반면 생성 완료. 지도 이미지 취득은 완료하지 못했습니다: '+str(exc))
|
|
filename='ground_model_3d_with_maps.json' if data['map_status']['status']=='mapped' else ('ground_model.json' if job.automatic else 'ground_model_without_maps.json')
|
|
atomic_json(out/filename,data)
|
|
# Refresh a former partial-result alias so opening it cannot silently show
|
|
# the old unmapped model after maps have succeeded for these same inputs.
|
|
prior=out/'ground_model.json'
|
|
if data['map_status']['status']=='mapped' and prior.is_file():
|
|
try:
|
|
previous=json.loads(prior.read_text(encoding='utf-8'))
|
|
provenance=previous.get('surface_provenance',{})
|
|
if provenance.get('sha256')==data['surface_provenance']['sha256'] and provenance.get('pier_source')==data['surface_provenance']['pier_source']:
|
|
atomic_json(prior,data)
|
|
except (OSError,ValueError):pass
|
|
if job.plots:
|
|
from .plot import plot_plan, plot_section
|
|
model=GroundModel(terrain,dtm,bores,issues,results,cfg)
|
|
plot_plan(model,out/'plan.png')
|
|
for result in results: plot_section(result,model,out/f'section_{result.pier.id}.png')
|
|
log(f'완료: {out/filename}\n유효 지반 격자 {int(np.isfinite(z).sum()):,}개 / 교각 {len(piers)}개')
|
|
if not piers: log('교각 좌표 미입력: 지반면과 지도만 생성했습니다. Excel IP 좌표/교각 JSON을 지정하면 교각이 추가됩니다.')
|
|
return data
|
|
|
|
|
|
def build_from_files(drawing,piers,log=lambda message:None,output=None):
|
|
"""Two-file entry point; no user coordinate/block configuration is required."""
|
|
name=re.sub(r'[^\w.-]+','_',Path(drawing).stem).strip('.') or 'new_drawing'
|
|
job=DrawingJob(drawing=str(drawing),piers=str(piers),
|
|
output=str(output or ROOT/'data/output'/('drawing_'+name)),crs='',
|
|
automatic=True,include_maps=True,allow_download=True)
|
|
return build_drawing(job,log)
|