Files
Pier_v1/pierzainer_ground/build_ground_models.py

73 lines
3.0 KiB
Python

"""Build production ground + satellite + 2D-map JSONs using preserved real data."""
import argparse
import base64
import copy
import io
import json
from pathlib import Path
from PIL import Image
from pierzainer_ground.project_coordinates import update_pier_centers, update_map_pier_features
from pierzainer_ground.survey_terrain import rebuild_terrain
ROOT = Path(__file__).resolve().parent
def read(path):
return json.loads(path.read_text(encoding='utf-8'))
def validate(data):
if data.get('schema') != 2:
raise ValueError('Expected existing schema 2')
grid = data['terrain_grid']
if len(grid['z']) != grid['nx'] * grid['ny']:
raise ValueError('Terrain grid dimensions do not match')
layers = {v['id']: v for v in data['visualization']['layers']}
for name in ('satellite', 'ground'):
layer = layers[name]
uri = layer['image']
if not uri.startswith('data:image/'):
raise ValueError(f'Missing embedded image: {name}')
with Image.open(io.BytesIO(base64.b64decode(uri.split(',', 1)[1]))) as image:
image.verify()
if len(layer['bounds']) != 2:
raise ValueError(f'Missing geographic bounds: {name}')
if not data['piers'] or not any(z is not None for z in grid['z']):
raise ValueError('Empty pier or terrain data')
def build(bridge):
folder = ROOT / 'data/input' / bridge
project = read(folder / 'project.json')
data = copy.deepcopy(read(folder / project['source_model']))
data['visualization'] = read(ROOT / 'data/cache' / bridge / 'map_layers.json')
if project['terrain_source_available']:
update_pier_centers(data, folder / project['workbook'])
data = rebuild_terrain(data, project)
data['crs'] = project['crs']
data['crs_assumed'] = True
update_map_pier_features(data, project['crs'])
validate(data)
target = ROOT / 'data/output' / bridge / 'ground_model_3d_with_maps.json'
target.parent.mkdir(parents=True, exist_ok=True)
content = json.dumps(data, ensure_ascii=False, indent=2, allow_nan=False)
# Validate the serialized payload before replacing the last successful file.
validate(json.loads(content))
temporary = target.with_suffix('.json.tmp')
temporary.write_text(content, encoding='utf-8')
temporary.replace(target)
print(json.dumps({'bridge': bridge, 'output': str(target),
'piers': len(data['piers']),
'terrain_nodes': sum(z is not None for z in data['terrain_grid']['z']),
'maps': ['satellite', 'ground'],
'terrain': 'source_DXF_TIN' if project['terrain_source_available'] else 'preserved_original_result'}, ensure_ascii=True))
return data
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--bridge', choices=['all', 'ungok', 'sanwoe'], default='all')
args = parser.parse_args()
for bridge in (['ungok', 'sanwoe'] if args.bridge == 'all' else [args.bridge]):
build(bridge)