aboutsummaryrefslogtreecommitdiffziptar.gz
path: root/tests/test_engine.py
blob: b0178e24d9fbfdbcebcd00d50634e84e9ad7bf3c (plain)
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)