worldgen: initial public history

1 files changed, 303 insertions, 0 deletions

diff --git a/tests/test_eras.py b/tests/test_eras.py
new file mode 100644
index 0000000..96b7c82
--- /dev/null
+++ b/tests/test_eras.py
@@ -0,0 +1,303 @@
+import shutil
+import importlib.util
+import tempfile
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+from mapgen import eras as ER, events as EV, pipeline as P
+from mapgen.graph import components, distance_to
+from mapgen.sphere import gc_dist_km, latlon_to_xyz, xyz_to_latlon
+from tests.test_render import small_world
+
+R = 12742.0
+ERAS = """
+[[event]]
+name = "cut"
+kind = "disintegrate"
+center = {cut}
+radius_km = 3000.0
+depth_m = 3000.0
+[[event]]
+name = "aura"
+kind = "zone"
+shape = "landmass"
+seed = {seed}
+reach_km = [500.0, 1000.0]
+edge_km = 5.0
+fields = {{ gravity_g = 0.35, pressure_bar = 2.0, o2_fraction = 0.35, fire_reactivity = {fire} }}
+[[event]]
+name = "rise"
+kind = "volcano"
+center = {vol}
+radius_km = 900.0
+peak_m = 1500.0
+peak_mode = "absolute"
+[[event]]
+name = "ghost"
+kind = "zone"
+shape = "landmass"
+seed = {seed}
+reach_km = [100.0, 200.0]
+edge_km = 5.0
+fields = {{ fire_reactivity = 0.2 }}
+[eras]
+order = ["before", "glow", "after", "risen", "haunt"]
+default = "after"
+[eras.before]
+label = "Before"
+events = []
+[eras.glow]
+label = "Glow"
+events = ["aura"]
+[eras.after]
+label = "After"
+events = ["cut"]
+[eras.risen]
+label = "Risen"
+events = ["rise"]
+years = 8000
+[eras.haunt]
+label = "Haunt"
+events = ["ghost"]
+"""
+
+
+def cli():
+    spec = importlib.util.spec_from_file_location("mapgen_cli", Path(ER.__file__).parents[1] / "mapgen.py")
+    m = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(m)                          # the script (mapgen/ is the package)
+    return m
+
+
+def latlon(p):
+    lat, lon = xyz_to_latlon(np.asarray(p, dtype=np.float64))
+    return [round(float(lat), 3), round(float(lon), 3)]
+
+
+class ErasTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.root = Path(tempfile.mkdtemp())
+        small_world(cls.root)
+        cls.base = P.build(cls.root, 2, log=lambda m: None)
+        g, d = cls.base.grid, cls.base.data
+        land = ~np.asarray(d["ocean"])
+        lab = components(g, land)
+        big = lab == np.bincount(lab[land]).argmax()
+        mid = g.xyz[big].mean(axis=0)
+        cls.cut = latlon(g.xyz[big][np.argmax(g.xyz[big] @ (mid / np.linalg.norm(mid)))])
+        far = gc_dist_km(g.xyz, latlon_to_xyz(*cls.cut), R)
+        cls.seed = latlon(g.xyz[np.argmax(np.where(big, far, -1.0))])     # stays land after the cut
+        cls.vol_i = int(np.argmax(distance_to(g, land)))                   # the loneliest sea
+        cls.vol = latlon(g.xyz[cls.vol_i])
+        cls.g0 = float(cls.base.cfg["planet"]["gravity_g"])
+        cls.write_eras()
+        for n in ("glow", "after", "risen", "haunt"):
+            ER.build_era(cls.root, 2, n, log=lambda m: None, base=cls.base)
+
+    def test_low_memory_era_writes_identical_files(self):
+        # oracle: the normal-mode era built in setUpClass
+        tmp = Path(tempfile.mkdtemp())
+        try:
+            shutil.copytree(self.root, tmp, dirs_exist_ok=True)
+            shutil.rmtree(ER.era_dir(tmp, 2, "glow"))
+            ER.build_era(tmp, 2, "glow", log=lambda m: None, low_memory=True)
+            a, b = ER.era_dir(self.root, 2, "glow"), ER.era_dir(tmp, 2, "glow")
+            fa = sorted(p.relative_to(a) for p in a.rglob("*") if p.is_file())
+            self.assertEqual(fa, sorted(p.relative_to(b) for p in b.rglob("*") if p.is_file()))
+            for f in fa:
+                self.assertEqual((a / f).read_bytes(), (b / f).read_bytes(), str(f))
+        finally:
+            shutil.rmtree(tmp)
+
+    @classmethod
+    def write_eras(cls, fire=0.5):
+        (cls.root / "config" / "eras.toml").write_text(ERAS.format(cut=cls.cut, seed=cls.seed, vol=cls.vol, fire=fire))
+
+    def load(self, name=None):
+        d = self.root / "out" / "r2" if name is None else ER.era_dir(self.root, 2, name)
+        with np.load(d / "cells.npz") as z:
+            return {k: z[k] for k in z.files}
+
+    def test_outside_the_mask_is_the_base(self):
+        b = self.load()
+        for n in ("glow", "after", "risen"):
+            e = self.load(n)
+            m = e["era_mask"]
+            self.assertTrue(m.any() and not m.all(), n)
+            for k, v in b.items():
+                if v.shape[:1] == m.shape:
+                    np.testing.assert_array_equal(e[k][~m], v[~m], err_msg=f"{n}: {k}")
+
+    def test_deposits_are_the_base_geology(self):
+        b = self.load()                                      # an event moves ground; it neither makes nor unmakes ore
+        for n in ("glow", "after", "risen"):
+            e = self.load(n)
+            for k in ("deposits", "deposit_main", "iron_potential"):
+                np.testing.assert_array_equal(e[k], b[k], err_msg=f"{n}: {k}")
+
+    def test_disintegration_only_lowers_and_reaches_depth(self):
+        b, e = self.load(), self.load("after")
+        zb, ze = b["elevation_eroded_m"].astype(np.float64), e["elevation_eroded_m"].astype(np.float64)
+        self.assertTrue(np.all(ze <= zb + 1e-3))
+        d = gc_dist_km(b["g_xyz"], latlon_to_xyz(*self.cut), R)
+        _, z_ref = EV.disintegrate(b["g_xyz"], zb, ~b.get("open_water", b["ocean"]), {"center": self.cut, "radius_km": 3000.0,
+                                                                "depth_m": 3000.0}, R)
+        inner = d < 1500.0
+        self.assertTrue(inner.any())
+        self.assertTrue(np.all(ze[inner] <= z_ref - 3000.0 * np.sqrt(0.75) + 1e-3))
+        self.assertTrue((e["ocean"] | e["lake"])[inner].all(), "the bowl fills with water")
+
+    def test_zone_fields_hold_inside_the_landmass(self):
+        b, e = self.load(), self.load("glow")
+        land, w = e["zone_aura_land"], e["zone_aura"]
+        self.assertTrue(land.any())
+        self.assertTrue(np.all(w[land] == 1.0))
+        np.testing.assert_allclose(e["gravity_g"][land], 0.35, rtol=1e-6)
+        np.testing.assert_allclose(e["fire_reactivity"][land], 0.5)
+        np.testing.assert_allclose(e["po2_bar"], e["o2_fraction"] * e["pressure_bar"], rtol=1e-5)
+        np.testing.assert_array_equal(e["gravity_g"][w == 0], b["gravity_g"][w == 0])
+
+    def test_zone_follows_the_landmass_before_the_era(self):
+        b, e = self.load(), self.load("after")
+        d = gc_dist_km(b["g_xyz"], latlon_to_xyz(*self.cut), R)
+        lost = ~b["open_water"] & e["open_water"] & (d < 1500.0)
+        self.assertTrue(lost.any())
+        self.assertTrue(e["zone_aura_land"][lost].all(), "the old coast, not the bitten one")
+        np.testing.assert_allclose(e["gravity_g"][lost], 0.35, rtol=1e-6)
+
+    def test_plants_in_a_zone(self):
+        from mapgen import environment as EN
+        e = self.load("glow")
+        land = e["zone_aura_land"] & ~e["ocean"]
+        zone, _ = EN.holdridge(e["biotemp"], e["P_ann"] * np.sqrt(2.0), e["T_min"])
+        np.testing.assert_array_equal(e["holdridge"][land], zone[land])
+        np.testing.assert_allclose(e["plant_height_x"], self.g0 / e["gravity_g"], rtol=1e-5)
+
+    def test_era_years_set_its_erosion(self):
+        e = self.load("risen")
+        self.assertGreater(float(e["elevation_eroded_m"][self.vol_i]), 1400.0, "8,000 years barely wear a cone")
+        self.assertNotEqual(ER.fingerprint("k", [("a", [], 8000)]), ER.fingerprint("k", [("a", [], 9000)]))
+
+    def test_steps_list_each_eras_own_events_and_years(self):
+        from mapgen import config as C
+        st = ER.steps(C.load(self.root)[1], "risen")
+        self.assertEqual([(n, [e["name"] for e in ev], y) for n, ev, y in st],
+                         [("before", [], None), ("glow", ["aura"], None), ("after", ["cut"], None),
+                          ("risen", ["rise"], 8000)])
+
+    def test_a_later_zone_follows_the_land_of_its_own_era(self):
+        b, e = self.load(), self.load("haunt")
+        d = gc_dist_km(b["g_xyz"], latlon_to_xyz(*self.cut), R)
+        lost = ~b["open_water"] & self.load("after")["open_water"] & (d < 1500.0)
+        self.assertTrue(lost.any())
+        self.assertFalse(e["zone_ghost_land"][lost].any(), "ghost came after the cut: the bitten coast")
+        self.assertTrue(e["zone_aura_land"][lost].all(), "aura still follows the land before the cut")
+
+    def test_a_new_label_needs_no_rebuild(self):
+        import json
+        f = ER.era_dir(self.root, 2, "glow") / "cells.npz"
+        before = f.stat().st_mtime_ns
+        p = self.root / "config" / "eras.toml"
+        p.write_text(p.read_text().replace('label = "Glow"', 'label = "The Glow"'))
+        try:
+            logs = []
+            ER.build_era(self.root, 2, "glow", log=logs.append)
+            self.assertIn("era glow: cached", logs)
+            meta = json.loads((ER.era_dir(self.root, 2, "glow") / "cells_meta.json").read_text())
+            self.assertEqual(meta["era"]["label"], "The Glow")
+            self.assertEqual(f.stat().st_mtime_ns, before)
+        finally:
+            self.write_eras()
+            ER.build_era(self.root, 2, "glow", log=lambda m: None)
+
+    def test_a_renamed_era_whose_folder_moved_needs_no_rebuild(self):
+        import json
+        old, new = ER.era_dir(self.root, 2, "glow"), ER.era_dir(self.root, 2, "shine")
+        before = (old / "cells.npz").stat().st_mtime_ns
+        p = self.root / "config" / "eras.toml"
+        p.write_text(p.read_text().replace('"glow"', '"shine"').replace("[eras.glow]", "[eras.shine]"))
+        old.rename(new)
+        try:
+            logs = []
+            ER.build_era(self.root, 2, "shine", log=logs.append)
+            self.assertIn("era shine: cached", logs)
+            self.assertEqual(json.loads((new / "cells_meta.json").read_text())["era"]["name"], "shine")
+            self.assertEqual((new / "cells.npz").stat().st_mtime_ns, before)
+        finally:
+            new.rename(old)
+            self.write_eras()
+            ER.build_era(self.root, 2, "glow", log=lambda m: None)
+
+    def test_one_lake_id_never_names_two_lakes(self):
+        b, e = self.load(), self.load("after")
+        m, ids = e["era_mask"], e["lake_id"]
+        for k in np.unique(ids[m & (ids >= 0)]).tolist():
+            if (ids[~m] == k).any():
+                np.testing.assert_array_equal(ids == k, b["lake_id"] == k, err_msg=f"lake {k}")
+
+    def test_zone_events_never_change_height(self):
+        b, e = self.load(), self.load("glow")
+        for k in ("elevation_m", "elevation_eroded_m", "z_surface_m"):
+            np.testing.assert_array_equal(e[k], b[k], err_msg=k)
+
+    def test_a_volcano_raises_new_land(self):
+        e = self.load("risen")
+        self.assertFalse(e["ocean"][self.vol_i])
+        self.assertGreater(float(e["elevation_eroded_m"][self.vol_i]), 0.0)
+
+    def test_era_is_cached_until_its_events_change(self):
+        logs = []
+        ER.build_era(self.root, 2, "glow", log=logs.append)
+        self.assertIn("era glow: cached", logs)
+        first = self.load("glow")
+        self.write_eras(fire=0.6)
+        try:
+            logs, base_logs = [], []
+            ER.build_era(self.root, 2, "glow", log=logs.append)
+            self.assertNotIn("era glow: cached", logs)
+            P.build(self.root, 2, log=base_logs.append)
+            self.assertTrue(base_logs and all(line.endswith("cached") for line in base_logs),
+                            "tuning an event never rebuilds the base")
+        finally:
+            self.write_eras()
+            ER.build_era(self.root, 2, "glow", log=lambda m: None)
+        again = self.load("glow")                             # repeatable: the same inputs, the same arrays
+        for k in first:
+            np.testing.assert_array_equal(again[k], first[k], err_msg=k)
+
+    def test_build_command_builds_eras_and_removes_stale_ones(self):
+        stale = ER.era_dir(self.root, 2, "gone")
+        stale.mkdir(parents=True, exist_ok=True)
+        self.assertEqual(cli().main(["build", "--res", "2"], root=self.root), 0)
+        self.assertFalse(stale.exists())
+        for n in ("glow", "after", "risen"):
+            self.assertTrue((ER.era_dir(self.root, 2, n) / "cells.npz").exists(), n)
+        self.assertFalse(ER.era_dir(self.root, 2, "before").exists(), "the base era is the base build")
+        self.assertEqual(cli().main(["era", "glow", "--res", "2"], root=self.root), 0)
+        self.assertEqual(cli().main(["era", "nowhen", "--res", "2"], root=self.root), 2)
+
+
+class MergeLakeIdsTest(unittest.TestCase):
+    def test_kept_lakes_keep_ids_changed_and_new_lakes_get_fresh_ones(self):
+        base = np.array([0, 0, 1, 1, -1, 2, -1], np.int32)
+        new = np.array([5, 5, 0, 0, 0, -1, 1], np.int32)   # 5 = base 0 as it was; 0 = base 1 grown; 1 new; 2 gone
+        out = ER.merge_lake_ids(base, new, np.ones(7, bool))
+        np.testing.assert_array_equal(out, [0, 0, 3, 3, 3, -1, 4])
+        self.assertEqual(out.dtype, np.int32)
+
+    def test_outside_the_mask_stays_the_base(self):
+        base = np.array([0, 0, -1, 1])
+        new = np.array([3, 3, 7, 7])
+        out = ER.merge_lake_ids(base, new, np.array([False, False, True, True]))
+        np.testing.assert_array_equal(out, [0, 0, 2, 2])
+
+
+class SameTest(unittest.TestCase):
+    def test_nan_equals_nan_in_float_arrays_only(self):
+        self.assertTrue(ER._same(np.array([np.nan, 1.0]), np.array([np.nan, 1.0])))
+        self.assertFalse(ER._same(np.array([np.nan, 1.0]), np.array([np.nan, 2.0])))
+        self.assertTrue(ER._same(np.array([1, 2]), np.array([1, 2])))
+        self.assertTrue(ER._same(np.array([True]), np.array([True])))