MM classes implementation.

epymorph/data/__init__.py

@@ -7,5 +7,4 @@
     'geo_library_static',
     'ipm_library',
     'mm_library',
-    'mm_library_parsed',
 ]

epymorph/data/mm/centroids.py

@@ -0,0 +1,60 @@
+from functools import cached_property
+
+import numpy as np
+from numpy.typing import NDArray
+
+from epymorph.data import registry
+from epymorph.data_shape import Shapes
+from epymorph.data_type import CentroidType, SimDType
+from epymorph.movement_model import EveryDay, MovementClause, MovementModel
+from epymorph.simulation import AttributeDef, Tick, TickDelta, TickIndex
+from epymorph.util import pairwise_haversine, row_normalize
+
+
+class CentroidsClause(MovementClause):
+    """The clause of the centroids model."""
+    requirements = (
+        AttributeDef('population', int, Shapes.N,
+                     comment="The total population at each node."),
+        AttributeDef('centroid', CentroidType, Shapes.N,
+                     comment="The centroids for each node as (longitude, latitude) tuples."),
+        AttributeDef('phi', float, Shapes.S, default_value=40.0,
+                     comment="Influences the distance that movers tend to travel."),
+        AttributeDef('commuter_proportion', float, Shapes.S, default_value=0.1,
+                     comment="Decides what proportion of the total population should be commuting normally.")
+    )
+
+    predicate = EveryDay()
+    leaves = TickIndex(step=0)
+    returns = TickDelta(step=1, days=0)
+
+    @cached_property
+    def dispersal_kernel(self) -> NDArray[np.float64]:
+        """
+        The NxN matrix or dispersal kernel describing the tendency for movers to move to a particular location.
+        In this model, the kernel is:
+            1 / e ^ (distance / phi)
+        which is then row-normalized.
+        """
+        centroid = self.data('centroid')
+        phi = self.data('phi')
+        distance = pairwise_haversine(centroid['longitude'], centroid['latitude'])
+        return row_normalize(1 / np.exp(distance / phi))
+
+    def evaluate(self, tick: Tick) -> NDArray[np.int64]:
+        pop = self.data('population')
+        comm_prop = self.data('commuter_proportion')
+        n_commuters = np.floor(pop * comm_prop).astype(SimDType)
+        return self.rng.multinomial(n_commuters, self.dispersal_kernel)
+
+
+@registry.mm('centroids')
+class Centroids(MovementModel):
+    """
+    The centroids MM describes a basic commuter movement where a fixed proportion
+    of the population commutes every day, travels to another location for 1/3 of a day
+    (with a location likelihood that decreases with distance), and then returns home for
+    the remaining 2/3 of the day.
+    """
+    steps = (1 / 3, 2 / 3)
+    clauses = (CentroidsClause(),)

epymorph/data/mm/flat.py

@@ -0,0 +1,53 @@
+from functools import cached_property
+
+import numpy as np
+from numpy.typing import NDArray
+
+from epymorph.data import registry
+from epymorph.data_shape import Shapes
+from epymorph.data_type import SimDType
+from epymorph.movement_model import EveryDay, MovementClause, MovementModel
+from epymorph.simulation import AttributeDef, Tick, TickDelta, TickIndex
+from epymorph.util import row_normalize
+
+
+class FlatClause(MovementClause):
+    """The clause of the flat model."""
+    requirements = (
+        AttributeDef('population', int, Shapes.N,
+                     comment="The total population at each node."),
+        AttributeDef('commuter_proportion', float, Shapes.S, default_value=0.1,
+                     comment="Decides what proportion of the total population should be commuting normally.")
+    )
+
+    predicate = EveryDay()
+    leaves = TickIndex(step=0)
+    returns = TickDelta(step=1, days=0)
+
+    @cached_property
+    def dispersal_kernel(self) -> NDArray[np.float64]:
+        """
+        The NxN matrix or dispersal kernel describing the tendency for movers to move to a particular location.
+        In this model, the kernel is full of 1s except for 0s on the diagonal, which is then row-normalized.
+        Effectively: every destination is equally likely.
+        """
+        ones = np.ones((self.dim.nodes, self.dim.nodes))
+        np.fill_diagonal(ones, 0)
+        return row_normalize(ones)
+
+    def evaluate(self, tick: Tick) -> NDArray[SimDType]:
+        pop = self.data('population')
+        comm_prop = self.data('commuter_proportion')
+        n_commuters = np.floor(pop * comm_prop).astype(SimDType)
+        return self.rng.multinomial(n_commuters, self.dispersal_kernel)
+
+
+@registry.mm('flat')
+class Flat(MovementModel):
+    """
+    This model evenly weights the probability of movement to all other nodes.
+    It uses parameter 'commuter_proportion' to determine how many people should
+    be moving, based on the total normal population of each node.
+    """
+    steps = (1 / 3, 2 / 3)
+    clauses = (FlatClause(),)

epymorph/data/mm/icecube.py

@@ -0,0 +1,43 @@
+import numpy as np
+from numpy.typing import NDArray
+
+from epymorph.data import registry
+from epymorph.data_shape import Shapes
+from epymorph.data_type import SimDType
+from epymorph.movement_model import EveryDay, MovementClause, MovementModel
+from epymorph.simulation import AttributeDef, Tick, TickDelta, TickIndex
+
+
+class IcecubeClause(MovementClause):
+    """The clause of the icecube model."""
+    requirements = (
+        AttributeDef('population', int, Shapes.N,
+                     comment="The total population at each node."),
+        AttributeDef('commuter_proportion', float, Shapes.S, default_value=0.1,
+                     comment="Decides what proportion of the total population should be commuting normally.")
+    )
+
+    predicate = EveryDay()
+    leaves = TickIndex(step=0)
+    returns = TickDelta(step=1, days=0)
+
+    def evaluate(self, tick: Tick) -> NDArray[np.int64]:
+        N = self.dim.nodes
+        pop = self.data('population')
+        comm_prop = self.data('commuter_proportion')
+        commuters = np.zeros((N, N), dtype=SimDType)
+        for src in range(N):
+            if (src + 1) < N:
+                commuters[src, src + 1] = pop[src] * comm_prop
+        return commuters
+
+
+@registry.mm('icecube')
+class Icecube(MovementModel):
+    """
+    A toy example: ice cube tray movement movement model
+    Each state sends a fixed number of commuters to the next
+    state in the line (without wraparound).
+    """
+    steps = (1 / 2, 1 / 2)
+    clauses = (IcecubeClause(),)

epymorph/data/mm/no.py

@@ -0,0 +1,29 @@
+import numpy as np
+from numpy.typing import NDArray
+
+from epymorph.data import registry
+from epymorph.data_type import SimDType
+from epymorph.movement_model import EveryDay, MovementClause, MovementModel
+from epymorph.simulation import Tick, TickDelta, TickIndex
+
+
+class NoClause(MovementClause):
+    """The clause of the "no" model."""
+    requirements = ()
+    predicate = EveryDay()
+    leaves = TickIndex(step=0)
+    returns = TickDelta(step=0, days=0)
+
+    def evaluate(self, tick: Tick) -> NDArray[np.int64]:
+        N = self.dim.nodes
+        return np.zeros((N, N), dtype=SimDType)
+
+
+@registry.mm('no')
+class No(MovementModel):
+    """
+    No movement at all. This is handy for cases when you want to disable movement
+    in an experiment, or for testing.
+    """
+    steps = (1.0,)
+    clauses = (NoClause(),)