worldhistory: initial public history

1 files changed, 128 insertions, 0 deletions

diff --git a/tests/test_engine.py b/tests/test_engine.py
new file mode 100644
index 0000000..b0178e2
--- /dev/null
+++ b/tests/test_engine.py
@@ -0,0 +1,128 @@
+import os
+import tempfile
+import tomllib
+import unittest
+from pathlib import Path
+
+import numpy as np
+
+from tests.helpers import fields, globe_world, line_world, make_history, make_race
+from worldhistory.config import CONDITIONS, link, load_config
+from worldhistory.engine import Engine
+from worldhistory.nudges import nudge_multipliers
+from worldhistory.regions import Regions
+from worldhistory.snapshot import load_snapshot
+
+
+def small_setup(seed=1):
+    w = globe_world(1)
+    w.base["ocean"] = w.lat < -40
+    w.eras["late"] = fields(w.n, ocean=w.lat < -30, gravity_g=np.where(w.lon > 90, 0.35, 1.0))
+    h = make_history(run={"years": 60, "step": 10, "snapshot_every": 20, "seed": seed},
+                     world={"eras": ["late"]},
+                     regions={"zone": {"kind": "changed", "era": "late", "fields": ["gravity_g"], "land": "base"}},
+                     event=[{"year": 0, "kind": "seed"},
+                            {"year": 20, "kind": "cull", "region": "zone", "share": 0.6},
+                            {"year": 20, "kind": "era_switch", "era": "late"}],
+                     hazard=[{"kind": "roaming", "start": 20, "home_region": "zone", "local": 3, "wanderers": 1}],
+                     stats={"regions": ["zone"]})
+    races = link(h, [make_race("a", seed={"clusters": 10, "heads": 500},
+                               demography={"density": 2.0, "growth": 0.1, "mobility": 0.1}),
+                     make_race("b", seed={"clusters": 10, "heads": 500}, conflict={"border": 0.01}),
+                     make_race("c", emerge={"parent": "a", "birth_min": 0.0, "birth_sure": 0.0, "birth_rate": 1.0,
+                                            "return_min": 1.0, "return_sure": 1.0})])
+    return w, h, races
+
+
+class EngineTest(unittest.TestCase):
+    def test_run_writes_outputs(self):
+        w, h, races = small_setup()
+        with tempfile.TemporaryDirectory() as d:
+            stats = Engine(w, h, races).run(d)
+            files = sorted(p.name for p in (Path(d) / "snap").iterdir())
+            self.assertEqual(files, ["y0000.npz", "y0020.npz", "y0040.npz", "y0060.npz"])
+            run = tomllib.loads((Path(d) / "run.toml").read_text())
+            self.assertEqual(run["seed"], 1)
+            snap = load_snapshot(Path(d) / "snap" / "y0060.npz", w.n)
+            self.assertEqual(snap["races"], ["a", "b", "c"])
+            self.assertTrue((Path(d) / "stats.json").exists())
+        self.assertEqual([s["year"] for s in stats["steps"]], [0, 10, 20, 30, 40, 50, 60])
+        self.assertEqual(stats["events"][0]["after"]["a"], 5000.0)     # 10 clusters × 500 at the gifting
+        kinds = [e["kind"] for e in stats["events"]]
+        self.assertEqual(kinds, ["seed", "cull", "era_switch"])
+        self.assertIn("zone", stats["steps"][-1]["regions"])
+        self.assertEqual(len(stats["emergence"]), 1)                  # exposure 10 yrs → c emerges once
+        self.assertGreater(stats["steps"][-1]["pop"]["c"], 0)
+        self.assertEqual(stats["steps"][-1]["family"]["a"],
+                         stats["steps"][-1]["pop"]["a"] + stats["steps"][-1]["pop"]["c"])
+
+    def test_no_birth_where_native_curves_do_not_fit(self):
+        w, h, _ = small_setup()
+        races = link(h, [make_race("a", seed={"clusters": 10, "heads": 500},
+                                   demography={"density": 2.0, "growth": 0.1, "mobility": 0.1}),
+                         make_race("c", emerge={"parent": "a", "birth_min": 0.0, "birth_sure": 0.0,
+                                                "birth_rate": 1.0},
+                                   tolerance={"temperature": {"optimum": -80, "width": 1, "lo": -90, "hi": 60}})])
+        with tempfile.TemporaryDirectory() as d:
+            stats = Engine(w, h, races).run(d)
+        self.assertEqual(stats["emergence"], [])
+
+    def test_progress_and_returns_in_stats(self):
+        w, h, races = small_setup()
+        with tempfile.TemporaryDirectory() as d:
+            stats = Engine(w, h, races).run(d)
+        self.assertIn("a>c", stats["steps"][-1]["progress"])
+        self.assertIsInstance(stats["returns"], list)
+        self.assertIsInstance(stats["raids"], list)
+        self.assertIn("a", stats["emergence"][0])
+
+    def test_deterministic(self):
+        a = Engine(*small_setup(1))
+        b = Engine(*small_setup(1))
+        c = Engine(*small_setup(2))
+        for e in (a, b, c):
+            for _ in range(5):
+                e.step()
+        np.testing.assert_array_equal(a.state.P, b.state.P)
+        self.assertFalse(np.array_equal(a.state.P, c.state.P))
+
+    def test_depth_creep_rate(self):
+        n = 16
+        w = line_world(n, ocean=[False] + [True] * (n - 1), elevation_m=[10] + [-100.0 * i for i in range(1, n)])
+        tol = {"depth": {"optimum": 0, "width": 30, "lo": 0, "hi": 5000, "rate": 2.0}}
+        h = make_history(run={"years": 600, "step": 10, "seed": 1}, event=[])
+        r = make_race("d", habitat={"realm": "both", "terms": [{"p": "land", "w": 1}, {"p": "sea", "w": 1}]},
+                      demography={"density": 5.0, "growth": 0.3, "founder": 2.0, "mobility": 0.2}, tolerance=tol)
+        e = Engine(w, h, link(h, [r]))
+        e.state.P[0, 0] = 1000.0
+        front = []
+        for _ in range(60):
+            e.step()
+            occ = np.flatnonzero(e.state.P[0] >= 1)
+            front.append(100.0 * occ.max())
+        self.assertEqual(front, sorted(front))                        # only deeper with time
+        self.assertGreater(front[-1], 0.4 * 2.0 * 600)                # ≈ rate × years (1,200 m)
+        self.assertLess(front[-1], 1.3 * 2.0 * 600)
+
+    def test_nudges(self):
+        w = line_world(3)
+        g, c, x, a = nudge_multipliers([{"race": "a", "region": "all", "years": [0, 100], "growth": 1.1,
+                                         "capacity": 1.2, "expansion": 1.3, "adapt": 2.0}],
+                                       [make_race("a")], Regions(w, {}), 50, 3)
+        np.testing.assert_allclose(g, 1.1)
+        np.testing.assert_allclose(c, 1.2)
+        np.testing.assert_allclose(a, 2.0)
+        g, c, x, a = nudge_multipliers([{"race": "a", "region": "all", "years": [0, 10], "growth": 1.1, "capacity": 1,
+                                         "expansion": 1, "adapt": 1}], [make_race("a")], Regions(w, {}), 50, 3)
+        np.testing.assert_allclose(g, 1.0)
+
+
+@unittest.skipUnless(os.environ.get("WORLDHISTORY_SMOKE_WORLD"), "set WORLDHISTORY_SMOKE_WORLD and _CONFIG")
+class SmokeTest(unittest.TestCase):
+    def test_100_years_on_a_real_world(self):
+        from worldhistory.world import load_world
+        h, races = load_config(os.environ["WORLDHISTORY_SMOKE_CONFIG"])
+        w = load_world(os.environ["WORLDHISTORY_SMOKE_WORLD"], eras=h.eras, extra_fields=h.world_fields)
+        with tempfile.TemporaryDirectory() as d:
+            stats = Engine(w, h, races).run(d, years=100)
+        self.assertGreater(sum(stats["steps"][-1]["pop"].values()), 0)