Files
Pier_v1/pierzainer_ground/inspect_recovered_sources.py
T

41 lines
2.4 KiB
Python

from pathlib import Path
from collections import Counter
import json,zipfile
from ezdxf import recover
from build_sample_grounds import ROOT,OUTPUT,save
from pierzainer_ground.dxf_terrain import convert_dwg_to_dxf,read_source_document
from pyproj import CRS
manifest=json.loads((ROOT/'data/input/samples_20261102/recovered/source_manifest.json').read_text(encoding='utf-8'))
results=[]
for record in manifest:
source=Path(record['local']);cache=ROOT/'data/cache/recovered_sources';cache.mkdir(exist_ok=True)
dxf=convert_dwg_to_dxf(source,cache);doc,_=read_source_document(dxf,cache)
layers={};blocks=[]
for b in doc.blocks:
counts=Counter(e.dxftype() for e in b)
if len(b)>100:blocks.append(dict(name=b.name,types=dict(counts)))
for e in doc.modelspace():
stats=layers.setdefault(e.dxf.layer,dict(count=0,types=Counter(),heights=[]))
stats['count']+=1;stats['types'][e.dxftype()]+=1
if e.dxftype()=='POINT':stats['heights'].append(float(e.dxf.location.z))
elif e.dxftype()=='LWPOLYLINE':stats['heights'].append(float(e.dxf.elevation))
elif e.dxftype()=='POLYLINE':stats['heights'].append(float(e.dxf.elevation.z))
for name,stats in layers.items():
h=stats.pop('heights');stats['z_range']=[min(h),max(h)] if h else None
result={**record,'dxf':str(dxf),'header_bounds':[list(doc.header.get('$EXTMIN',(0,0,0))),list(doc.header.get('$EXTMAX',(0,0,0)))],
'layers':layers,'blocks':blocks}
if record.get('archive_entry'):
with zipfile.ZipFile(record['source']) as archive:
prjs=[i for i in archive.infolist() if i.filename.lower().endswith('.prj')]
result['projection_files']=[]
for p in prjs:
text=archive.read(p).decode('utf-8-sig',errors='replace')
try:epsg=CRS.from_wkt(text).to_epsg()
except Exception:epsg=None
result['projection_files'].append(dict(entry=p.filename,wkt=text,epsg=epsg))
results.append(result);save(OUTPUT/'recovered_source_inspection.json',results)
print(json.dumps({'samples':record['samples'],'file':source.name,'bounds':result['header_bounds'],'blocks':blocks[:4],
'top_layers':sorted(layers.items(),key=lambda pair:pair[1]['count'],reverse=True)[:7],
'prj':result.get('projection_files')},ensure_ascii=False),flush=True)