raw · 6929 bytes
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 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) |