diff options
Diffstat (limited to 'tests/test_tiles.py')
| -rw-r--r-- | tests/test_tiles.py | 1383 |
1 files changed, 1383 insertions, 0 deletions
diff --git a/tests/test_tiles.py b/tests/test_tiles.py new file mode 100644 index 0000000..bf0e618 --- /dev/null +++ b/tests/test_tiles.py @@ -0,0 +1,1383 @@ +import io +import json +import os +import shutil +import tempfile +import threading +import time +import unittest +from pathlib import Path +from unittest import mock + +import numpy as np +from PIL import Image + +from mapgen.sphere import latlon_to_xyz + +import serve +import tiles as T +from tests.test_serve import Running, built_world, request +from tests.helpers import big_river + + +def tile_at(lat, lon, z): + span = 180.0 / 2 ** z + return int((lon + 180) // span), int((90 - lat) // span) + + +class TilesTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.root = built_world() + cls.cache = Path(tempfile.mkdtemp()) + cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=cls.cache) + + @classmethod + def tearDownClass(cls): + shutil.rmtree(cls.cache) + + def test_tile_scheme(self): + self.assertTrue(T.tile_ok(0, 1, 0)) + self.assertFalse(T.tile_ok(0, 2, 0)) + self.assertFalse(T.tile_ok(3, 0, 8)) + self.assertFalse(T.tile_ok(17, 0, 0)) + lat, lon = T.tile_lat_lon(0, 0, 0, np.array([0, 255])) + self.assertAlmostEqual(lat[0], 90 - 0.5 / 256 * 180) + self.assertAlmostEqual(lon[1], -180 + 255.5 / 256 * 180) + + def test_coarse_zoom_is_the_build_raster(self): + f = self.src.fields(0, 0, 0, 0, ("z", "cat")) + base = self.src.bilinear(self.src.surface(), f["lat"], f["lon"]) + dry = ~f["lake"].astype(bool) # (lakes lie flat at their level) + np.testing.assert_array_equal(f["z"][dry], base[dry]) # the 8 km detail is below this zoom's pixel size + + def test_detail_appears_when_zoomed_in(self): + f = self.src.fields(12, *tile_at(10.0, 20.0, 12), 0, ("z",)) + base = self.src.bilinear(self.src.raster("elevation"), f["lat"], f["lon"]) + self.assertGreater(float(np.std(f["z"] - base)), 1.0) + self.assertTrue(np.all(np.isfinite(f["z"]))) + + def test_seams_match(self): + z, x, y = 7, *tile_at(5.0, 10.0, 7) + a = self.src.fields(z, x, y, 1, ("z", "cat")) + east = self.src.fields(z, x + 1, y, 1, ("z", "cat")) + south = self.src.fields(z, x, y + 1, 1, ("z", "cat")) + np.testing.assert_allclose(a["z"][:, -1], east["z"][:, 1], atol=1e-6) + np.testing.assert_array_equal(a["holdridge"][:, -1], east["holdridge"][:, 1]) + np.testing.assert_allclose(a["z"][-1, :], south["z"][1, :], atol=1e-6) + + def test_antimeridian_seam(self): + z = 6 + y = tile_at(0.0, 0.0, z)[1] + west = self.src.fields(z, 2 ** (z + 1) - 1, y, 1, ("z",)) + east = self.src.fields(z, 0, y, 1, ("z",)) + np.testing.assert_allclose(west["z"][:, -1], east["z"][:, 1], atol=1e-6) + + def test_detail_amplitude_is_continuous(self): + # blending landform amplitudes across cells must not jump where the set of nearest cells changes + lon = np.linspace(-60.0, 60.0, 40001) + amp, ridge, sea = self.src._amp(latlon_to_xyz(np.full_like(lon, 10.0), lon)) + for a in (amp, ridge, sea): + span = a.max() - a.min() + if span > 0: + self.assertLess(np.abs(np.diff(a)).max(), 0.02 * span) + + def test_coast_rule_is_continuous(self): + sea = np.linspace(0.0, 1.0, 1001) + for z in (-3000.0, -100.0, 0.0, 100.0, 3000.0): + w = T.coast_water(np.full_like(sea, z), sea) + self.assertFalse(w[sea < 0.001].any()) + self.assertTrue(w[sea > 0.999].all()) + self.assertEqual(int(np.sum(np.diff(w.astype(int)) != 0)), 1, z) # one clean switch, no hard cut at a threshold + self.assertEqual(bool(T.coast_water(np.array([z]), np.array([0.5]))[0]), z < 0) + # below sea level, water reaches further into the mixed band than a flat threshold would allow + self.assertTrue(T.coast_water(np.array([-3000.0]), np.array([0.05]))[0]) + + def test_coast_band(self): + f = self.src.fields(4, *tile_at(10.0, -70.0, 4), 0, ("z", "water")) # an 11° tile over the small world's west coast + self.assertTrue(f["water"][f["sea"] > 0.999].all()) + self.assertFalse(f["water"][f["sea"] < 0.001].any()) + self.assertTrue(f["water"].any() and (~f["water"]).any(), "tile should hold both land and sea") + + def test_layers_render(self): + for layer in T.LAYERS: + im = Image.open(io.BytesIO(self.src.tile(layer, 5, *tile_at(10.0, 20.0, 5)))) + self.assertEqual((im.format, im.size), ("JPEG", (256, 256)), layer) + h = Image.open(io.BytesIO(self.src.tile("height", 5, *tile_at(10.0, 20.0, 5)))) + self.assertEqual((h.format, h.size), ("PNG", (256, 256))) + self.assertIn(h.mode, ("I;16", "I")) + for y in (0, 2 ** 5 - 1): # polar rows + Image.open(io.BytesIO(self.src.tile("relief", 5, 3, y))).load() + + def test_deterministic_and_cached(self): + args = ("relief", 9, *tile_at(12.0, 25.0, 9)) + a = self.src.tile(*args) + other = T.TileSource(serve.World(self.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + self.assertEqual(a, other.render(*args)) + self.assertTrue((self.cache / self.src.tag / "relief" / "9").is_dir()) + t = time.time() + self.src.tile(*args) + self.assertLess(time.time() - t, 0.05) + + def test_speed(self): + t = time.time() + self.src.render("relief", 14, *tile_at(3.0, 5.0, 14)) + self.assertLess(time.time() - t, 3.0) + + def test_fingerprint_follows_the_build(self): + import os + a = T.TileSource(serve.World(self.root, 2), seed=7) + self.assertEqual(a.fingerprint, T.TileSource(serve.World(self.root, 2), seed=7).fingerprint) + self.assertNotEqual(a.fingerprint, T.TileSource(serve.World(self.root, 2), seed=8).fingerprint) + self.assertIn(a.fingerprint, str(a.cache_dir)) + self.assertIn(a.fingerprint, a.meta()["url"]) + elev = self.root / "out" / "r2" / "raster" / "elevation.png" + st = elev.stat() + os.utime(elev, ns=(st.st_atime_ns, st.st_mtime_ns + 10**9)) # a rebuild rewrites the rasters + self.assertNotEqual(a.fingerprint, T.TileSource(serve.World(self.root, 2), seed=7).fingerprint) + + def test_z_at(self): + z = self.src.z_at(10.0, 20.0) + self.assertEqual(z.shape, (1,)) + self.assertTrue(np.isfinite(z[0])) + + + def test_mesh_heights(self): + z, (x, y) = 9, tile_at(40.0, 10.0, 9) + h = self.src.mesh(z, x, y)[0] + self.assertEqual(h.shape, (T.MESH_N, T.MESH_N)) + self.assertTrue(np.isfinite(h).all()) + np.testing.assert_array_equal(h[:, -1], self.src.mesh(z, x + 1, y)[0][:, 0]) # shared edges: same points, same heights + np.testing.assert_array_equal(h[-1, :], self.src.mesh(z, x, y + 1)[0][0, :]) + n = 2 ** (z + 1) + np.testing.assert_allclose(self.src.mesh(z, n - 1, y)[0][:, -1], self.src.mesh(z, 0, y)[0][:, 0], atol=1e-3) + + def test_mesh_true_heights_and_water(self): + for x in range(64): # a coastal z5 tile + h, water = self.src.mesh(5, x, 12) + if water.any() and (~water).any(): + break + else: + self.fail("no coastal z5 tile on row 12") + span = 180.0 / 32 + f = np.arange(T.MESH_N) / (T.MESH_N - 1) + LA, LO = np.meshgrid(90.0 - (12 + f) * span, -180.0 + (x + f) * span, indexing="ij") + z, wet = self.src.surface_at(LA.ravel(), LO.ravel(), span / 128 * np.pi / 180 * self.src.R) + np.testing.assert_allclose(h, z.reshape(h.shape), atol=1e-6) + self.assertLess(float(h[water].min()), -100.0, "the sea floor, not the water surface") + self.assertTrue(np.all(h[water] < 60.0), "water only where the coast rule puts sea") + + def test_mesh_cached_and_fast(self): + args = (11, *tile_at(-20.0, -35.0, 11)) + t = time.time() + b = self.src.tile("mesh", *args) + self.assertLess(time.time() - t, 0.5) + self.assertEqual(len(b), T.MESH_BYTES) + h, water = self.src.mesh(*args) + n2 = T.MESH_N * T.MESH_N + np.testing.assert_array_equal(np.frombuffer(b[:4 * n2], "<f4"), h.astype("<f4").ravel()) + np.testing.assert_array_equal(np.frombuffer(b[4 * n2:], np.uint8), water.astype(np.uint8).ravel()) + self.assertTrue((self.cache / self.src.tag / "mesh" / "11" / str(args[1]) / f"{args[2]}.bin").is_file()) + + def test_a_saved_mesh_of_the_wrong_size_is_rendered_again(self): + args = (10, *tile_at(-21.0, -36.0, 10)) + f = self.src.cache_path("mesh", *args) + f.parent.mkdir(parents=True, exist_ok=True) + f.write_bytes(b"\0" * (T.MESH_N * T.MESH_N * 4)) + self.assertEqual(len(self.src.tile("mesh", *args, remember=False)), T.MESH_BYTES) + self.assertEqual(f.stat().st_size, T.MESH_BYTES) + + def test_surface_matches_the_tile_pixels(self): + z, (x, y) = 8, tile_at(10.0, -70.0, 8) # the test world's west coast + f = self.src.fields(z, x, y, 0, ("z", "cat")) + zz, water = self.src.surface_at(f["lat"].ravel(), f["lon"].ravel(), f["px_km"]) + tile_water = (f["water"] | f["lake"].astype(bool)).ravel() + self.assertLess(np.mean(water != tile_water), 0.02) + self.assertLess(np.mean(np.abs(zz - f["z"].ravel())), 10.0) + + + def _river_tile(self, z): + net = self.src.river_net + a = self.src.w.arrays + wet = a["lake"].astype(bool) | a["ocean"].astype(bool) + ok = a["river"][net.seg_a].astype(bool) & ~wet[net.seg_a] & ~wet[a["recv"][net.seg_a]] # on over land + i = int(net.seg_a[np.argmax(np.where(ok, net.half_w, 0))]) # the widest river + xyz = self.src.w.arrays["g_xyz"][i] + lat, lon = np.degrees(np.arcsin(xyz[2])), np.degrees(np.arctan2(xyz[1], xyz[0])) + return (z, *tile_at(lat, lon, z)) + + def test_valleys_cut_never_raise_and_hold_water(self): + args = self._river_tile(9) + f = self.src.fields(*args, 0, ("z", "cat")) + raw = self.src.fields(*args, 0, ("z", "cat"), valleys=False) + rel = self.src.relief_at(f["lat"].ravel(), f["lon"].ravel(), f["px_km"]).reshape(f["z"].shape) + self.assertTrue(np.all(f["z"] <= np.maximum(raw["z"], rel) + 1.0), "at most a micro-relief dip is kept dry") + self.assertTrue(f["river"].any(), "the river channel is water in its tile") + self.assertTrue((raw["z"] - f["z"]).max() > 1.0, "a valley was cut") + + def test_valley_seams_match(self): + z, x, y = self._river_tile(10) + a = self.src.fields(z, x, y, 1, ("z",)) + east = self.src.fields(z, x + 1, y, 1, ("z",)) + np.testing.assert_allclose(a["z"][:, -1], east["z"][:, 1], atol=1e-6) + + def test_valley_tiles_stay_fast_near_big_rivers(self): + for z in (12, 16): + args = self._river_tile(z) + t = time.time() + self.src.render("relief", *args) + self.assertLess(time.time() - t, 1.5, f"z{z}") + t = time.time() + self.src.surface_at(np.linspace(-40, 40, 256), np.full(256, 20.0), 20.0) # a 17,800 km profile + self.assertLess(time.time() - t, 0.5) + + def test_valleys_fade_in(self): + z, x, y = self._river_tile(5) + f = self.src.fields(z, x, y, 0, ("z",)) + raw = self.src.fields(z, x, y, 0, ("z",), valleys=False) + self.assertLess(np.abs(f["z"] - raw["z"]).max(), 2000.0) + + + def test_relief_draws_the_river(self): + args = self._river_tile(11) + f = self.src.fields(*args, 1, ("z", "water", "cat")) + img = self.src.rgb("relief", *args) + lake = np.array([79, 143, 192]) + river = f["river_draw"][1:-1, 1:-1] & ~f["water"][1:-1, 1:-1] + self.assertTrue(river.any()) + shade = img[river].astype(float) / lake + self.assertTrue(np.all(np.abs(shade - shade[:, :1]) < 0.02), "river pixels are the (shaded) water colour") + + def test_tiles_draw_micro_relief_over_the_refine_surface(self): + z, x, y = 15, *tile_at(12.0, 25.0, 15) + f = self.src.fields(z, x, y, 0, ("z",), valleys=False) + rel = self.src.relief_at(f["lat"].ravel(), f["lon"].ravel(), f["px_km"]).reshape(f["z"].shape) + d = f["z"] - rel # relief_at (refine's start) stays without it + self.assertGreater(float(np.std(d)), 0.5) + self.assertLess(float(np.abs(d).max()), 60.0) + + def test_micro_relief_never_digs_dry_ground_below_the_river(self): + z, x, y = self._river_tile(14) + f = self.src.fields(z, x, y, 0, ("z",)) + ch, zz = f["river"], f["z"] + self.assertTrue(ch.any() and (~ch).any()) + for s in ((0, 1), (0, -1), (1, 0), (-1, 0)): # dry pixels beside the channel: not below its water + wet, h = np.roll(ch, s, (0, 1)), np.roll(zz, s, (0, 1)) + dry = wet & ~ch + dry[[0, -1], :] = dry[:, [0, -1]] = False + self.assertTrue(np.all(zz[dry] >= h[dry] - 1e-6), s) + + def test_land_colour_varies_within_one_zone(self): + args = (13, *tile_at(12.0, 25.0, 13)) + f = self.src.fields(*args, 0, ("z", "water", "cat")) + img = self.src.rgb("relief", *args).astype(float) + land = ~f["water"] & ~f["lake"].astype(bool) & ~f["river_draw"] & (f["ice"] == 0) + zone = f["holdridge"] == np.bincount(f["holdridge"][land]).argmax() + hue = (img[..., 0] - img[..., 1]) / (img[..., 0] + img[..., 1] + 1.0) + self.assertGreater(float(np.std(hue[land & zone])), 0.004, "not one flat colour under the hillshade") + self.assertLess(float(np.std(hue[land & zone])), 0.05) + + +class RegionTilesTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + import refine as RF + cls.root = built_world() + from tests.test_refine import clear_regions + clear_regions(cls.root) + w = RF.WorldCells(cls.root / "out" / "r2") + i = big_river(w.a) + cls.lat, cls.lon = float(w.a["g_lat"][i]), float(w.a["g_lon"][i]) + reg = cls.root / "places" / "regions.json" + reg.parent.mkdir(exist_ok=True) + d = 3 + reg.write_text(json.dumps({"regions": [{"id": "r", "name": "t", "outline": [[cls.lat - d, cls.lon - d], [cls.lat - d, cls.lon + d], [cls.lat + d, cls.lon + d], [cls.lat + d, cls.lon - d]]}]})) + from mapgen import config as C + seed = int(C.load(cls.root)[0]["build"]["seed"]) # the build's seed, as the server uses + cls.plain = T.TileSource(serve.World(cls.root, 2), seed=seed, cache_dir=Path(tempfile.mkdtemp())) + RF.build_areas(cls.root, 2, reg, log=lambda s: None) + cls.src = T.TileSource(serve.World(cls.root, 2), seed=seed, cache_dir=Path(tempfile.mkdtemp())) + + @classmethod + def tearDownClass(cls): + from tests.test_refine import clear_regions + clear_regions(cls.root) + + def test_tag_changes_with_regions(self): + self.assertFalse(self.src.regions.empty) + self.assertNotEqual(self.src.tag, self.plain.tag) + self.assertRegex(self.src.tag, r"^v\d+\.\d+-[0-9a-f]+$") + + def test_region_tile_urls_follow_the_region_rendering_version(self): + from unittest import mock + with mock.patch.object(T, "REGION_VERSION", T.REGION_VERSION + 1): + other = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + self.assertNotEqual(other.tag, self.src.tag, "new region rendering: new tile URLs, no stale saved tiles") + self.assertEqual(other.base_tag, self.src.base_tag, "the world's own tiles stay") + self.assertRegex(other.tag, r"^v\d+\.\d+-[0-9a-f]+$") + + def test_inside_uses_refined_heights(self): + z, (x, y) = 8, tile_at(self.lat, self.lon, 8) + a, b = self.src.fields(z, x, y, 0, ("z",)), self.plain.fields(z, x, y, 0, ("z",)) + self.assertGreater(np.abs(a["z"] - b["z"]).mean(), 1.0) + + def test_region_edge_seams(self): + for dx in (-3, -2, 2, 3): + z = 8 + x, y = tile_at(self.lat, self.lon + dx, z) + a, e = self.src.fields(z, x, y, 1, ("z",)), self.src.fields(z, x + 1, y, 1, ("z",)) + np.testing.assert_allclose(a["z"][:, -1], e["z"][:, 1], atol=1e-6) + + def test_blend_band_has_no_step(self): + lat = np.full(400, self.lat) + lon = self.lon + np.linspace(2.0, 4.0, 400) # across the east edge (lon + 3) + z = self.src.z_at(lat, lon, 1.0) + xyz = latlon_to_xyz(lat, lon).reshape(-1, 3) + lake = self.src._flat_lakes(np.zeros(len(lat)), lat, lon, *self.src._warp_offsets(xyz, 4.0)) != 0 + shore = lake[1:] != lake[:-1] # (a world lake's shore may be a bank) + k = int(np.argmax(np.where(shore, 0.0, np.abs(np.diff(z))))) + fine = np.linspace(lon[k], lon[k + 1], 41) # 10× finer around the worst step + zf = self.src.z_at(np.full(41, self.lat), fine, 1.0) + self.assertLess(np.abs(np.diff(zf)).max(), 0.3 * abs(z[k + 1] - z[k]) + 1.0, + "continuous: the worst step shrinks with the sample spacing (a cliff would not)") + + def test_outside_unchanged_and_no_regions_unchanged(self): + z, (x, y) = 8, tile_at(self.lat + 20, self.lon, 8) + np.testing.assert_array_equal(self.src.fields(z, x, y, 0, ("z",))["z"], self.plain.fields(z, x, y, 0, ("z",))["z"]) + self.assertTrue(self.plain.regions.empty) + + def test_inspector_answers_refined(self): + w = serve.World(self.root, 2) + w.regions = self.src.regions + d = w.describe(self.lat, self.lon) + self.assertTrue(d.get("refined")) + far = w.describe(self.lat + 20, self.lon) + self.assertFalse(far.get("refined", False)) + + + def _exit(self): + import refine as RF + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + a = dict(np.load(d / "cells.npz")) + halo = a["halo"].astype(bool) + feed = np.where(a["river"].astype(bool) & ~halo & halo[a["recv"]])[0] + return a, feed + + def test_rivers_run_on_to_the_halo(self): + a, feed = self._exit() + net = self.src.regions.net + self.assertTrue(len(feed), "the test area has rivers leaving it") + self.assertTrue(np.all(net.seg_len > 0), "no zero-length segments") + ends = {tuple(np.round(v, 9)) for v in net.b_xyz} + for i in feed: + self.assertIn(tuple(np.round(a["g_xyz"][a["recv"][i]], 9)), ends, "the river reaches its halo cell") + + def test_world_rivers_stay_drawn_across_the_blend_band(self): + a, feed = self._exit() + found = False + for i in feed: + lat, lon = float(a["g_lat"][a["recv"][i]]), float(a["g_lon"][a["recv"][i]]) + for dz in (7, 8): + x, y = tile_at(lat, lon, dz) + f, g = self.src.fields(dz, x, y, 0, ("z",)), self.plain.fields(dz, x, y, 0, ("z",)) + xyz = latlon_to_xyz(f["lat"], f["lon"]).reshape(-1, 3) + w = self.src.regions.weight(xyz).reshape(f["z"].shape) + band = g["river_draw"] & (w < 1) + if band.any(): + found = True + self.assertTrue(np.all(f["river_draw"][band]), "world rivers drawn wherever the world still shows") + self.assertTrue(found) + + def test_region_valleys_reach_up_to_the_shoulders(self): + import refine as RF + import rivers as RV + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + tmp = Path(tempfile.mkdtemp()) + dst = RF.results_dir(tmp, "k") / d.name + shutil.copytree(d, dst) + a = dict(np.load(dst / "cells.npz")) + src = ~np.isin(np.arange(len(a["recv"])), a["recv"][a["recv"] != np.arange(len(a["recv"]))]) + dry = src & ~a["halo"].astype(bool) & ~a["ocean"].astype(bool) & ~a["lake"].astype(bool) # (coarse test cells + a["river"] = a["river"].astype(bool) & ~dry # are all rivers) + a["z_surface_m"] = np.where(dry, a["z_surface_m"] + 500.0, a["z_surface_m"]) # canyons + np.savez(dst / "cells.npz", **a) + rs = RF.RegionSet(tmp, "k", self.src.w.legends, self.src.R) + plain = RV.RiverNet(rs.arrays, self.src.w.legends, rs.R, levels=rs.arrays["z_surface_m"]) + self.assertTrue(np.any(rs.net.reach[: len(plain.reach)] > plain.reach + 1e-9), "walls reach up to the shoulders") + + def test_profile_inside_ignores_world_lakes(self): + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + src.w.arrays = dict(src.w.arrays, lake=np.ones(len(src.w.arrays["g_ids"]), bool)) # lakes everywhere + lat, lon = np.full(50, self.lat), self.lon + np.linspace(-0.5, 0.5, 50) + np.testing.assert_array_equal(src.surface_at(lat, lon)[1], self.src.surface_at(lat, lon)[1]) + + def test_inspector_band_answers_from_the_world(self): + rs = self.src.regions + a, feed = self._exit() + pa, pb = a["g_xyz"][feed[0]], a["g_xyz"][a["recv"][feed[0]]] + pts = pa + np.linspace(0.0, 0.6, 121)[:, None] * (pb - pa) # from the border cell toward the halo + pts /= np.linalg.norm(pts, axis=1, keepdims=True) + lat, lon = np.degrees(np.arcsin(pts[:, 2])), np.degrees(np.arctan2(pts[:, 1], pts[:, 0])) + band = [k for k in range(len(pts)) if rs.weight(pts[k:k + 1])[0] < 0.5 and rs.index_of(lat[k], lon[k]) is not None] + self.assertTrue(band, "a point of a refined cell in the outer blend band") + world = serve.World(self.root, 2) + world.regions = rs + d = world.describe(float(lat[band[0]]), float(lon[band[0]])) + self.assertFalse(d["refined"]) + self.assertEqual(d["inner_km"], 0.0, "no reuse of an answer next to a refined area") + + def test_a_server_without_regions_leaves_region_tile_caches_alone(self): + root = self.root / "out" / "r2" / "tiles" + plain = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, regions_dir=Path(tempfile.mkdtemp())) # no regions + other_world, region = root / "v1-deadbeef00", root / (plain.base_tag + "abc123") + for d in (other_world, region): + d.mkdir(parents=True, exist_ok=True) + plain.warm() + self.assertFalse(other_world.exists(), "another world's tiles go") + self.assertTrue(region.exists(), "a smoke server must not wipe the real server's region tiles") + shutil.rmtree(region) + + def test_worker_processes_render_the_same_tiles(self): + from concurrent.futures import ThreadPoolExecutor + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + src.start_workers(2) + try: + args = [("relief", 8, *tile_at(self.lat, self.lon, 8)), ("mesh", 8, *tile_at(self.lat, self.lon, 8)), + ("relief", 6, *tile_at(self.lat + 20, self.lon, 6))] + with ThreadPoolExecutor(3) as ex: + got = list(ex.map(lambda a: src.tile(*a), args)) + self.assertEqual(got, [self.src.render(*a) for a in args]) + self.assertEqual(src.pool.alive, 2) + finally: + src.pool.close() + + def test_workers_exit_when_the_server_process_dies(self): + import os + import signal + import subprocess + import sys + import textwrap + code = textwrap.dedent(f""" + import sys, socket, tempfile, pathlib; sys.path.insert(0, {str(Path(T.__file__).parent)!r}) + import tiles as T, serve + lsock = socket.socket(); lsock.bind(("127.0.0.1", 0)); lsock.listen() # a server's listening socket + src = T.TileSource(serve.World(pathlib.Path({str(self.root)!r}), 2), seed=7, cache_dir=pathlib.Path(tempfile.mkdtemp())) + src.start_workers(2) + print(lsock.getsockname()[1], *[p.pid for p in src.pool.procs], flush=True) + import time; time.sleep(60) + """) + p = subprocess.Popen([sys.executable, "-c", code], stdout=subprocess.PIPE, text=True) + port, *pids = map(int, p.stdout.readline().split()) + os.kill(p.pid, signal.SIGKILL) # no atexit: as a killed server + p.wait() + deadline = time.time() + 10 + while time.time() < deadline and any(os.path.exists(f"/proc/{q}") and open(f"/proc/{q}/stat").read().split()[2] != "Z" + for q in pids): + time.sleep(0.2) + alive = [q for q in pids if os.path.exists(f"/proc/{q}") and open(f"/proc/{q}/stat").read().split()[2] != "Z"] + for q in alive: + os.kill(q, signal.SIGKILL) + self.assertEqual(alive, [], "orphaned render workers linger") + import socket + s = socket.socket() + self.assertNotEqual(s.connect_ex(("127.0.0.1", port)), 0, "nobody still holds the server's listening socket") + s.close() + + def test_workers_follow_a_region_reload(self): + import refine as RF + empty = Path(tempfile.mkdtemp()) + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), + regions_dir=empty) + src.start_workers(1) + try: + args = ("relief", 8, *tile_at(self.lat, self.lon, 8)) + self.assertEqual(src.tile(*args), self.plain.render(*args)) + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + shutil.copytree(d, RF.results_dir(empty, src.fingerprint) / d.name) + src.reload_regions() + self.assertEqual(src.tile(*args), self.src.render(*args)) + finally: + src.pool.close() + + def test_prerender_saves_the_region_tiles(self): + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + seen = [] + n = src.prerender(max_z=6, progress=lambda done, total: seen.append((done, total))) + for z in (5, 6): + for layer in ("relief", "mesh"): + self.assertTrue(src.cache_path(layer, z, *tile_at(self.lat, self.lon, z)).exists(), (layer, z)) + self.assertFalse(src.cache_path("relief", 6, *tile_at(self.lat + 40, self.lon, 6)).exists()) + self.assertGreater(n, 0) + self.assertEqual(seen[-1], (n, n)) + self.assertEqual(src.prerender(max_z=6), 0, "saved tiles are not rendered again") + + def _empty_src(self): + return T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), + regions_dir=Path(tempfile.mkdtemp())) + + def _add_area(self, src): + import refine as RF + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + dst = RF.results_dir(src.regions_dir, src.fingerprint) / d.name + shutil.copytree(d, dst) + return dst + + def test_a_reload_during_a_render_never_caches_mixed_tiles(self): + src = self._empty_src() + args = ("relief", 8, *tile_at(self.lat, self.lon, 8)) + orig, calls = src.render, [] + + def render(*a): + data = orig(*a) # rendered with the old (no) regions … + if not calls: + calls.append(1) + self._add_area(src) + src.reload_regions() # … while a build finished + return data + src.render = render + want = self.src.render(*args) + self.assertEqual(src.tile(*args), want) + self.assertEqual(src.tile(*args), want, "nor from the memory cache") + base = src.cache_root / src.base_tag / "relief" / "8" / str(args[2]) / f"{args[3]}.jpg" + self.assertFalse(base.exists(), "no pre-build bytes in the world's cache for a tile the region covers") + + def test_workers_never_render_a_half_built_region_set(self): + src = self._empty_src() + src.start_workers(1) # forked without regions + try: + area = self._add_area(src) + src.reload_regions() # the server has the region … + shutil.rmtree(area) # … a new build has pruned it on disk meanwhile + args = ("relief", 8, *tile_at(self.lat, self.lon, 8)) + self.assertEqual(src.tile(*args), self.src.render(*args)) + finally: + src.pool.close() + + def test_dead_workers_never_hang_tile_requests(self): + import os + import signal + src = self._empty_src() + orig = src.render + src.render = lambda *a: (time.sleep(1.0), orig(*a))[1] # slow renders (the forked worker inherits this) + src.start_workers(1) + try: + args = [("relief", 9, *tile_at(self.lat, self.lon, 9)), ("relief", 9, *tile_at(self.lat + 10, self.lon, 9))] + got = [None, None] + + def run(k): + got[k] = src.tile(*args[k]) + ts = [threading.Thread(target=run, args=(k,), daemon=True) for k in (0, 1)] + ts[0].start() + time.sleep(0.15) # the first request is inside the worker … + ts[1].start() + time.sleep(0.15) # … the second waits for it … + os.kill(src.pool.procs[0].pid, signal.SIGKILL) # … and the worker dies + for t in ts: + t.join(30) + self.assertEqual(got, [self.plain.render(*a) for a in args], "both requests answered (rendered here)") + finally: + src.pool.close() + + def test_a_region_change_keeps_the_saved_tiles_of_unchanged_areas(self): + import refine as RF + from tests.test_refine import square + tmp = Path(tempfile.mkdtemp()) + far_lat = self.lat - 50 if self.lat > 0 else self.lat + 50 + a, b = square(self.lat, self.lon, 3, "a"), square(far_lat, self.lon + 90, 3, "b") + (tmp / "one.json").write_text(json.dumps({"regions": [a]})) + (tmp / "two.json").write_text(json.dumps({"regions": [a, b]})) + res = tmp / "results" + RF.build_areas(self.root, 2, tmp / "one.json", log=lambda s: None, regions_root=res) + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=tmp / "cache", regions_dir=res) + first = src.prerender(max_z=6, layers=("relief",)) + near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6)) + old_tag = src.tag + RF.build_areas(self.root, 2, tmp / "two.json", log=lambda s: None, regions_root=res) + src.reload_regions() + self.assertNotEqual(src.tag, old_tag) + self.assertTrue(src.cache_path(*near_a).exists(), "area a's saved tiles moved to the new region set") + self.assertFalse((src.cache_root / old_tag).exists()) + again = src.prerender(max_z=6, layers=("relief",)) + self.assertGreater(again, 0, "area b's tiles are new") + self.assertLess(again, first + again - 1, "a's are not rendered again") + self.assertLessEqual(again, first * 1.5) + + def _two_areas(self): + import refine as RF + from tests.test_refine import square + tmp = Path(tempfile.mkdtemp()) + far_lat = self.lat - 50 if self.lat > 0 else self.lat + 50 + a, b = square(self.lat, self.lon, 3, "a"), square(far_lat, self.lon + 90, 3, "b") + (tmp / "one.json").write_text(json.dumps({"regions": [a]})) + (tmp / "two.json").write_text(json.dumps({"regions": [a, b]})) + RF.build_areas(self.root, 2, tmp / "one.json", log=lambda s: None, regions_root=tmp / "results") + return tmp + + def _source(self, tmp): + return T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=tmp / "cache", + regions_dir=tmp / "results") + + def _rekey(self, src): + """Area a's content changes (as after a new refine model or config) while its outline stays.""" + import refine as RF + (link,) = RF.built_areas(src.regions_dir, src.fingerprint) + old = Path(os.path.realpath(link)) + new = old.with_name(old.name + "x") + shutil.copytree(old, new) + os.utime(new / "cells.npz", ns=(time.time_ns(), time.time_ns() + 10 ** 9)) + link.unlink() + link.symlink_to(os.path.relpath(new, link.parent)) + + def test_a_region_change_while_stopped_keeps_the_saved_tiles_of_unchanged_areas(self): + import refine as RF + tmp = self._two_areas() + src = self._source(tmp) + src.carry_stale() # a server start + src.prerender(max_z=6, layers=("relief",)) + near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6)) + old_tag = src.tag + RF.build_areas(self.root, 2, tmp / "two.json", log=lambda s: None, regions_root=tmp / "results") + src = self._source(tmp) # the next start + self.assertNotEqual(src.tag, old_tag) + src.carry_stale() + self.assertTrue(src.cache_path(*near_a).exists(), "area a's saved tiles moved to the new region set") + self.assertFalse((src.cache_root / old_tag).exists()) + + def test_saved_tiles_of_an_area_whose_content_changed_are_dropped(self): + tmp = self._two_areas() + src = self._source(tmp) + src.carry_stale() + src.prerender(max_z=6, layers=("relief",)) + near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6)) + self._rekey(src) + src.reload_regions() + self.assertFalse(src.cache_path(*near_a).exists(), "a live reload drops them") + src.carry_stale() + src.prerender(max_z=6, layers=("relief",)) + self._rekey(src) + src = self._source(tmp) + src.carry_stale() + self.assertFalse(src.cache_path(*near_a).exists(), "so does the next start") + + def test_saved_region_tiles_drawn_by_an_older_region_version_are_not_carried_over(self): + tmp = self._two_areas() + src = self._source(tmp) + src.carry_stale() + src.prerender(max_z=6, layers=("relief",)) + near_a = ("relief", 6, *tile_at(self.lat, self.lon, 6)) + with mock.patch.object(T, "REGION_VERSION", T.REGION_VERSION + 1): # the next start draws refined areas anew + src = self._source(tmp) + src.carry_stale() + self.assertFalse(src.cache_path(*near_a).exists(), "tiles drawn the old way re-render") + + def test_prerendering_never_delays_the_server_exit(self): + import subprocess + import sys + import textwrap + code = textwrap.dedent(f""" + import sys, tempfile, threading, time, pathlib; sys.path.insert(0, {str(Path(T.__file__).parent)!r}) + import tiles as T, serve + src = T.TileSource(serve.World(pathlib.Path({str(self.root)!r}), 2), seed={self.src.seed}, + cache_dir=pathlib.Path(tempfile.mkdtemp())) + threading.Thread(target=src.prerender, kwargs=dict(max_z=12, threads=2), daemon=True).start() + time.sleep(1.0) + print("exiting", flush=True) # as after Ctrl-C: main returns + """) + t = time.time() + p = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, timeout=120) + self.assertIn("exiting", p.stdout, p.stderr[-500:]) + self.assertLess(time.time() - t, 20, "the process ends without rendering the whole queue first") + + def test_a_reload_forks_fresh_workers_sharing_the_new_regions(self): + src = self._empty_src() + src.start_workers(2) + try: + before = [p.pid for p in src.pool.procs] + self._add_area(src) + src.reload_regions() + after = [p.pid for p in src.pool.procs] + self.assertTrue(set(before).isdisjoint(after), "new processes (they share the loaded regions)") + self.assertEqual(src.pool.alive, 2) + args = ("relief", 8, *tile_at(self.lat, self.lon, 8)) + self.assertEqual(src.tile(*args), self.src.render(*args)) + time.sleep(0.5) + self.assertFalse(any(os.path.exists(f"/proc/{q}") and open(f"/proc/{q}/stat").read().split()[2] != "Z" + for q in before), "the old workers are gone") + finally: + src.pool.close() + + def test_workers_forked_by_a_short_lived_thread_keep_working(self): + src = self._empty_src() + src.start_workers(2) + try: + self._add_area(src) + t = threading.Thread(target=src.reload_regions) # as a region build's thread does + t.start() + t.join() + time.sleep(0.5) # the forking thread is gone now + self.assertEqual(src.pool.alive, 2) + self.assertTrue(all(p.is_alive() for p in src.pool.procs), "workers outlive the thread that forked them") + args = ("relief", 8, *tile_at(self.lat, self.lon, 8)) + data = src.pool.render(src.name, *args, src.regions.fingerprint) # straight to a worker, no fallback + self.assertEqual(data, self.src.render(*args)) + finally: + src.pool.close() + + def test_prerender_keeps_to_a_tile_budget_by_zoom(self): + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + full = src.prerender(max_z=6, layers=("relief",)) + src2 = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + seen = [] + n = src2.prerender(max_z=6, layers=("relief",), max_tiles=full - 1, progress=lambda d, t: seen.append(t)) + self.assertLess(n, full, "the deepest zoom that would overrun the budget is left out") + self.assertLessEqual(seen[-1], full - 1) + self.assertTrue(src2.cache_path("relief", 5, *tile_at(self.lat, self.lon, 5)).exists(), "coarse zooms first") + + def test_no_water_walls_where_a_lake_sits_on_a_cliff(self): + import refine as RF + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + tmp = Path(tempfile.mkdtemp()) + dst = RF.results_dir(tmp, self.src.fingerprint) / d.name + shutil.copytree(d, dst) + a = dict(np.load(dst / "cells.npz")) + halo, xyz = a["halo"].astype(bool), a["g_xyz"] + land = ~halo & ~a["ocean"].astype(bool) & ~a["lake"].astype(bool) + mid = latlon_to_xyz(self.lat, self.lon).reshape(3) # well inside the area: by its centre + c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))]) + a["river"][c] = False # (the coarse test cells are all rivers) + nb = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[1:7] # its six neighbours: a cliff below it + a["z_surface_m"][nb] = a["z_surface_m"][c] - 1500.0 + a["lake"][c], a["lake_id"][c] = True, int(a["lake_id"].max()) + 1 + a["z_filled_m"][c] = a["z_surface_m"][c] + 50.0 + a["depression_depth_m"][c] = 50.0 + np.savez(dst / "cells.npz", **a) + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp) + p = xyz[c] + np.linspace(0, 1, 400)[:, None] * (xyz[nb[0]] - xyz[c]) # from the lake over the cliff edge + p /= np.linalg.norm(p, axis=1, keepdims=True) + lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0])) + z = src.z_at(lat, lon, 1.0) + step = np.linalg.norm(p[1] - p[0]) * src.R + self.assertLess(np.abs(np.diff(z)).max(), 50.0 + RF.LAKE_FLOOR_M + 60.0, + "the lake's edge stands at most its depth (+ margin, detail) over the drop, not 1.5 km") + + def test_no_dry_pits_inside_a_lake(self): + import refine as RF + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + tmp = Path(tempfile.mkdtemp()) + dst = RF.results_dir(tmp, self.src.fingerprint) / d.name + shutil.copytree(d, dst) + a = dict(np.load(dst / "cells.npz")) + halo, xyz = a["halo"].astype(bool), a["g_xyz"] + land = ~halo & ~a["ocean"].astype(bool) + mid = latlon_to_xyz(self.lat, self.lon).reshape(3) # well inside the area: by its centre + c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))]) + lake = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[:7] # a cell and its ring: one lake + bed = float(a["z_surface_m"][c]) + a["z_surface_m"][lake] = bed + a["lake"][lake], a["river"][lake] = True, False + a["lake_id"][lake] = int(a["lake_id"].max()) + 1 + a["z_filled_m"][lake] = bed + 100.0 + a["depression_depth_m"][lake] = 100.0 + np.savez(dst / "cells.npz", **a) + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp) + rng = np.random.default_rng(3) + sp = src.regions.spacing_km / src.R + p = xyz[c] + rng.normal(size=(400, 3)) * sp * 0.12 # well inside the lake (its middle + p /= np.linalg.norm(p, axis=1, keepdims=True) # cell and ring: 90 km cells here) + lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0])) + z, water = src.surface_at(lat, lon, 1.0) + below = z < bed + 100.0 - 1e-6 + self.assertFalse((below & ~water).any(), f"{(below & ~water).sum()} dry pits below the lake's surface") + np.testing.assert_allclose(z[water], bed + 100.0, atol=1e-6) # (detail may rise as islands) + + def test_no_dry_pits_where_a_lake_bed_deepens_between_its_cells(self): + import refine as RF + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + tmp = Path(tempfile.mkdtemp()) + dst = RF.results_dir(tmp, self.src.fingerprint) / d.name + shutil.copytree(d, dst) + a = dict(np.load(dst / "cells.npz")) + halo, xyz = a["halo"].astype(bool), a["g_xyz"] + land = ~halo & ~a["ocean"].astype(bool) + mid = latlon_to_xyz(self.lat, self.lon).reshape(3) # well inside the area: by its centre + c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))]) + lake = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[:19] # a cell and two rings: one lake + bed = float(a["z_surface_m"][c]) + a["z_surface_m"][lake] = bed + a["z_surface_m"][c] = bed - 200.0 # its middle much deeper + a["lake"][lake], a["river"][lake] = True, False + a["lake_id"][lake] = int(a["lake_id"].max()) + 1 + a["z_filled_m"][lake] = bed + 100.0 + a["depression_depth_m"][lake] = 100.0 + a["depression_depth_m"][c] = 300.0 + isle = lake[1] # and an island beside it + a["lake"][isle], a["z_surface_m"][isle], a["z_filled_m"][isle] = False, bed + 150.0, bed + 150.0 + np.savez(dst / "cells.npz", **a) + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp) + p = np.concatenate([xyz[c] + np.linspace(0.2, 0.9, 100)[:, None] * (xyz[n] - xyz[c]) for n in lake[2:7]]) + p /= np.linalg.norm(p, axis=1, keepdims=True) # from the deep middle to the ring + lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0])) + z, water = src.surface_at(lat, lon, 1.0) + below = z < bed + 100.0 - 1e-6 + self.assertFalse((below & ~water).any(), f"{(below & ~water).sum()} dry pits below the lake's surface") + + def test_ground_beside_a_lake_below_its_surface_is_under_water(self): + import refine as RF + (d,) = RF.built_areas(self.root / "out" / "r2" / "regions", self.src.fingerprint) + tmp = Path(tempfile.mkdtemp()) + dst = RF.results_dir(tmp, self.src.fingerprint) / d.name + shutil.copytree(d, dst) + a = dict(np.load(dst / "cells.npz")) + halo, xyz = a["halo"].astype(bool), a["g_xyz"] + land = ~halo & ~a["ocean"].astype(bool) + mid = latlon_to_xyz(self.lat, self.lon).reshape(3) # well inside the area: by its centre + c = int(np.where(land)[0][np.argmin(np.linalg.norm(xyz[land] - mid, axis=1))]) + ring = np.argsort(np.linalg.norm(xyz - xyz[c], axis=1))[:7] + lake, shore = ring[:6], ring[6] # a lake and a dry cell beside it + bed = float(a["z_surface_m"][c]) + a["z_surface_m"][ring] = bed + a["lake"][lake], a["river"][ring] = True, False + a["lake"][shore] = False + a["lake_id"][lake] = int(a["lake_id"].max()) + 1 + a["z_filled_m"][lake] = bed + 100.0 + a["depression_depth_m"][lake] = 100.0 + a["z_surface_m"][shore] = a["z_filled_m"][shore] = bed + 60.0 # 40 m below the lake's surface + np.savez(dst / "cells.npz", **a) + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), regions_dir=tmp) + p = xyz[c] + np.linspace(0.55, 0.65, 50)[:, None] * (xyz[shore] - xyz[c]) # just past the cells' border + p /= np.linalg.norm(p, axis=1, keepdims=True) + lat, lon = np.degrees(np.arcsin(p[:, 2])), np.degrees(np.arctan2(p[:, 1], p[:, 0])) + z, water = src.surface_at(lat, lon, 1.0) + below = z < bed + 100.0 - 1e-6 + self.assertFalse((below & ~water).any(), f"{(below & ~water).sum()} dry pits below the lake's surface") + + def test_tiles_away_from_regions_share_the_world_cache(self): + far, near = ("relief", 8, *tile_at(self.lat + 20, self.lon, 8)), ("relief", 8, *tile_at(self.lat, self.lon, 8)) + self.assertIn(self.plain.tag, self.src.cache_path(*far).parts, "a region change keeps the tiles away from it") + self.assertIn(self.src.tag, self.src.cache_path(*near).parts) + + def test_a_region_set_lists_the_built_areas_once(self): + import refine as RF + real, calls = RF.built_areas, [] + with mock.patch.object(RF, "built_areas", side_effect=lambda *a: calls.append(1) or real(*a)): + rs = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R) + self.assertEqual(len(calls), 1, "names and fingerprint from one listing: a build finishing between is not mixed") + self.assertEqual(rs.fingerprint, RF.regions_fingerprint(self.src.regions_dir, self.src.fingerprint)) + + def test_cached_region_set_equals_the_uncached_one(self): + import refine as RF + fresh = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R, use_cache=False) + cached = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R) + self.assertIsNotNone(cached.cache) + self.assertTrue((cached.cache / "done").exists()) + self.assertEqual(set(cached.arrays), set(fresh.arrays)) + for k in fresh.arrays: + np.testing.assert_array_equal(np.asarray(cached.arrays[k]), fresh.arrays[k], err_msg=k) + for k in ("z_env", "lake_level", "lake_floor"): + np.testing.assert_array_equal(getattr(cached, k), getattr(fresh, k), err_msg=k) + self.assertEqual(cached.spacing_km, fresh.spacing_km) + self.assertEqual(cached.area_names, fresh.area_names) + import rivers as RV + for k in RV.STATE: + np.testing.assert_array_equal(getattr(cached.net, k), getattr(fresh.net, k), err_msg=k) + + def test_low_memory_region_set_spills_to_disk_with_equal_arrays(self): + import refine as RF + dirs = RF.built_areas(self.src.regions_dir, self.src.fingerprint) + want, wmeta = RF._compute_region_set(dirs, self.src.w.legends, self.src.R) # oracle: the in-RAM build + with tempfile.TemporaryDirectory() as spill: + got, gmeta = RF._compute_region_set(dirs, self.src.w.legends, self.src.R, spill=Path(spill)) + self.assertEqual(set(got), set(want)) + self.assertEqual(gmeta, wmeta) + for k in want: + np.testing.assert_array_equal(np.asarray(got[k]), want[k], err_msg=k) + big = [k for k in got if k.startswith("a.")] + self.assertTrue(big and all(isinstance(got[k], np.memmap) for k in big), "fields kept on disk") + del got + + def test_low_memory_region_set_leaves_no_spill_folder(self): + import refine as RF + with tempfile.TemporaryDirectory() as d, mock.patch.dict(os.environ, {"WORLDGEN_LOW_MEMORY": "1"}): + regions = Path(d) / "regions" + shutil.copytree(self.src.regions_dir, regions, symlinks=True) + for p in regions.glob("*/serve-*"): + shutil.rmtree(p) + rs = RF.RegionSet(regions, self.src.fingerprint, self.src.w.legends, self.src.R) + self.assertTrue((rs.cache / "done").exists()) + self.assertEqual(list(regions.rglob(".spill-*")), []) + + def test_cached_tiles_and_meshes_are_byte_identical(self): + import refine as RF + fresh = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + fresh.regions = RF.RegionSet(fresh.regions_dir, fresh.fingerprint, fresh.w.legends, fresh.R, use_cache=False) + fresh.tag = fresh.base_tag + fresh._region_tag(fresh.regions) + for z in (6, 8): + x, y = tile_at(self.lat, self.lon, z) + self.assertEqual(self.src.render("relief", z, x, y), fresh.render("relief", z, x, y), f"relief z{z}") + np.testing.assert_array_equal(self.src.mesh(z, x, y), fresh.mesh(z, x, y)) + + def test_trees_are_not_built_at_load(self): + import refine as RF + rs = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R) + self.assertIsNone(rs._tree) + self.assertIsNone(rs._halo_tree) + self.assertEqual(len(rs.area_trees._built), 0) + rs.warm() + self.assertIsNotNone(rs._tree) + self.assertIsNotNone(rs._halo_tree) + self.assertIsNotNone(rs.net._tree) + + def test_start_workers_warms_region_trees_before_forking(self): + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp())) + order = [] + with mock.patch.object(src.regions, "warm", side_effect=lambda: order.append("warm")), \ + mock.patch.object(T, "RenderPool", side_effect=lambda s, n: order.append("fork")): + src.start_workers(2) + self.assertEqual(order, ["warm", "fork"]) + + def test_rebuilt_regions_get_a_new_cache_and_the_old_is_pruned(self): + import refine as RF + own = Path(tempfile.mkdtemp()) / "regions" + shutil.copytree(self.src.regions_dir, own, symlinks=True) # the class's areas keep their mtimes + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), + regions_dir=own) + old = src.regions.cache + area = RF.built_areas(src.regions_dir, src.fingerprint)[0] / "cells.npz" + os.utime(area) # a rebuilt area: new mtime → new regions fingerprint + src.reload_regions(cleanup=True) + self.assertNotEqual(src.regions.cache, old) + self.assertTrue(src.regions.cache.exists()) + self.assertFalse(old.exists()) + self.assertTrue(self.src.regions.cache.exists(), "the shared fixture's cache is untouched") + + def test_build_areas_keeps_serve_cache_folders(self): + import refine as RF + import servecache as SC + base = RF.results_dir(self.src.regions_dir, self.src.fingerprint) + marker = SC.cache_dir(base, "somefp") + marker.mkdir(parents=True, exist_ok=True) + try: + RF.build_areas(self.root, 2, self.root / "places" / "regions.json", log=lambda s: None) + self.assertTrue(marker.exists(), "serve caches are pruned by the server, not by builds") + finally: + shutil.rmtree(marker, ignore_errors=True) + + def test_region_cache_follows_the_code_that_computes_it(self): + import refine as RF + code = Path(tempfile.mkdtemp()) / "rivers.py" + code.write_text("A = 1\n") + with mock.patch.object(RF, "CODE_FILES", (code,)): + a = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R).cache + code.write_text("A = 2\n") # e.g. a river constant changed: the stored net is stale + b = RF.RegionSet(self.src.regions_dir, self.src.fingerprint, self.src.w.legends, self.src.R).cache + self.assertNotEqual(a, b) + self.assertTrue((b / "done").exists()) + + def test_an_empty_region_set_prunes_the_last_cache(self): + import refine as RF + regions = Path(tempfile.mkdtemp()) / "regions" + shutil.copytree(self.src.regions_dir, regions) # (the shared fixture stays as it is) + src = T.TileSource(serve.World(self.root, 2), seed=self.src.seed, cache_dir=Path(tempfile.mkdtemp()), + regions_dir=regions) + old = src.regions.cache + self.assertTrue((old / "done").exists()) + for d in RF.built_areas(regions, src.fingerprint): # the last region deleted + shutil.rmtree(d) + src.reload_regions(cleanup=True) + self.assertTrue(src.regions.empty) + self.assertFalse(old.exists()) + + +class ProfileSurfaceTest(unittest.TestCase): + def test_profile_water_matches_the_tiles(self): + server = serve.make_server(built_world(), 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json") + with Running(server) as base: + prof = json.loads(request(base + "/api/profile?path=10,-85;10,-55&n=64")[1]) # across the west coast + S = prof["samples"] + step = S[1]["d_km"] - S[0]["d_km"] + z, water = server.tiles.surface_at([x["lat"] for x in S], [x["lon"] for x in S], step / 2) + np.testing.assert_allclose([x["z_m"] for x in S], z, atol=0.1) + self.assertEqual([x["water"] for x in S], [bool(w) for w in water]) + self.assertTrue(any(water) and not all(water)) + + +class TileEndpointTest(unittest.TestCase): + def test_endpoints(self): + v = T.VERSION + with Running(serve.make_server(built_world(), 2, 0, pins_path=Path(tempfile.mkdtemp()) / "p.json")) as base: + meta = json.loads(request(base + "/api/meta")[1]) + tag = meta["tiles"]["url"].split("/")[2] # v1-<build fingerprint> + s1, b1 = request(base + f"/tiles/{tag}/relief/5/40/10.jpg") + s2, _ = request(base + f"/tiles/{tag}/height/5/40/10.png") + s3, b3 = request(base + f"/tiles/{tag}/mesh/5/40/10.bin") + s4 = request(base + f"/tiles/{tag}/mesh/5/40/10.jpg")[0] + bad = [request(base + p)[0] for p in (f"/tiles/{tag}/nope/5/1/1.jpg", f"/tiles/{tag}/relief/5/99/1.jpg", + f"/tiles/{tag}/relief/17/0/0.jpg", f"/tiles/v{v}-0000000000/relief/5/1/1.jpg", f"/tiles/{tag}/relief/5/1/1.png", + f"/tiles/{tag}/height/5/1/1.jpg", f"/tiles/v{v}/relief/5/1/1.jpg")] + cell = json.loads(request(base + "/api/cell?lat=10&lon=20")[1]) + prof = json.loads(request(base + "/api/profile?path=10,10;10,30&n=32")[1]) + self.assertRegex(meta["tiles"]["url"], rf"^/tiles/v{v}\.{T.DRAW}-[0-9a-f]{{10}}/{{layer}}/{{z}}/{{x}}/{{y}}\.jpg$") + self.assertEqual((meta["tiles"]["min_z"], meta["tiles"]["max_z"]), (5, 16)) + self.assertEqual((s1, Image.open(io.BytesIO(b1)).size), (200, (256, 256))) + self.assertEqual(s2, 200) + self.assertEqual(bad, [404] * 7) + self.assertRegex(meta["tiles"]["mesh_url"], rf"^/tiles/v{v}\.{T.DRAW}-[0-9a-f]{{10}}/mesh/{{z}}/{{x}}/{{y}}\.bin$") + self.assertEqual(T.MESH_N, 129, "3D heights: 129 × 129 per tile (author 2026-09-25)") + self.assertEqual((s3, len(b3), s4), (200, T.MESH_BYTES, 404)) + self.assertEqual((T.VERSION, T.DRAW), (6, 5)) + f = {x["key"]: x for x in cell["fields"]} + self.assertEqual(f["z_detail_m"]["label"], "Local elevation (procedural)") + self.assertIn("spans", f["z_detail_m"]["alt"]) + self.assertTrue(all("z_cell_m" in s for s in prof["samples"])) + + +class SharedPoolTest(unittest.TestCase): + def test_two_sources_share_one_pool(self): + a_root, b_root = built_world(), built_world() + a = T.TileSource(serve.World(a_root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp()), name="a") + b = T.TileSource(serve.World(b_root, 2), seed=8, cache_dir=Path(tempfile.mkdtemp()), name="b") + T.join([a, b]) + self.assertIs(a.peers, b.peers) + a.start_workers(1) + try: + self.assertIs(a.pool, b.pool) + args = (6, *tile_at(10.0, 20.0, 6)) + self.assertEqual(b.pool.render("b", "relief", *args, b.regions.fingerprint), b.render("relief", *args)) + self.assertEqual(a.pool.render("a", "relief", *args, a.regions.fingerprint), a.render("relief", *args)) + self.assertNotEqual(a.render("relief", *args), b.render("relief", *args), "seeds differ: two worlds") + old = a.pool + b.reload_regions() + self.assertIsNot(a.pool, old, "a reload forks fresh workers for every peer") + self.assertIs(a.pool, b.pool) + made = [] + real = T.RenderPool + with mock.patch.object(T, "RenderPool", side_effect=lambda *x: made.append(1) or real(*x)): + old = a.pool + T.reload_all([a, b]) + self.assertEqual(len(made), 1, "every era reloaded, then one fork") + self.assertIsNot(a.pool, old) + self.assertIs(a.pool, b.pool) + finally: + a.pool.close() + + +class _Hangs: + name = "h" + + class regions: + fingerprint = "fp" + + def render(self, layer, z, x, y): + time.sleep(3600) + + +class PoolTimeoutTest(unittest.TestCase): + def test_a_hung_worker_is_stopped_and_the_request_fails(self): + pool = T.RenderPool({"h": _Hangs()}, 1) + try: + with mock.patch.object(T, "RENDER_TIMEOUT_S", 0.5): + t = time.time() + with self.assertRaisesRegex(RuntimeError, "timed out"): + pool.render("h", "relief", 5, 1, 1, "fp") + self.assertLess(time.time() - t, 5) + self.assertEqual(pool.alive, 0) + pool.procs[0].join(timeout=5) + self.assertFalse(pool.procs[0].is_alive(), "the stuck process is gone") + finally: + pool.close() + + +class EraShareTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + import refine as RF + from mapgen import config as C + from mapgen import eras as ER + from tests.test_refine import era_world, square, write_regions + cls.root, cls.far, cls.cut, _ = era_world() + tmp = Path(tempfile.mkdtemp()) + reg = write_regions(tmp / "regions.json", square(*cls.far, 2, "far"), square(*cls.cut, 2, "cut")) + cls.rdir = tmp / "res" + RF.build_areas(cls.root, 2, reg, log=lambda s: None, regions_root=cls.rdir) + seed = int(C.load(cls.root)[0]["build"]["seed"]) # the areas were built for the world's own seed + cls.base = T.TileSource(serve.World(cls.root, 2), seed=seed, cache_dir=tmp / "tb", regions_dir=cls.rdir) + cls.era = T.TileSource(serve.World(cls.root, 2, out=ER.era_dir(cls.root, 2, "after")), seed=seed, + cache_dir=tmp / "te", regions_dir=cls.rdir, name="after", share=cls.base) + T.join([cls.base, cls.era]) + + def test_area_keys_name_the_content(self): + rs_b, rs_e = self.base.regions, self.era.regions + self.assertEqual(sorted(rs_b.area_keys), sorted(rs_b.area_names)) + same = [h for h in rs_e.area_names if rs_e.area_keys[h] == rs_b.area_keys.get(h)] + self.assertEqual(len(same), 1, "the far area is shared, the cut one is the era's own") + + def test_tiles_away_from_the_era_are_the_bases(self): + args = (7, *tile_at(*self.far, 7)) + self.assertTrue(self.era.shared(*args)) + self.assertEqual(self.era.tile("relief", *args), self.base.tile("relief", *args)) + self.assertEqual(self.era.cache_path("relief", *args), self.base.cache_path("relief", *args)) + self.assertFalse(any((self.era.cache_root).rglob("*.jpg")), "nothing saved twice") + + def test_a_shared_tile_keeps_the_render_limit(self): + import threading + args = (8, *tile_at(*self.far, 8)) + self.assertTrue(self.era.shared(*args)) + gate = threading.BoundedSemaphore(1) + gate.acquire() # no slot free + with self.assertRaises(T.Busy): + self.era.tile("seafloor", *args, remember=False, gate=gate) + + def test_tiles_near_the_mask_are_the_eras_own(self): + args = (7, *tile_at(*self.cut, 7)) + self.assertFalse(self.era.shared(*args)) + self.assertNotEqual(self.era.tile("height", *args), self.base.tile("height", *args)) + + def test_the_mask_edge_alone_decides_without_refined_areas(self): + from mapgen import eras as ER + from mapgen.sphere import gc_dist_km + tmp = Path(tempfile.mkdtemp()) + era = T.TileSource(serve.World(self.root, 2, out=ER.era_dir(self.root, 2, "after")), seed=self.base.seed, + cache_dir=tmp / "te", regions_dir=tmp / "none", name="after", share=self.base) + with np.load(ER.era_dir(self.root, 2, "after") / "cells.npz") as a: + xyz, m = np.asarray(a["g_xyz"], np.float64), a["era_mask"].astype(bool) + z, span = 9, 180.0 / 2 ** 9 + near = far = None + for k in range(1, 4000): + lon = self.cut[1] + k * span + x, y = tile_at(self.cut[0], lon, z) + c = latlon_to_xyz(90 - (y + 0.5) * span, -180 + (x + 0.5) * span) + d = gc_dist_km(xyz, c, era.R) + to_mask = float(d[m].min()) + if near is None and not m[int(np.argmin(d))] and to_mask < 2 * era._spacing_km: + near = (x, y) + if to_mask > 2 * era._spacing_km + T.CARRY_KM + 2 * span * np.pi / 180 * era.R: + far = (x, y) + break + self.assertIsNotNone(near) + self.assertIsNotNone(far) + self.assertFalse(era.shared(z, *near), "just outside the mask: the era's own") + self.assertTrue(era.shared(z, *far), "beyond the mask's reach: the base's") + + def test_the_base_shares_nothing(self): + self.assertFalse(self.base.shared(7, *tile_at(*self.far, 7))) + + +class SeaFloorTileTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.root = built_world() + cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + + def test_depth_shading_darkens_with_depth(self): + z = np.tile(np.linspace(-6000.0, -10.0, 64), (64, 1)) + rgb = T.seafloor_rgb(z, np.full(z.shape, 0.7), np.full(z.shape, 3), np.zeros(z.shape, bool)).astype(float) + self.assertLess(rgb[:, :8].mean(), rgb[:, -8:].mean(), "deep is darker") + + def test_type_tints_the_sea_and_land_keeps_relief(self): + z = np.full((8, 8), -3000.0) + land = np.zeros((8, 8), bool) + land[:, :4] = True + a = T.seafloor_rgb(np.where(land, 500.0, z), np.full(z.shape, 0.7), np.full(z.shape, 3), land) + b = T.seafloor_rgb(np.where(land, 500.0, z), np.full(z.shape, 0.7), np.full(z.shape, 7), land) + self.assertFalse(np.array_equal(a[:, 4:], b[:, 4:]), "clay vs vent field") + np.testing.assert_array_equal(a[:, :4], b[:, :4]) + + def test_seafloor_tile(self): + self.assertIn("seafloor", T.LAYERS) + args = (5, *tile_at(-30.0, -150.0, 5)) + im = Image.open(io.BytesIO(self.src.tile("seafloor", *args))) + self.assertEqual(im.size, (256, 256)) + f = self.src.fields(*args, 0, ("cat",)) + self.assertIn("seabed_type", f) + + def test_world_holds_the_new_fields(self): + for k in ("seabed_type", "bottom_temp_c", "pressure_bar", "fire_reactivity", "o2_fraction"): + self.assertIn(k, self.src.w.arrays, k) + + def test_new_data_layers_render(self): + args = (6, *tile_at(10.0, 20.0, 6)) + for layer in ("seabed", "minerals", "bottom_temp", "sediment", "vent_potential", "pressure", "fire"): + self.assertIn(layer, T.LAYERS, layer) + self.assertEqual(Image.open(io.BytesIO(self.src.tile(layer, *args))).size, (256, 256), layer) + + def test_ocean_layers(self): + for layer in ("currents", "sst", "productivity"): + self.assertIn(layer, T.LAYERS, layer) + self.assertEqual(T.CONT["currents"], "current_speed") + + +class LandColourTest(unittest.TestCase): + """Land colour = terrain shading + a vegetation mosaic (woods vs open ground; author 2026-10-03, spike E).""" + @classmethod + def setUpClass(cls): + cls.root = built_world() + cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + a = cls.src.w.arrays + i = int(np.flatnonzero(~np.asarray(a["ocean"]).astype(bool))[0]) + q = np.asarray(a["g_xyz"][i], dtype=float) + cls.land = float(np.degrees(np.arcsin(q[2] / np.linalg.norm(q)))), float(np.degrees(np.arctan2(q[1], q[0]))) + names = list(cls.src.w.legends["holdridge"]) + cls.zone = {n: names.index(n) for n in names} + + def synthetic(self, z, x, y, zone, height=None, water=None): + B = T.LAND_BORDER + f = dict(self.src.fields(z, x, y, B, ("z",), valleys=False)) + n = f["lat"].shape[0] + f["z"] = f["base"] + (0.0 if height is None else height) # height: above the smooth base surface + f["water"] = np.zeros((n, n), bool) if water is None else water + f["lake"] = np.zeros((n, n), np.int64) + f["river_draw"] = np.zeros((n, n), bool) + f["holdridge"] = np.full((n, n), self.zone[zone]) + f["ground"] = np.zeros((n, n), np.int64) + return f + + def wood_share(self, zone, z=9): + x0, y0 = tile_at(*self.land, z) + return float(np.mean([self.src._veg(self.synthetic(z, x0 + i, y0 + j, zone), T.LAND_BORDER)["wood"].mean() + for i in range(3) for j in range(3)])) + + def test_wide_border_keeps_the_interior(self): # a wider border must not change what the tile itself holds + z, x, y = 10, *tile_at(*self.land, 10) + a = self.src.fields(z, x, y, 1, ("z", "water", "cat")) + b = self.src.fields(z, x, y, 9, ("z", "water", "cat")) + for k in ("z", "water", "holdridge", "river_draw", "lake"): + np.testing.assert_array_equal(a[k], b[k][8:-8, 8:-8], k) + + def test_border_is_the_neighbours_own_pixels(self): # a wide border holds what the next tile draws there + B = T.LAND_BORDER + z, x, y = 10, *tile_at(*self.land, 10) + f = self.src.fields(z, x, y, B, ("z", "water", "cat")) + west = self.src.fields(z, x - 1, y, 0, ("z", "water", "cat")) + np.testing.assert_allclose(f["z"][B:-B, :B], west["z"][:, -B:], atol=1e-6) + np.testing.assert_array_equal(f["holdridge"][B:-B, :B], west["holdridge"][:, -B:]) + + def test_colour_does_not_depend_on_the_border(self): # seamless: nothing reaches past LAND_BORDER + B = T.LAND_BORDER + z, x, y = 10, *tile_at(*self.land, 10) + a = self.src._veg(self.src.fields(z, x, y, B, ("z", "water", "cat")), B) + b = self.src._veg(self.src.fields(z, x, y, 2 * B, ("z", "water", "cat")), 2 * B) + for k in ("wood", "bright", "tint"): + np.testing.assert_allclose(a[k], b[k], atol=1e-9, err_msg=k) + + def test_cover_by_zone(self): + c = T.veg_cover(list(self.src.w.legends["holdridge"])) + z = self.zone + self.assertEqual(c[z["polar desert"]], 0.0) + self.assertLess(c[z["tropical desert"]], c[z["tropical thorn woodland"]]) + self.assertLess(c[z["tropical thorn woodland"]], c[z["tropical dry forest"]]) + self.assertLess(c[z["tropical dry forest"]], c[z["tropical rain forest"]]) + self.assertLess(c[z["cool temperate steppe"]], c[z["cool temperate moist forest"]]) + self.assertLess(c[z["subpolar dry tundra"]], c[z["subpolar wet tundra"]]) + + def test_wetter_zones_are_more_wooded(self): + rain, dry, desert = (self.wood_share(n) for n in ("tropical rain forest", "tropical dry forest", "tropical desert")) + self.assertGreater(rain, 0.6) + self.assertLess(desert, 0.12) + self.assertLess(desert, dry) + self.assertLess(dry, rain) + + def test_far_out_the_mosaic_is_the_zone_mean(self): # patches of a few px read as noise: none at z4 + f = self.synthetic(4, *tile_at(*self.land, 4), "tropical dry forest") + c = T.veg_cover(list(self.src.w.legends["holdridge"]))[self.zone["tropical dry forest"]] + np.testing.assert_allclose(self.src._veg(f, T.LAND_BORDER)["wood"], c, atol=1e-12) + + def test_salt_flats_bare(self): + f = self.synthetic(9, *tile_at(*self.land, 9), "tropical rain forest") + f["ground"][:] = list(self.src.w.legends["ground"]).index("salt flat") + self.assertEqual(float(self.src._veg(f, T.LAND_BORDER)["wood"].max()), 0.0) + + def test_woods_in_hollows_not_on_crests(self): # the same pixels, flat vs sine ridges (the patches cancel) + z, (x, y) = 11, tile_at(*self.land, 11) + n, B = T.TILE + 2 * T.LAND_BORDER, T.LAND_BORDER + s = np.sin(2 * np.pi * np.arange(n) / 48.0) + zone = "warm temperate dry forest" # cover ½: room to gain and to lose + d = np.hstack([self.src._veg(self.synthetic(z, x + i, y, zone, height=np.tile(120.0 * s, (n, 1))), B)["wood"] + - self.src._veg(self.synthetic(z, x + i, y, zone), B)["wood"] for i in range(3)]) + s = np.tile(s[B:-B], 3) + self.assertLess(d[:, s > 0.97].mean(), -0.03, "fewer woods on crests") # flat tops and bottoms: no slope term + self.assertGreater(d[:, s < -0.97].mean(), 0.03, "more in hollows") + + def test_woods_along_water(self): + z, x, y = 12, *tile_at(*self.land, 12) + n, B = T.TILE + 2 * T.LAND_BORDER, T.LAND_BORDER + wet = np.zeros((n, n), bool) + wet[:, n // 2] = True + zone = "warm temperate dry forest" + d = self.src._veg(self.synthetic(z, x, y, zone, water=wet), B)["wood"] - \ + self.src._veg(self.synthetic(z, x, y, zone), B)["wood"] + mid = T.TILE // 2 + self.assertGreater(d[:, np.r_[mid - 3:mid, mid + 1:mid + 4]].mean(), 0.15, "woods along the water") + self.assertAlmostEqual(float(np.abs(d[:, :mid - 20]).max()), 0.0, 9, msg="none far away") + + def test_relief_tile_renders_and_repeats(self): + z, x, y = 9, *tile_at(*self.land, 9) + a, b = self.src.rgb("relief", z, x, y), self.src.rgb("relief", z, x, y) + self.assertEqual(a.shape, (T.TILE, T.TILE, 3)) + np.testing.assert_array_equal(a, b) + + +class GullyTest(unittest.TestCase): + """Relief detail with erosion gullies on slopes (author 2026-10-03: spike C; full rebuild accepted).""" + @classmethod + def setUpClass(cls): + cls.root = built_world() + cls.src = T.TileSource(serve.World(cls.root, 2), seed=7, cache_dir=Path(tempfile.mkdtemp())) + + def line(self, lat0, lon0, bearing, km, n=400): # points along a straight east (90°) or north (0°) line + d = np.linspace(0.0, km, n) / self.src.R + if bearing == 90: + return np.full(n, lat0), lon0 + np.degrees(d) / np.cos(np.radians(lat0)) + return lat0 + np.degrees(d), np.full(n, lon0) + + def detail(self, lat, lon, px_km, slope): # on a plane rising east at `slope` m/m, landform amplitude 450 m + xyz = latlon_to_xyz(lat, lon) + base = lambda la, lo: np.radians(lo) * self.src.R * 1000.0 * np.cos(np.radians(10.0)) * slope + n = len(lat) + return self.src._detail(xyz, px_km, np.full(n, 450.0), np.full(n, 0.5), lat, lon, base=base) + + def test_gullies_run_downhill(self): # the first (4 km) octave alone (px 0.6 km): stripes along the slope + px = 0.6 # (finer octaves follow the gully walls: that is the branching) + down = self.detail(*self.line(10.0, 20.0, 90, 40.0), px, 0.4) + across = self.detail(*self.line(10.0, 20.0, 0, 40.0), px, 0.4) + self.assertGreater(np.std(np.diff(across)), 2.0 * np.std(np.diff(down))) + + def test_flat_ground_keeps_its_detail(self): # no slope: (almost) the plain fractal detail + lat, lon = self.line(10.0, 20.0, 0, 3.0) + xyz = latlon_to_xyz(lat, lon) + n = len(lat) + flat = self.src._detail(xyz, 0.03, np.full(n, 25.0), np.full(n, 0.0), lat, lon, base=lambda la, lo: 0.0 * la) + plain = 25.0 * self.src._octaves(xyz, 0.03, T.TOP_KM, T.MIN_KM, T.GAIN, self.src.seed + 7001, np.zeros(n)) + self.assertLess(np.abs(flat - plain).max(), 0.1 * 25.0) # ≤ 10 % of the landform amplitude + + def test_far_out_no_gullies(self): # gullies need ≥ 4 px per wavelength: none at z6 + lat, lon = self.line(10.0, 20.0, 0, 300.0) + xyz, n = latlon_to_xyz(lat, lon), len(lat) + px = 2.0 + d = self.detail(lat, lon, px, 0.4) + plain = 450.0 * self.src._octaves(xyz, px, T.TOP_KM, T.MIN_KM, T.GAIN, self.src.seed + 7001, np.full(n, 0.5)) + np.testing.assert_array_equal(d, plain) + + def test_heights_agree_everywhere(self): # tiles, profile/3D points and the refinement start share one detail + z, x, y = 11, *tile_at(10.0, 20.0, 11) + f = self.src.fields(z, x, y, 0, ("z",), valleys=False) + lat, lon = f["lat"][::37, ::41].ravel(), f["lon"][::37, ::41].ravel() + ra = self.src.relief_at(lat, lon, f["px_km"]) + xyz = latlon_to_xyz(lat, lon) + amp, ridge, _ = self.src._amp(xyz) + np.testing.assert_allclose(ra - self.src.bilinear(self.src.raster("elevation"), lat, lon), + self.src._detail(xyz, f["px_km"], amp, ridge, lat, lon), atol=1e-9) + + def test_compiled_stripes_match_numpy(self): + if T._phacelle_jit is None: + self.skipTest("numba not installed") + rng = np.random.default_rng(3) + p = rng.normal(size=(5000, 3)) + p /= np.linalg.norm(p, axis=1, keepdims=True) + e, n = T.east_north(p) + ang = rng.uniform(0, 2 * np.pi, 5000)[:, None] + a = np.cos(ang) * e + np.sin(ang) * n + jit = self.src._phacelle(p, a, 0.5, 123) + with mock.patch.object(T, "_phacelle_jit", None): + ref = self.src._phacelle(p, a, 0.5, 123) + np.testing.assert_allclose(jit[0], ref[0], atol=1e-9) + np.testing.assert_allclose(jit[1], ref[1], atol=1e-9) |