s0wlz (Matthias Puchstein)
f2d71da8e1
Replaced the memory-intensive full image read of per-tile orthophotos with a windowed read from the global Ortho VRT. This fixes memory crashes and ensures correct texture coverage for buildings that cross tile boundaries.
108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
import numpy as np
|
|
import os
|
|
from geodata_pipeline.buildings import export_buildings
|
|
from geodata_pipeline.config import Config
|
|
|
|
class TestBuildingsTexture(unittest.TestCase):
|
|
@patch("geodata_pipeline.buildings.gdal.Open")
|
|
@patch("geodata_pipeline.buildings._ensure_cityjson_for_tile")
|
|
@patch("geodata_pipeline.buildings._load_cityjson")
|
|
@patch("geodata_pipeline.buildings._collect_faces")
|
|
@patch("geodata_pipeline.buildings._compose_glb")
|
|
@patch("geodata_pipeline.buildings.ensure_dir")
|
|
@patch("geodata_pipeline.buildings.os.path.exists")
|
|
@patch("builtins.open")
|
|
def test_vrt_windowed_sampling(self, mock_open_file, mock_exists, mock_ensure_dir, mock_compose, mock_collect, mock_load_cj, mock_ensure_cj, mock_gdal_open):
|
|
# Setup mocks
|
|
mock_exists.return_value = True
|
|
|
|
# Mock file handle for manifest CSV
|
|
mock_handle = MagicMock()
|
|
mock_open_file.return_value.__enter__.return_value = mock_handle
|
|
mock_handle.__iter__.return_value = [
|
|
"tile_id,xmin,ymin,xmax,ymax,global_min,global_max,out_res,tile_key,tile_min,tile_max\n",
|
|
"tile1,1000,1000,2000,2000,0,100,1025,1_1,0,100\n"
|
|
]
|
|
|
|
mock_ensure_cj.return_value = "dummy.json"
|
|
mock_load_cj.return_value = {"CityObjects": {}}
|
|
mock_collect.return_value = (
|
|
[[10.0, 10.0, 5.0]], # vertices (local to tile)
|
|
[([0, 0, 0], "WallSurface")] # faces
|
|
)
|
|
|
|
# Mock GDAL
|
|
# 1. Heightmap VRT (called first for ground snapping)
|
|
# 2. Ortho VRT (called second for texturing)
|
|
mock_height_ds = MagicMock()
|
|
mock_ortho_ds = MagicMock()
|
|
|
|
def gdal_open_side_effect(path):
|
|
if "dgm.vrt" in path:
|
|
return mock_height_ds
|
|
if "dop.vrt" in path:
|
|
return mock_ortho_ds
|
|
# Current implementation might try to open a JPG
|
|
if ".jpg" in path:
|
|
return MagicMock()
|
|
return None
|
|
|
|
mock_gdal_open.side_effect = gdal_open_side_effect
|
|
|
|
# Setup Heightmap DS
|
|
mock_height_ds.GetGeoTransform.return_value = (0, 1, 0, 3000, 0, -1)
|
|
mock_height_ds.RasterXSize = 5000
|
|
mock_height_ds.RasterYSize = 5000
|
|
mock_height_band = MagicMock()
|
|
mock_height_ds.GetRasterBand.return_value = mock_height_band
|
|
mock_height_band.GetNoDataValue.return_value = -9999
|
|
mock_height_band.ReadAsArray.return_value = np.array([[50.0]])
|
|
|
|
# Setup Ortho DS
|
|
mock_ortho_ds.GetGeoTransform.return_value = (0, 0.2, 0, 3000, 0, -0.2) # 20cm resolution
|
|
mock_ortho_ds.RasterXSize = 15000
|
|
mock_ortho_ds.RasterYSize = 15000
|
|
mock_ortho_ds.RasterCount = 3
|
|
mock_ortho_band = MagicMock()
|
|
mock_ortho_ds.GetRasterBand.return_value = mock_ortho_band
|
|
mock_ortho_band.ReadAsArray.return_value = np.array([[255, 0, 0]]) # Red pixel
|
|
|
|
cfg = Config.default()
|
|
cfg.work.heightmap_vrt = "work/dgm.vrt"
|
|
cfg.work.ortho_vrt = "work/dop.vrt"
|
|
cfg.export.manifest_path = "manifest.csv"
|
|
cfg.export.ortho_dir = "ortho_jpg"
|
|
|
|
export_buildings(cfg)
|
|
|
|
# Verify that we opened the Ortho VRT
|
|
mock_gdal_open.assert_any_call("work/dop.vrt")
|
|
|
|
# Verify that ReadAsArray was called with a window on the Ortho DS bands
|
|
# We check the arguments of the LAST call to ReadAsArray on the ortho band
|
|
# (since it iterates over 3 bands)
|
|
args, kwargs = mock_ortho_band.ReadAsArray.call_args
|
|
|
|
# It should be windowed: (xoff, yoff, xsize, ysize)
|
|
self.assertEqual(len(args), 4)
|
|
|
|
# Verify window calculation
|
|
# tile bounds: xmin=1000, ymin=1000, xmax=2000, ymax=2000
|
|
# ortho GT: origin=(0, 3000), res=0.2
|
|
# xoff = (1000 - 0) / 0.2 = 5000
|
|
# yoff = (2000 - 3000) / -0.2 = 5000
|
|
xoff, yoff, xsize, ysize = args
|
|
self.assertEqual(xoff, 5000)
|
|
self.assertEqual(yoff, 5000)
|
|
|
|
# xsize = (2000 - 1000) / 0.2 = 5000
|
|
# ysize = (2000 - 1000) / 0.2 = 5000
|
|
# Since implementation adds +1 for safety/inclusive bounds:
|
|
self.assertTrue(xsize >= 5000)
|
|
self.assertTrue(ysize >= 5000)
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|