Skip to content

API reference

Core geometry and designs

Planting-site geometry.

The canonical geometry is an arbitrary set of two-dimensional plant coordinates. Rectangles, masks, polygons, GPS-like point sets and other shapes are convenience constructors around that representation.

Source code in src/cropmix/geometry.py
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@dataclass(frozen=True)
class Field:
    """Planting-site geometry.

    The canonical geometry is an arbitrary set of two-dimensional plant
    coordinates.  Rectangles, masks, polygons, GPS-like point sets and other
    shapes are convenience constructors around that representation.
    """

    coordinates: np.ndarray
    site_ids: tuple[str, ...] | None = None
    boundary: np.ndarray | None = None
    grid_shape: tuple[int, int] | None = None
    metadata: dict[str, object] = dc_field(default_factory=dict)

    def __post_init__(self) -> None:
        coords = _as_coordinates(self.coordinates)
        object.__setattr__(self, "coordinates", coords)

        if self.site_ids is None:
            ids = tuple(str(i) for i in range(coords.shape[0]))
        else:
            ids = tuple(str(value) for value in self.site_ids)
            if len(ids) != coords.shape[0]:
                raise ValidationError("site_ids must have one entry per coordinate.")
            if len(set(ids)) != len(ids):
                raise ValidationError("site_ids must be unique.")
        object.__setattr__(self, "site_ids", ids)

        if self.boundary is not None:
            boundary = np.asarray(self.boundary, dtype=float)
            if boundary.ndim != 2 or boundary.shape[1] != 2 or len(boundary) < 3:
                raise ValidationError("boundary must have shape (n_vertices, 2), n_vertices >= 3.")
            if not np.isfinite(boundary).all():
                raise ValidationError("boundary coordinates must be finite.")
            boundary.setflags(write=False)
            object.__setattr__(self, "boundary", boundary)

        if self.grid_shape is not None:
            rows, cols = self.grid_shape
            if rows <= 0 or cols <= 0 or rows * cols != self.n_sites:
                raise ValidationError("grid_shape must match the number of sites.")

    @property
    def n_sites(self) -> int:
        return int(self.coordinates.shape[0])

    @property
    def x(self) -> np.ndarray:
        return self.coordinates[:, 0]

    @property
    def y(self) -> np.ndarray:
        return self.coordinates[:, 1]

    @property
    def extent(self) -> tuple[float, float, float, float]:
        return (
            float(self.x.min()),
            float(self.x.max()),
            float(self.y.min()),
            float(self.y.max()),
        )

    def distance_matrix(self) -> np.ndarray:
        delta = self.coordinates[:, None, :] - self.coordinates[None, :, :]
        return np.sqrt(np.sum(delta * delta, axis=2))

    def site_index(self, site_id: str) -> int:
        try:
            return self.site_ids.index(str(site_id))
        except ValueError as exc:
            raise KeyError(f"Unknown site_id: {site_id!r}") from exc

    @classmethod
    def from_coordinates(
        cls,
        coordinates: Iterable[Sequence[float]],
        *,
        site_ids: Sequence[str] | None = None,
        boundary: Iterable[Sequence[float]] | None = None,
        metadata: dict[str, object] | None = None,
    ) -> Field:
        return cls(
            coordinates=np.asarray(list(coordinates), dtype=float),
            site_ids=None if site_ids is None else tuple(site_ids),
            boundary=None if boundary is None else np.asarray(list(boundary), dtype=float),
            metadata={} if metadata is None else dict(metadata),
        )

    @classmethod
    def rectangular(
        cls,
        rows: int,
        columns: int,
        *,
        spacing: float | tuple[float, float] = 1.0,
        origin: tuple[float, float] = (0.0, 0.0),
    ) -> Field:
        if rows <= 0 or columns <= 0:
            raise ValidationError("rows and columns must be positive.")
        if isinstance(spacing, tuple):
            sx, sy = float(spacing[0]), float(spacing[1])
        else:
            sx = sy = float(spacing)
        if sx <= 0 or sy <= 0:
            raise ValidationError("spacing must be positive.")

        ox, oy = map(float, origin)
        coords = np.array(
            [(ox + c * sx, oy + r * sy) for r in range(rows) for c in range(columns)],
            dtype=float,
        )
        boundary = np.array(
            [
                (ox, oy),
                (ox + (columns - 1) * sx, oy),
                (ox + (columns - 1) * sx, oy + (rows - 1) * sy),
                (ox, oy + (rows - 1) * sy),
            ],
            dtype=float,
        )
        return cls(
            coordinates=coords,
            boundary=boundary,
            grid_shape=(rows, columns),
            metadata={"constructor": "rectangular", "spacing": (sx, sy)},
        )

    @classmethod
    def from_mask(
        cls,
        mask: np.ndarray,
        *,
        spacing: float | tuple[float, float] = 1.0,
        origin: tuple[float, float] = (0.0, 0.0),
    ) -> Field:
        mask = np.asarray(mask, dtype=bool)
        if mask.ndim != 2:
            raise ValidationError("mask must be two-dimensional.")
        if not mask.any():
            raise ValidationError("mask must contain at least one True cell.")
        if isinstance(spacing, tuple):
            sx, sy = map(float, spacing)
        else:
            sx = sy = float(spacing)
        if sx <= 0 or sy <= 0:
            raise ValidationError("spacing must be positive.")
        ox, oy = map(float, origin)
        coords = []
        ids = []
        for r, c in zip(*np.where(mask), strict=False):
            coords.append((ox + c * sx, oy + r * sy))
            ids.append(f"r{r}c{c}")
        return cls(
            coordinates=np.asarray(coords),
            site_ids=tuple(ids),
            metadata={"constructor": "mask", "mask_shape": tuple(mask.shape), "spacing": (sx, sy)},
        )

    @classmethod
    def from_polygon(
        cls,
        boundary: Iterable[Sequence[float]],
        *,
        spacing: float | tuple[float, float] = 1.0,
        origin: tuple[float, float] | None = None,
    ) -> Field:
        """Generate a regular planting lattice clipped to an arbitrary polygon.

        For already surveyed planting positions, prefer :meth:`from_coordinates`.
        """
        polygon = np.asarray(list(boundary), dtype=float)
        if polygon.ndim != 2 or polygon.shape[1] != 2 or len(polygon) < 3:
            raise ValidationError("boundary must contain at least three 2D vertices.")
        if isinstance(spacing, tuple):
            sx, sy = map(float, spacing)
        else:
            sx = sy = float(spacing)
        if sx <= 0 or sy <= 0:
            raise ValidationError("spacing must be positive.")

        xmin, ymin = polygon.min(axis=0)
        xmax, ymax = polygon.max(axis=0)
        if origin is None:
            ox, oy = float(xmin), float(ymin)
        else:
            ox, oy = map(float, origin)

        xs = np.arange(ox, xmax + sx * 0.5, sx)
        ys = np.arange(oy, ymax + sy * 0.5, sy)
        coords = [(x, y) for y in ys for x in xs if _point_in_polygon((x, y), polygon)]
        if not coords:
            raise ValidationError("No planting sites fall inside the polygon at the requested spacing.")
        return cls(
            coordinates=np.asarray(coords, dtype=float),
            boundary=polygon,
            metadata={"constructor": "polygon", "spacing": (sx, sy)},
        )

from_polygon(boundary, *, spacing=1.0, origin=None) classmethod

Generate a regular planting lattice clipped to an arbitrary polygon.

For already surveyed planting positions, prefer :meth:from_coordinates.

Source code in src/cropmix/geometry.py
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
@classmethod
def from_polygon(
    cls,
    boundary: Iterable[Sequence[float]],
    *,
    spacing: float | tuple[float, float] = 1.0,
    origin: tuple[float, float] | None = None,
) -> Field:
    """Generate a regular planting lattice clipped to an arbitrary polygon.

    For already surveyed planting positions, prefer :meth:`from_coordinates`.
    """
    polygon = np.asarray(list(boundary), dtype=float)
    if polygon.ndim != 2 or polygon.shape[1] != 2 or len(polygon) < 3:
        raise ValidationError("boundary must contain at least three 2D vertices.")
    if isinstance(spacing, tuple):
        sx, sy = map(float, spacing)
    else:
        sx = sy = float(spacing)
    if sx <= 0 or sy <= 0:
        raise ValidationError("spacing must be positive.")

    xmin, ymin = polygon.min(axis=0)
    xmax, ymax = polygon.max(axis=0)
    if origin is None:
        ox, oy = float(xmin), float(ymin)
    else:
        ox, oy = map(float, origin)

    xs = np.arange(ox, xmax + sx * 0.5, sx)
    ys = np.arange(oy, ymax + sy * 0.5, sy)
    coords = [(x, y) for y in ys for x in xs if _point_in_polygon((x, y), polygon)]
    if not coords:
        raise ValidationError("No planting sites fall inside the polygon at the requested spacing.")
    return cls(
        coordinates=np.asarray(coords, dtype=float),
        boundary=polygon,
        metadata={"constructor": "polygon", "spacing": (sx, sy)},
    )

Assign exactly one named variety to every planting site.

Source code in src/cropmix/design.py
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@dataclass(frozen=True)
class MixtureDesign:
    """Assign exactly one named variety to every planting site."""

    field: Field
    assignment: tuple[str, ...]

    def __post_init__(self) -> None:
        assignment = tuple(str(x) for x in self.assignment)
        if len(assignment) != self.field.n_sites:
            raise ValidationError(
                f"assignment has {len(assignment)} labels but field has {self.field.n_sites} sites."
            )
        if any(not name.strip() for name in assignment):
            raise ValidationError("Variety labels cannot be empty.")
        object.__setattr__(self, "assignment", assignment)

    @property
    def n_sites(self) -> int:
        return self.field.n_sites

    @property
    def varieties(self) -> tuple[str, ...]:
        return tuple(sorted(set(self.assignment)))

    @property
    def counts(self) -> dict[str, int]:
        return dict(Counter(self.assignment))

    @property
    def proportions(self) -> dict[str, float]:
        return {name: count / self.n_sites for name, count in self.counts.items()}

    def variety_indices(self, variety: str) -> np.ndarray:
        return np.flatnonzero(np.asarray(self.assignment, dtype=object) == variety)

    def as_grid(self) -> np.ndarray:
        if self.field.grid_shape is None:
            raise ValidationError("This field has no complete rectangular grid_shape.")
        return np.asarray(self.assignment, dtype=object).reshape(self.field.grid_shape)

    def to_dataframe(self) -> pd.DataFrame:
        return pd.DataFrame(
            {
                "site_id": self.field.site_ids,
                "x": self.field.x,
                "y": self.field.y,
                "variety": self.assignment,
            }
        )

    def swapped(self, first: int, second: int) -> MixtureDesign:
        if not (0 <= first < self.n_sites and 0 <= second < self.n_sites):
            raise IndexError("Swap indices are outside the field.")
        values = list(self.assignment)
        values[first], values[second] = values[second], values[first]
        return MixtureDesign(self.field, tuple(values))

    def plot(self, ax=None, *, marker_size: float = 80, legend: bool = True):
        """Plot the planting assignment. Requires the optional `viz` extra."""
        try:
            import matplotlib.pyplot as plt
        except ImportError as exc:  # pragma: no cover - optional dependency
            raise ImportError("Install plotting support with `pip install cropmix[viz]`.") from exc

        if ax is None:
            _, ax = plt.subplots()
        for variety in self.varieties:
            idx = self.variety_indices(variety)
            ax.scatter(self.field.x[idx], self.field.y[idx], s=marker_size, label=variety)
        if self.field.boundary is not None:
            boundary = np.vstack([self.field.boundary, self.field.boundary[0]])
            ax.plot(boundary[:, 0], boundary[:, 1], linewidth=1)
        ax.set_aspect("equal", adjustable="box")
        ax.set_xlabel("x")
        ax.set_ylabel("y")
        if legend:
            ax.legend()
        return ax

    @classmethod
    def monoculture(cls, field: Field, variety: str) -> MixtureDesign:
        return cls(field=field, assignment=tuple([variety] * field.n_sites))

    @classmethod
    def random(
        cls,
        field: Field,
        counts: Mapping[str, int],
        *,
        seed: int | None = None,
    ) -> MixtureDesign:
        counts = {str(name): int(count) for name, count in counts.items()}
        if any(count < 0 for count in counts.values()):
            raise ValidationError("Variety counts cannot be negative.")
        if sum(counts.values()) != field.n_sites:
            raise ValidationError(
                f"Counts sum to {sum(counts.values())}, but field has {field.n_sites} sites."
            )
        values: list[str] = []
        for name, count in counts.items():
            values.extend([name] * count)
        rng = np.random.default_rng(seed)
        rng.shuffle(values)
        return cls(field=field, assignment=tuple(values))

    @classmethod
    def from_grid(cls, grid: Sequence[Sequence[str]], *, spacing: float = 1.0) -> MixtureDesign:
        array = np.asarray(grid, dtype=object)
        if array.ndim != 2:
            raise ValidationError("grid must be two-dimensional.")
        field = Field.rectangular(array.shape[0], array.shape[1], spacing=spacing)
        return cls(field=field, assignment=tuple(str(x) for x in array.ravel()))

    @classmethod
    def from_dataframe(
        cls,
        dataframe: pd.DataFrame,
        *,
        x: str = "x",
        y: str = "y",
        variety: str = "variety",
        site_id: str | None = "site_id",
    ) -> MixtureDesign:
        required = [x, y, variety]
        missing = [column for column in required if column not in dataframe.columns]
        if missing:
            raise ValidationError(f"Missing columns: {missing}")
        ids = None
        if site_id is not None and site_id in dataframe.columns:
            ids = tuple(dataframe[site_id].astype(str))
        field = Field.from_coordinates(
            dataframe[[x, y]].to_numpy(float),
            site_ids=ids,
        )
        return cls(field=field, assignment=tuple(dataframe[variety].astype(str)))

plot(ax=None, *, marker_size=80, legend=True)

Plot the planting assignment. Requires the optional viz extra.

Source code in src/cropmix/design.py
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def plot(self, ax=None, *, marker_size: float = 80, legend: bool = True):
    """Plot the planting assignment. Requires the optional `viz` extra."""
    try:
        import matplotlib.pyplot as plt
    except ImportError as exc:  # pragma: no cover - optional dependency
        raise ImportError("Install plotting support with `pip install cropmix[viz]`.") from exc

    if ax is None:
        _, ax = plt.subplots()
    for variety in self.varieties:
        idx = self.variety_indices(variety)
        ax.scatter(self.field.x[idx], self.field.y[idx], s=marker_size, label=variety)
    if self.field.boundary is not None:
        boundary = np.vstack([self.field.boundary, self.field.boundary[0]])
        ax.plot(boundary[:, 0], boundary[:, 1], linewidth=1)
    ax.set_aspect("equal", adjustable="box")
    ax.set_xlabel("x")
    ax.set_ylabel("y")
    if legend:
        ax.legend()
    return ax

Biology

Host-dependent virus transmission rates.

Parameters:

Name Type Description Default
acquisition_rate float

Rate at which a virus-free vector acquires virus while feeding on an infectious plant of this variety, in day^-1.

required
inoculation_rate float

Per-virus-bearing-vector inoculation rate for a susceptible plant of this variety, in vector^-1 day^-1.

required
Source code in src/cropmix/biology.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass(frozen=True)
class HostTransmission:
    """Host-dependent virus transmission rates.

    Parameters
    ----------
    acquisition_rate:
        Rate at which a virus-free vector acquires virus while feeding on an
        infectious plant of this variety, in day^-1.
    inoculation_rate:
        Per-virus-bearing-vector inoculation rate for a susceptible plant of
        this variety, in vector^-1 day^-1.
    """

    acquisition_rate: float
    inoculation_rate: float

    def __post_init__(self) -> None:
        _nonnegative("acquisition_rate", self.acquisition_rate)
        _nonnegative("inoculation_rate", self.inoculation_rate)

Plant-side epidemic rates, in day^-1.

Source code in src/cropmix/biology.py
45
46
47
48
49
50
51
52
53
54
@dataclass(frozen=True)
class PlantParameters:
    """Plant-side epidemic rates, in day^-1."""

    latent_progression_rate: float
    roguing_rate: float = 0.0

    def __post_init__(self) -> None:
        _nonnegative("latent_progression_rate", self.latent_progression_rate)
        _nonnegative("roguing_rate", self.roguing_rate)

Healthy and infectious yields in a common user-chosen unit.

Source code in src/cropmix/biology.py
57
58
59
60
61
62
63
64
65
66
67
68
69
@dataclass(frozen=True)
class YieldParameters:
    """Healthy and infectious yields in a common user-chosen unit."""

    healthy: float
    infected: float
    unit: str = "t/ha"

    def __post_init__(self) -> None:
        _nonnegative("healthy yield", self.healthy)
        _nonnegative("infected yield", self.infected)
        if not self.unit:
            raise ValidationError("Yield unit cannot be empty.")

One crop variety used in a mixture.

Source code in src/cropmix/biology.py
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass(frozen=True)
class Variety:
    """One crop variety used in a mixture."""

    name: str
    transmission: HostTransmission
    plant: PlantParameters
    yield_model: YieldParameters

    def __post_init__(self) -> None:
        if not self.name or not self.name.strip():
            raise ValidationError("Variety name cannot be empty.")

    def with_transmission(self, transmission: HostTransmission) -> Variety:
        """Return a copy with updated transmission rates."""
        return replace(self, transmission=transmission)

with_transmission(transmission)

Return a copy with updated transmission rates.

Source code in src/cropmix/biology.py
85
86
87
def with_transmission(self, transmission: HostTransmission) -> Variety:
    """Return a copy with updated transmission rates."""
    return replace(self, transmission=transmission)

Vector demographic and movement rates, in day^-1.

Source code in src/cropmix/biology.py
90
91
92
93
94
95
96
97
98
99
@dataclass(frozen=True)
class VectorParameters:
    """Vector demographic and movement rates, in day^-1."""

    mortality_rate: float
    dispersal_rate: float

    def __post_init__(self) -> None:
        _nonnegative("mortality_rate", self.mortality_rate)
        _nonnegative("dispersal_rate", self.dispersal_rate)

Virus-vector parameters shared across host varieties.

vector_latent_progression_rate is required only for PT transmission. Cropmix 0.1 can infer PT parameters through the EpiPvr bridge, but the spatial simulation engine currently implements SPT dynamics only.

Source code in src/cropmix/biology.py
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
129
130
131
132
@dataclass(frozen=True)
class PathogenParameters:
    """Virus-vector parameters shared across host varieties.

    `vector_latent_progression_rate` is required only for PT transmission.
    Cropmix 0.1 can infer PT parameters through the EpiPvr bridge, but the
    spatial simulation engine currently implements SPT dynamics only.
    """

    vector_clearance_rate: float
    transmission_mode: TransmissionMode = "SPT"
    vector_latent_progression_rate: float | None = None

    def __post_init__(self) -> None:
        _nonnegative("vector_clearance_rate", self.vector_clearance_rate)
        if self.transmission_mode not in ("SPT", "PT"):
            raise ValidationError("transmission_mode must be 'SPT' or 'PT'.")
        if self.transmission_mode == "PT":
            if self.vector_latent_progression_rate is None:
                raise ValidationError(
                    "PT transmission requires vector_latent_progression_rate."
                )
            _positive(
                "vector_latent_progression_rate",
                self.vector_latent_progression_rate,
            )
        elif self.vector_latent_progression_rate is not None:
            _nonnegative(
                "vector_latent_progression_rate",
                self.vector_latent_progression_rate,
            )

Biological parameters plus the vector movement kernel.

Source code in src/cropmix/system.py
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
@dataclass(frozen=True)
class CropMixSystem:
    """Biological parameters plus the vector movement kernel."""

    varieties: tuple[Variety, ...]
    vector: VectorParameters
    pathogen: PathogenParameters
    kernel: ExponentialKernel = ExponentialKernel()

    def __post_init__(self) -> None:
        varieties = tuple(self.varieties)
        if not varieties:
            raise ValidationError("At least one variety is required.")
        names = [variety.name for variety in varieties]
        if len(set(names)) != len(names):
            raise ValidationError("Variety names must be unique within a system.")
        yield_units = {variety.yield_model.unit for variety in varieties}
        if len(yield_units) != 1:
            raise ValidationError("All varieties must use the same yield unit.")
        object.__setattr__(self, "varieties", varieties)

    @property
    def variety_names(self) -> tuple[str, ...]:
        return tuple(variety.name for variety in self.varieties)

    @property
    def variety_map(self) -> dict[str, Variety]:
        return {variety.name: variety for variety in self.varieties}

    @property
    def yield_unit(self) -> str:
        return self.varieties[0].yield_model.unit

    def variety(self, name: str) -> Variety:
        try:
            return self.variety_map[name]
        except KeyError as exc:
            raise KeyError(f"Unknown variety {name!r}. Available: {self.variety_names}") from exc

    def validate_design(self, design: MixtureDesign) -> None:
        missing = set(design.varieties) - set(self.variety_names)
        if missing:
            raise ValidationError(f"Design uses varieties not present in the system: {sorted(missing)}")

    def ensure_spatial_supported(self) -> None:
        if self.pathogen.transmission_mode != "SPT":
            raise UnsupportedModelError(
                "Cropmix 0.1 spatial simulation implements SPT dynamics only. "
                "PT inference is supported by the EpiPvr bridge, but PT spatial simulation "
                "requires an explicit exposed-vector compartment and is intentionally not guessed."
            )

    def with_kernel_scale(self, scale: float) -> CropMixSystem:
        return replace(self, kernel=self.kernel.with_scale(scale))

    def point_transmission_draw(self) -> TransmissionDraw:
        return TransmissionDraw(
            acquisition_rates={v.name: v.transmission.acquisition_rate for v in self.varieties},
            inoculation_rates={v.name: v.transmission.inoculation_rate for v in self.varieties},
            vector_clearance_rate=self.pathogen.vector_clearance_rate,
            vector_latent_progression_rate=self.pathogen.vector_latent_progression_rate,
        )

    @classmethod
    def from_iterable(
        cls,
        varieties: Iterable[Variety],
        *,
        vector: VectorParameters,
        pathogen: PathogenParameters,
        kernel: ExponentialKernel | None = None,
    ) -> CropMixSystem:
        return cls(
            varieties=tuple(varieties),
            vector=vector,
            pathogen=pathogen,
            kernel=ExponentialKernel() if kernel is None else kernel,
        )

Simulation

Simulate a supplied planting design under the spatial SPT model.

Parameters:

Name Type Description Default
design MixtureDesign

Arbitrary assignment of system varieties to field coordinates.

required
system CropMixSystem

Varieties, vector/pathogen parameters and movement kernel.

required
scenario Scenario

Season duration, vector burden and initial inoculum.

required
n_runs int

Number of stochastic epidemic replicates.

100
seed int

Master seed. Reusing it across candidate designs supplies common random numbers for paired comparisons and optimization.

12345
observation_times ndarray | None

Times at which trajectories are recorded. Defaults to 101 equally spaced points including 0 and harvest.

None
transmission_draw TransmissionDraw | None

Optional coherent parameter draw. If omitted, point values in system are used.

None
store_final_states bool

If false, final state arrays are discarded after computing summaries.

True
Notes

The event loop is Numba compiled. The first call in a Python process pays a compilation cost; subsequent calls reuse cached compiled functions when possible and are substantially faster than the reference pure-Python loop.

Source code in src/cropmix/simulation.py
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
def simulate_mixture(
    design: MixtureDesign,
    system: CropMixSystem,
    scenario: Scenario,
    *,
    n_runs: int = 100,
    seed: int = 12345,
    observation_times: np.ndarray | None = None,
    transmission_draw: TransmissionDraw | None = None,
    store_final_states: bool = True,
) -> SimulationResult:
    """Simulate a supplied planting design under the spatial SPT model.

    Parameters
    ----------
    design:
        Arbitrary assignment of system varieties to field coordinates.
    system:
        Varieties, vector/pathogen parameters and movement kernel.
    scenario:
        Season duration, vector burden and initial inoculum.
    n_runs:
        Number of stochastic epidemic replicates.
    seed:
        Master seed. Reusing it across candidate designs supplies common random
        numbers for paired comparisons and optimization.
    observation_times:
        Times at which trajectories are recorded. Defaults to 101 equally
        spaced points including 0 and harvest.
    transmission_draw:
        Optional coherent parameter draw. If omitted, point values in `system`
        are used.
    store_final_states:
        If false, final state arrays are discarded after computing summaries.

    Notes
    -----
    The event loop is Numba compiled. The first call in a Python process pays a
    compilation cost; subsequent calls reuse cached compiled functions when
    possible and are substantially faster than the reference pure-Python loop.
    """
    if n_runs <= 0:
        raise ValidationError("n_runs must be positive.")

    system.ensure_spatial_supported()
    system.validate_design(design)

    if observation_times is None:
        observation_times = np.linspace(0.0, scenario.duration, 101)
    else:
        observation_times = np.asarray(observation_times, dtype=float)
        if observation_times.ndim != 1 or len(observation_times) == 0:
            raise ValidationError("observation_times must be a non-empty 1D array.")
        if np.any(np.diff(observation_times) < 0):
            raise ValidationError("observation_times must be sorted.")
        if observation_times[0] < 0 or observation_times[-1] > scenario.duration:
            raise ValidationError("observation_times must lie inside [0, scenario.duration].")

    resolved_draw = _resolve_transmission_draw(system, transmission_draw)
    prepared_kernel = system.kernel.prepare(design.field)

    names = system.variety_names
    variety_to_index = {name: i for i, name in enumerate(names)}
    cultivar = np.asarray([variety_to_index[name] for name in design.assignment], dtype=np.int16)

    acquisition = np.asarray(
        [resolved_draw.acquisition_rates[name] for name in names], dtype=np.float64
    )
    inoculation = np.asarray(
        [resolved_draw.inoculation_rates[name] for name in names], dtype=np.float64
    )
    gamma = np.asarray(
        [variety.plant.latent_progression_rate for variety in system.varieties],
        dtype=np.float64,
    )
    rho = np.asarray(
        [variety.plant.roguing_rate for variety in system.varieties], dtype=np.float64
    )
    yield_healthy = np.asarray(
        [variety.yield_model.healthy for variety in system.varieties], dtype=np.float64
    )
    yield_infected = np.asarray(
        [variety.yield_model.infected for variety in system.varieties], dtype=np.float64
    )

    vector_inoculum = scenario.vector_inoculum
    if vector_inoculum.origin_weights is None:
        counts = np.bincount(cultivar.astype(np.int64), minlength=len(names)).astype(float)
        initial_origin_weights = counts / counts.sum()
    else:
        if len(vector_inoculum.origin_weights) != len(names):
            raise ValidationError(
                "vector_inoculum.origin_weights must have one value per system variety."
            )
        initial_origin_weights = np.asarray(vector_inoculum.origin_weights, dtype=np.float64)
        initial_origin_weights = initial_origin_weights / initial_origin_weights.sum()

    destination_cdf = np.cumsum(prepared_kernel.probabilities, axis=1)
    destination_cdf[:, -1] = 1.0

    initial_sites = _initial_sites_table(scenario, design.n_sites, n_runs, seed)
    event_seeds = np.arange(seed, seed + n_runs, dtype=np.int64)

    (
        yields,
        final_incidence,
        final_states,
        incidence,
        vector_prevalence,
        by_variety,
        errors,
    ) = _batch_gillespie_numba(
        cultivar,
        destination_cdf,
        scenario.vectors_per_plant,
        initial_sites,
        np.asarray(observation_times, dtype=np.float64),
        float(scenario.duration),
        event_seeds,
        acquisition,
        inoculation,
        gamma,
        rho,
        float(system.vector.dispersal_rate),
        float(system.vector.mortality_rate),
        float(resolved_draw.vector_clearance_rate),
        yield_healthy,
        yield_infected,
        float(vector_inoculum.infectious_fraction),
        initial_origin_weights,
    )

    if np.any(errors != 0):
        raise RuntimeError("Vector-count invariant failed inside the compiled simulator.")

    final_states_output = final_states if store_final_states else None
    infection_probability = (
        np.mean(final_states == I_STATE, axis=0) if store_final_states else None
    )
    incidence_by_variety_runs = {
        name: by_variety[:, index, :] for index, name in enumerate(names)
    }

    return SimulationResult(
        design=design,
        scenario=scenario,
        time=np.asarray(observation_times, dtype=float).copy(),
        yield_runs=yields,
        final_incidence_runs=final_incidence,
        incidence_runs=incidence,
        vector_prevalence_runs=vector_prevalence,
        incidence_by_variety_runs=incidence_by_variety_runs,
        final_states=final_states_output,
        infection_probability=infection_probability,
        kernel_scale=prepared_kernel.scale,
        mean_step_distance=prepared_kernel.mean_step_distance,
        seed=seed,
        yield_unit=system.yield_unit,
    )

Monte Carlo output for one supplied planting design.

Source code in src/cropmix/results.py
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@dataclass
class SimulationResult:
    """Monte Carlo output for one supplied planting design."""

    design: MixtureDesign
    scenario: Scenario
    time: np.ndarray
    yield_runs: np.ndarray
    final_incidence_runs: np.ndarray
    incidence_runs: np.ndarray
    vector_prevalence_runs: np.ndarray
    incidence_by_variety_runs: dict[str, np.ndarray]
    final_states: np.ndarray | None
    infection_probability: np.ndarray | None
    kernel_scale: float
    mean_step_distance: float
    seed: int
    yield_unit: str

    @property
    def mean_yield(self) -> float:
        return float(np.mean(self.yield_runs))

    @property
    def yield_sd(self) -> float:
        return float(np.std(self.yield_runs, ddof=1)) if len(self.yield_runs) > 1 else 0.0

    @property
    def yield_se(self) -> float:
        return self.yield_sd / np.sqrt(len(self.yield_runs)) if len(self.yield_runs) else float("nan")

    @property
    def mean_final_incidence(self) -> float:
        return float(np.mean(self.final_incidence_runs))

    @property
    def final_incidence_sd(self) -> float:
        return (
            float(np.std(self.final_incidence_runs, ddof=1))
            if len(self.final_incidence_runs) > 1
            else 0.0
        )

    @property
    def mean_incidence(self) -> np.ndarray:
        return np.mean(self.incidence_runs, axis=0)

    @property
    def incidence_sd(self) -> np.ndarray:
        """Pointwise standard deviation of stochastic incidence trajectories."""
        if self.incidence_runs.shape[0] <= 1:
            return np.zeros_like(self.mean_incidence)
        return np.std(self.incidence_runs, axis=0, ddof=1)

    @property
    def mean_vector_prevalence(self) -> np.ndarray:
        return np.mean(self.vector_prevalence_runs, axis=0)

    @property
    def vector_prevalence_sd(self) -> np.ndarray:
        """Pointwise standard deviation of stochastic vector-prevalence trajectories."""
        if self.vector_prevalence_runs.shape[0] <= 1:
            return np.zeros_like(self.mean_vector_prevalence)
        return np.std(self.vector_prevalence_runs, axis=0, ddof=1)

    @property
    def incidence_by_variety_sd(self) -> dict[str, np.ndarray]:
        """Pointwise standard deviations for each cultivar-specific incidence trajectory."""
        output: dict[str, np.ndarray] = {}
        for name, values in self.incidence_by_variety_runs.items():
            if np.isnan(values).all():
                output[name] = np.full(values.shape[1], np.nan, dtype=float)
            elif values.shape[0] <= 1:
                output[name] = np.zeros(values.shape[1], dtype=float)
            else:
                output[name] = np.nanstd(values, axis=0, ddof=1)
        return output

    def summary(self) -> pd.DataFrame:
        return pd.DataFrame(
            [
                {
                    "n_runs": len(self.yield_runs),
                    "mean_yield": self.mean_yield,
                    "yield_sd": self.yield_sd,
                    "yield_se": self.yield_se,
                    "yield_unit": self.yield_unit,
                    "mean_final_incidence": self.mean_final_incidence,
                    "final_incidence_sd": self.final_incidence_sd,
                    "kernel_scale": self.kernel_scale,
                    "mean_step_distance": self.mean_step_distance,
                }
            ]
        )

    def trajectory_dataframe(self) -> pd.DataFrame:
        incidence_mean = self.mean_incidence
        incidence_sd = self.incidence_sd
        vector_mean = self.mean_vector_prevalence
        vector_sd = self.vector_prevalence_sd
        data: dict[str, object] = {
            "time": self.time,
            "incidence": incidence_mean,
            "incidence_sd": incidence_sd,
            "incidence_lower_1sd": np.clip(incidence_mean - incidence_sd, 0.0, 1.0),
            "incidence_upper_1sd": np.clip(incidence_mean + incidence_sd, 0.0, 1.0),
            "vector_prevalence": vector_mean,
            "vector_prevalence_sd": vector_sd,
            "vector_prevalence_lower_1sd": np.clip(vector_mean - vector_sd, 0.0, 1.0),
            "vector_prevalence_upper_1sd": np.clip(vector_mean + vector_sd, 0.0, 1.0),
        }
        by_variety_sd = self.incidence_by_variety_sd
        for name, values in self.incidence_by_variety_runs.items():
            mean = (
                np.full(values.shape[1], np.nan, dtype=float)
                if np.isnan(values).all()
                else np.nanmean(values, axis=0)
            )
            sd = by_variety_sd[name]
            data[f"incidence_{name}"] = mean
            data[f"incidence_{name}_sd"] = sd
            data[f"incidence_{name}_lower_1sd"] = np.clip(mean - sd, 0.0, 1.0)
            data[f"incidence_{name}_upper_1sd"] = np.clip(mean + sd, 0.0, 1.0)
        return pd.DataFrame(data)

    def plot_incidence(
        self,
        ax=None,
        *,
        by_variety: bool = False,
        show_sd: bool = True,
        sd_multiplier: float = 1.0,
        envelope_alpha: float = 0.2,
    ):
        try:
            import matplotlib.pyplot as plt
        except ImportError as exc:  # pragma: no cover
            raise ImportError("Install plotting support with `pip install cropmix[viz]`.") from exc
        if ax is None:
            _, ax = plt.subplots()
        field_mean = self.mean_incidence
        field_sd = self.incidence_sd
        field_line = ax.plot(self.time, field_mean, label="field")[0]
        if show_sd:
            ax.fill_between(
                self.time,
                np.clip(field_mean - sd_multiplier * field_sd, 0.0, 1.0),
                np.clip(field_mean + sd_multiplier * field_sd, 0.0, 1.0),
                alpha=envelope_alpha,
                color=field_line.get_color(),
                linewidth=0,
            )
        if by_variety:
            by_variety_sd = self.incidence_by_variety_sd
            for name, values in self.incidence_by_variety_runs.items():
                mean = (
                    np.full(values.shape[1], np.nan, dtype=float)
                    if np.isnan(values).all()
                    else np.nanmean(values, axis=0)
                )
                sd = by_variety_sd[name]
                line = ax.plot(self.time, mean, label=name)[0]
                if show_sd:
                    ax.fill_between(
                        self.time,
                        np.clip(mean - sd_multiplier * sd, 0.0, 1.0),
                        np.clip(mean + sd_multiplier * sd, 0.0, 1.0),
                        alpha=envelope_alpha,
                        color=line.get_color(),
                        linewidth=0,
                    )
        ax.set_xlabel("Time (days)")
        ax.set_ylabel("Infectious plant fraction")
        ax.legend()
        return ax

    def plot_vector_prevalence(
        self,
        ax=None,
        *,
        show_sd: bool = True,
        sd_multiplier: float = 1.0,
        envelope_alpha: float = 0.2,
    ):
        try:
            import matplotlib.pyplot as plt
        except ImportError as exc:  # pragma: no cover
            raise ImportError("Install plotting support with `pip install cropmix[viz]`.") from exc
        if ax is None:
            _, ax = plt.subplots()
        mean = self.mean_vector_prevalence
        sd = self.vector_prevalence_sd
        line = ax.plot(self.time, mean)[0]
        if show_sd:
            ax.fill_between(
                self.time,
                np.clip(mean - sd_multiplier * sd, 0.0, 1.0),
                np.clip(mean + sd_multiplier * sd, 0.0, 1.0),
                alpha=envelope_alpha,
                color=line.get_color(),
                linewidth=0,
            )
        ax.set_xlabel("Time (days)")
        ax.set_ylabel("Virus-bearing vector prevalence")
        return ax

    def plot_final_infection_probability(self, ax=None, *, marker_size: float = 100):
        if self.infection_probability is None:
            raise ValueError("Final states were not stored; infection probability is unavailable.")
        try:
            import matplotlib.pyplot as plt
        except ImportError as exc:  # pragma: no cover
            raise ImportError("Install plotting support with `pip install cropmix[viz]`.") from exc
        if ax is None:
            _, ax = plt.subplots()
        scatter = ax.scatter(
            self.design.field.x,
            self.design.field.y,
            c=self.infection_probability,
            s=marker_size,
            vmin=0,
            vmax=1,
        )
        plt.colorbar(scatter, ax=ax, label="P(infectious at T)")
        ax.set_aspect("equal", adjustable="box")
        ax.set_xlabel("x")
        ax.set_ylabel("y")
        return ax

incidence_by_variety_sd property

Pointwise standard deviations for each cultivar-specific incidence trajectory.

incidence_sd property

Pointwise standard deviation of stochastic incidence trajectories.

vector_prevalence_sd property

Pointwise standard deviation of stochastic vector-prevalence trajectories.

Mean field and calibration

Solve the n-variety PLOS-style SPT mean-field model.

The deterministic model preserves the PLOS convention in which plant latent/infectious states are fractions of the entire field and vector provenance states are counts. It requires a common vector-clearance rate, which is represented at pathogen level in CropMixSystem.

Source code in src/cropmix/meanfield.py
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def solve_mean_field(
    design: MixtureDesign,
    system: CropMixSystem,
    scenario: Scenario,
    *,
    observation_times: np.ndarray | None = None,
) -> MeanFieldResult:
    """Solve the n-variety PLOS-style SPT mean-field model.

    The deterministic model preserves the PLOS convention in which plant
    latent/infectious states are fractions of the entire field and vector
    provenance states are counts.  It requires a common vector-clearance rate,
    which is represented at pathogen level in `CropMixSystem`.
    """

    if system.pathogen.transmission_mode != "SPT":
        raise UnsupportedModelError("The PLOS-style mean-field solver currently implements SPT only.")
    system.validate_design(design)

    if observation_times is None:
        observation_times = np.linspace(0.0, scenario.duration, 101)
    else:
        observation_times = np.asarray(observation_times, dtype=float)
    if np.any(np.diff(observation_times) < 0):
        raise ValidationError("observation_times must be sorted.")
    if observation_times[0] < 0 or observation_times[-1] > scenario.duration:
        raise ValidationError("observation_times must lie inside the simulation horizon.")

    names = system.variety_names
    n = len(names)
    name_to_index = {name: i for i, name in enumerate(names)}
    K = design.n_sites
    m = scenario.vectors_per_plant
    theta = np.zeros(n, dtype=float)
    for name, count in design.counts.items():
        theta[name_to_index[name]] = count / K

    alpha = np.asarray([v.transmission.acquisition_rate for v in system.varieties], dtype=float)
    beta = np.asarray([v.transmission.inoculation_rate for v in system.varieties], dtype=float)
    gamma = np.asarray([v.plant.latent_progression_rate for v in system.varieties], dtype=float)
    rho = np.asarray([v.plant.roguing_rate for v in system.varieties], dtype=float)

    sigma = system.vector.dispersal_rate
    omega = system.vector.mortality_rate
    clearance = system.pathogen.vector_clearance_rate
    q = omega + clearance
    psi = 1.0 / (sigma + q)
    F = float(m * K)

    initial_infectious = np.zeros(n, dtype=float)
    if scenario.inoculum.sites is not None:
        for site in scenario.inoculum.sites:
            if site >= K:
                raise ValidationError("Explicit inoculum site is outside the design.")
            initial_infectious[name_to_index[design.assignment[site]]] += 1.0 / K
    else:
        initial_infectious = theta * (scenario.inoculum.count / K)

    vector_inoculum = scenario.vector_inoculum
    if vector_inoculum.origin_weights is None:
        origin_weights = theta.copy()
    else:
        if len(vector_inoculum.origin_weights) != n:
            raise ValidationError(
                "vector_inoculum.origin_weights must have one value per system variety."
            )
        origin_weights = np.asarray(vector_inoculum.origin_weights, dtype=float)
        origin_weights = origin_weights / origin_weights.sum()
    initial_vectors = F * vector_inoculum.infectious_fraction * origin_weights

    y0 = np.concatenate(
        [
            np.zeros(n, dtype=float),
            initial_infectious,
            initial_vectors,
        ]
    )

    def rhs(_time: float, state: np.ndarray) -> np.ndarray:
        latent = state[:n]
        infectious = state[n : 2 * n]
        vector_origin = state[2 * n :]
        vector_total = float(vector_origin.sum())
        susceptible = np.maximum(theta - latent - infectious, 0.0)

        d_latent = (
            (sigma * psi / K) * beta * susceptible * vector_total
            - gamma * latent
        )
        d_infectious = gamma * latent - rho * infectious
        d_vector = alpha * (
            F * infectious
            - psi * (sigma * infectious * vector_total + q * vector_origin)
        ) - q * vector_origin
        return np.concatenate([d_latent, d_infectious, d_vector])

    solution = solve_ivp(
        rhs,
        (0.0, scenario.duration),
        y0,
        t_eval=observation_times,
        method="LSODA",
        rtol=1e-9,
        atol=1e-12,
    )
    if not solution.success:
        raise RuntimeError(solution.message)

    latent = solution.y[:n]
    infectious = solution.y[n : 2 * n]
    vector_origin = solution.y[2 * n :]

    final_yield = 0.0
    for index, variety in enumerate(system.varieties):
        final_yield += (
            variety.yield_model.healthy * (theta[index] - infectious[index, -1])
            + variety.yield_model.infected * infectious[index, -1]
        )

    result = MeanFieldResult(
        time=solution.t,
        latent=latent,
        infectious=infectious,
        viruliferous_by_origin=vector_origin,
        variety_names=names,
        proportions=theta,
        vector_burden=m,
        total_vectors=F,
        final_yield=float(final_yield),
        yield_unit=system.yield_unit,
    )
    return result

Deprecated compatibility alias for :func:assess_mean_field_consistency.

The historical name calibrate_kernel suggested biological parameter estimation. Cropmix 0.2 retains it only so older scripts keep working.

Source code in src/cropmix/calibration.py
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def calibrate_kernel(*args, **kwargs):
    """Deprecated compatibility alias for :func:`assess_mean_field_consistency`.

    The historical name ``calibrate_kernel`` suggested biological parameter
    estimation. Cropmix 0.2 retains it only so older scripts keep working.
    """
    import warnings
    warnings.warn(
        "calibrate_kernel() is deprecated: mean-field matching does not estimate "
        "a biological movement scale. Use assess_mean_field_consistency().",
        DeprecationWarning,
        stacklevel=2,
    )
    return assess_mean_field_consistency(*args, **kwargs)

Optimization

Search for a high-performing planting design with fixed variety counts.

Cropmix 0.1 uses swap-based simulated annealing. A swap proposal preserves the requested counts exactly. Candidate designs are evaluated with the same Monte Carlo seed block (common random numbers), reducing simulation noise in pairwise design comparisons.

The optimizer is heuristic: combinatorial design spaces become enormous even for modest fields, so global optimality is not claimed.

Source code in src/cropmix/optimization.py
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def optimize_mixture(
    field: Field,
    variety_counts: Mapping[str, int],
    system: CropMixSystem,
    scenario: Scenario,
    *,
    objective: Objective = "expected_yield",
    config: OptimizationConfig | None = None,
    initial_design: MixtureDesign | None = None,
    transmission_draw: TransmissionDraw | None = None,
) -> OptimizationResult:
    """Search for a high-performing planting design with fixed variety counts.

    Cropmix 0.1 uses swap-based simulated annealing.  A swap proposal preserves
    the requested counts exactly. Candidate designs are evaluated with the same
    Monte Carlo seed block (common random numbers), reducing simulation noise in
    pairwise design comparisons.

    The optimizer is heuristic: combinatorial design spaces become enormous
    even for modest fields, so global optimality is not claimed.
    """

    config = OptimizationConfig() if config is None else config
    counts = {str(name): int(value) for name, value in variety_counts.items()}
    if any(value < 0 for value in counts.values()):
        raise ValidationError("Variety counts cannot be negative.")
    if sum(counts.values()) != field.n_sites:
        raise ValidationError("Variety counts must sum to the number of field sites.")
    unknown = set(counts) - set(system.variety_names)
    if unknown:
        raise ValidationError(f"Counts contain varieties absent from system: {sorted(unknown)}")

    if initial_design is None:
        current = MixtureDesign.random(field, counts, seed=config.seed)
    else:
        if initial_design.field is not field and not np.array_equal(
            initial_design.field.coordinates, field.coordinates
        ):
            raise ValidationError("initial_design does not use the supplied field geometry.")
        if initial_design.counts != counts:
            raise ValidationError("initial_design counts do not match variety_counts.")
        current = initial_design

    rng = np.random.default_rng(config.seed)
    cache: dict[tuple[str, ...], tuple[float, SimulationResult]] = {}
    history_rows: list[dict[str, object]] = []

    def evaluate(design: MixtureDesign) -> tuple[float, SimulationResult, str]:
        key = _design_key(design)
        if key not in cache:
            result = simulate_mixture(
                design,
                system,
                scenario,
                n_runs=config.n_runs_per_candidate,
                seed=config.seed + 1_000_000,
                transmission_draw=transmission_draw,
                store_final_states=False,
            )
            score, objective_name = _objective_value(result, objective)
            cache[key] = (score, result)
        else:
            score, result = cache[key]
            objective_name = objective if isinstance(objective, str) else getattr(objective, "__name__", "custom")
        return score, result, str(objective_name)

    current_score, _, objective_name = evaluate(current)
    best = current
    best_score = current_score

    history_rows.append(
        {
            "iteration": 0,
            "current_score": current_score,
            "best_score": best_score,
            "accepted": True,
            "temperature": config.initial_temperature,
        }
    )

    for iteration in range(1, config.iterations + 1):
        pair = _random_heterotypic_pair(rng, current.assignment)
        if pair is None:
            break
        candidate = current.swapped(*pair)
        candidate_score, _, _ = evaluate(candidate)
        temperature = config.initial_temperature * (config.cooling_rate ** (iteration - 1))
        delta = candidate_score - current_score
        accept = delta >= 0 or rng.random() < exp(delta / max(temperature, 1e-15))

        if accept:
            current = candidate
            current_score = candidate_score

        if current_score > best_score:
            best = current
            best_score = current_score

        history_rows.append(
            {
                "iteration": iteration,
                "current_score": current_score,
                "best_score": best_score,
                "accepted": bool(accept),
                "temperature": temperature,
            }
        )

    final_result = simulate_mixture(
        best,
        system,
        scenario,
        n_runs=config.final_runs,
        seed=config.seed + config.final_seed_offset,
        transmission_draw=transmission_draw,
        store_final_states=True,
    )

    return OptimizationResult(
        best_design=best,
        best_score=float(best_score),
        final_result=final_result,
        history=pd.DataFrame(history_rows),
        objective_name=objective_name,
        variety_counts=counts,
        evaluations=len(cache),
    )
Source code in src/cropmix/optimization.py
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@dataclass(frozen=True)
class OptimizationConfig:
    iterations: int = 500
    n_runs_per_candidate: int = 20
    final_runs: int = 500
    initial_temperature: float = 0.5
    cooling_rate: float = 0.99
    seed: int = 12345
    final_seed_offset: int = 50_000_000

    def __post_init__(self) -> None:
        if self.iterations < 0:
            raise ValidationError("iterations cannot be negative.")
        if self.n_runs_per_candidate <= 0 or self.final_runs <= 0:
            raise ValidationError("Simulation replication counts must be positive.")
        if self.initial_temperature <= 0:
            raise ValidationError("initial_temperature must be positive.")
        if not 0 < self.cooling_rate <= 1:
            raise ValidationError("cooling_rate must lie in (0, 1].")
Source code in src/cropmix/optimization.py
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
@dataclass
class OptimizationResult:
    best_design: MixtureDesign
    best_score: float
    final_result: SimulationResult
    history: pd.DataFrame
    objective_name: str
    variety_counts: dict[str, int]
    evaluations: int

    def summary(self) -> pd.DataFrame:
        frame = self.final_result.summary().copy()
        frame.insert(0, "objective", self.objective_name)
        frame.insert(1, "optimization_score", self.best_score)
        frame.insert(2, "evaluations", self.evaluations)
        return frame

EpiPvr

One varying-access-period sub-assay.

Source code in src/cropmix/epipvr/models.py
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
@dataclass(frozen=True)
class AccessPeriodAssay:
    """One varying-access-period sub-assay."""

    duration: tuple[float, ...]
    tested: tuple[int, ...]
    infected: tuple[int, ...]

    def __post_init__(self) -> None:
        duration = tuple(float(x) for x in self.duration)
        tested = tuple(int(x) for x in self.tested)
        infected = tuple(int(x) for x in self.infected)
        if not (len(duration) == len(tested) == len(infected)) or not duration:
            raise ValidationError("duration, tested and infected must have equal non-zero lengths.")
        if any(x <= 0 for x in duration):
            raise ValidationError("Access durations must be positive.")
        if any(x <= 0 for x in tested):
            raise ValidationError("Numbers tested must be positive.")
        if any(x < 0 for x in infected):
            raise ValidationError("Numbers infected cannot be negative.")
        if any(i > n for i, n in zip(infected, tested, strict=False)):
            raise ValidationError("infected cannot exceed tested.")
        object.__setattr__(self, "duration", duration)
        object.__setattr__(self, "tested", tested)
        object.__setattr__(self, "infected", infected)

    def to_dataframe(self) -> pd.DataFrame:
        return pd.DataFrame(
            {"T_vec": self.duration, "R_vec": self.tested, "I_vec": self.infected}
        )

A complete EpiPvr SPT or PT access-period experiment.

Source code in src/cropmix/epipvr/models.py
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@dataclass(frozen=True)
class AccessPeriodExperiment:
    """A complete EpiPvr SPT or PT access-period experiment."""

    transmission_type: EpiPvrMode
    acquisition: AccessPeriodAssay
    inoculation: AccessPeriodAssay
    fixed_durations: np.ndarray
    vectors_per_plant: int
    latent: AccessPeriodAssay | None = None

    def __post_init__(self) -> None:
        mode = self.transmission_type.upper()
        if mode not in ("SPT", "PT"):
            raise ValidationError("transmission_type must be SPT or PT.")
        object.__setattr__(self, "transmission_type", mode)
        if self.vectors_per_plant <= 0:
            raise ValidationError("vectors_per_plant must be positive.")
        matrix = np.asarray(self.fixed_durations, dtype=float)
        expected_shape = (2, 2) if mode == "SPT" else (3, 3)
        if matrix.shape != expected_shape:
            raise ValidationError(
                f"fixed_durations must have shape {expected_shape} for {mode}."
            )
        if mode == "PT" and self.latent is None:
            raise ValidationError("PT experiments require a latent-period sub-assay.")
        if mode == "SPT" and self.latent is not None:
            raise ValidationError("SPT experiments must not contain a latent-period sub-assay.")
        object.__setattr__(self, "fixed_durations", matrix)

    @classmethod
    def spt(
        cls,
        *,
        acquisition: AccessPeriodAssay,
        inoculation: AccessPeriodAssay,
        fixed_inoculation_for_acquisition: float,
        fixed_acquisition_for_inoculation: float,
        vectors_per_plant: int,
    ) -> AccessPeriodExperiment:
        matrix = np.array(
            [
                [-1.0, float(fixed_inoculation_for_acquisition)],
                [float(fixed_acquisition_for_inoculation), -1.0],
            ]
        )
        return cls(
            transmission_type="SPT",
            acquisition=acquisition,
            inoculation=inoculation,
            fixed_durations=matrix,
            vectors_per_plant=vectors_per_plant,
        )

    @classmethod
    def pt(
        cls,
        *,
        acquisition: AccessPeriodAssay,
        latent: AccessPeriodAssay,
        inoculation: AccessPeriodAssay,
        fixed_when_acquisition_varies: tuple[float, float],
        fixed_when_latency_varies: tuple[float, float],
        fixed_when_inoculation_varies: tuple[float, float],
        vectors_per_plant: int,
    ) -> AccessPeriodExperiment:
        # Columns are AAP, LAP, IAP; rows correspond to the varying component.
        matrix = np.array(
            [
                [-1.0, fixed_when_acquisition_varies[0], fixed_when_acquisition_varies[1]],
                [fixed_when_latency_varies[0], -1.0, fixed_when_latency_varies[1]],
                [fixed_when_inoculation_varies[0], fixed_when_inoculation_varies[1], -1.0],
            ],
            dtype=float,
        )
        return cls(
            transmission_type="PT",
            acquisition=acquisition,
            latent=latent,
            inoculation=inoculation,
            fixed_durations=matrix,
            vectors_per_plant=vectors_per_plant,
        )

    def write_bundle(self, directory: Path) -> None:
        directory.mkdir(parents=True, exist_ok=True)
        self.acquisition.to_dataframe().to_csv(directory / "AAP.csv", index=False)
        self.inoculation.to_dataframe().to_csv(directory / "IAP.csv", index=False)
        if self.latent is not None:
            self.latent.to_dataframe().to_csv(directory / "LAP.csv", index=False)
        pd.DataFrame(self.fixed_durations).to_csv(directory / "durations.csv", index=False, header=False)
        pd.DataFrame(
            [
                {
                    "transmission_type": self.transmission_type,
                    "vectors_per_plant": self.vectors_per_plant,
                }
            ]
        ).to_csv(directory / "metadata.csv", index=False)

Call EpiPvr through Rscript while keeping the public API Python-only.

Source code in src/cropmix/epipvr/backend.py
 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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
class EpiPvrBackend:
    """Call EpiPvr through `Rscript` while keeping the public API Python-only."""

    def __init__(self, rscript: str = "Rscript") -> None:
        self.rscript = rscript

    def check_installation(self) -> dict[str, object]:
        executable = shutil.which(self.rscript)
        if executable is None:
            return {"rscript": False, "epipvr": False, "message": "Rscript not found on PATH."}
        command = [
            executable,
            "-e",
            "cat(ifelse(requireNamespace('EpiPvr', quietly=TRUE), as.character(packageVersion('EpiPvr')), 'MISSING'))",
        ]
        completed = subprocess.run(command, capture_output=True, text=True, check=False)
        version = completed.stdout.strip()
        installed = completed.returncode == 0 and version not in {"", "MISSING"}
        return {
            "rscript": True,
            "epipvr": installed,
            "version": version if installed else None,
            "message": completed.stderr.strip(),
        }

    def require_installation(self) -> None:
        status = self.check_installation()
        if not status["rscript"]:
            raise EpiPvrError("Rscript was not found. Install R and add Rscript to PATH.")
        if not status["epipvr"]:
            raise EpiPvrError(
                "EpiPvr is not installed in the R library visible to Rscript. "
                "Run in R once: install.packages('EpiPvr')."
            )

    def fit(
        self,
        experiment: AccessPeriodExperiment,
        *,
        options: EpiPvrFitOptions | None = None,
    ) -> EpiPvrFit:
        """Fit EpiPvr without exposing an R object to the Python caller."""
        self.require_installation()
        options = EpiPvrFitOptions() if options is None else options

        bridge = files("cropmix.epipvr").joinpath("resources", "fit_bridge.R")
        with tempfile.TemporaryDirectory(prefix="cropmix_epipvr_") as temporary:
            workdir = Path(temporary)
            input_dir = workdir / "input"
            output_dir = workdir / "output"
            experiment.write_bundle(input_dir)
            output_dir.mkdir()

            with as_file(bridge) as bridge_path:
                command = [
                    self.rscript,
                    str(bridge_path),
                    str(input_dir),
                    str(output_dir),
                    str(options.survival_upper_days),
                    str(options.d_num_pts_pd),
                    str(options.warmup),
                    str(options.iterations),
                    str(options.chains),
                    str(options.parallel),
                    str(options.seed),
                ]
                completed = subprocess.run(command, capture_output=True, text=True, check=False)

            if completed.returncode != 0:
                raise EpiPvrError(
                    "EpiPvr fitting failed.\n\nSTDOUT:\n"
                    + completed.stdout
                    + "\n\nSTDERR:\n"
                    + completed.stderr
                )

            posterior = pd.read_csv(output_dir / "posterior.csv")
            summary = pd.read_csv(output_dir / "summary.csv")
            diagnostics_frame = pd.read_csv(output_dir / "diagnostics.csv")
            diagnostics = dict(
                zip(diagnostics_frame["key"].astype(str), diagnostics_frame["value"].astype(str), strict=False)
            )
            bayes_path = output_dir / "bayes_r2.csv"
            bayes_r2 = pd.read_csv(bayes_path) if bayes_path.exists() else pd.DataFrame()

        return EpiPvrFit(
            transmission_type=experiment.transmission_type,
            posterior_hourly=posterior,
            summary_table=summary,
            diagnostics=diagnostics,
            bayes_r2=bayes_r2,
            stdout=completed.stdout,
            stderr=completed.stderr,
        )

    def epidemic_probability(
        self,
        *,
        vectors_per_plant: int,
        virus_parameters_per_day: tuple[float, float, float],
        local_parameters: LocalEpidemicParameters,
        initial_interval: float = 0.1,
    ) -> EpidemicProbabilityResult:
        """Call EpiPvr's branching-process epidemic-probability calculator."""
        self.require_installation()
        if vectors_per_plant <= 0:
            raise EpiPvrError("vectors_per_plant must be positive.")

        bridge = files("cropmix.epipvr").joinpath("resources", "epidemic_bridge.R")
        with tempfile.TemporaryDirectory(prefix="cropmix_epipvr_bp_") as temporary:
            output = Path(temporary) / "probabilities.csv"
            with as_file(bridge) as bridge_path:
                command = [
                    self.rscript,
                    str(bridge_path),
                    str(output),
                    str(vectors_per_plant),
                    str(initial_interval),
                    *(str(float(x)) for x in virus_parameters_per_day),
                    str(local_parameters.dispersal_rate),
                    str(local_parameters.roguing_rate),
                    str(local_parameters.harvest_rate),
                    str(local_parameters.vector_mortality_rate),
                    str(local_parameters.plant_latent_progression_rate),
                ]
                completed = subprocess.run(command, capture_output=True, text=True, check=False)
            if completed.returncode != 0:
                raise EpiPvrError(
                    "EpiPvr epidemic-probability calculation failed.\n"
                    + completed.stdout
                    + "\n"
                    + completed.stderr
                )
            probabilities = pd.read_csv(output)["probability"].to_numpy(float)

        return EpidemicProbabilityResult(
            probabilities=np.asarray(probabilities, dtype=float),
            vectors_per_plant=vectors_per_plant,
        )

epidemic_probability(*, vectors_per_plant, virus_parameters_per_day, local_parameters, initial_interval=0.1)

Call EpiPvr's branching-process epidemic-probability calculator.

Source code in src/cropmix/epipvr/backend.py
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
def epidemic_probability(
    self,
    *,
    vectors_per_plant: int,
    virus_parameters_per_day: tuple[float, float, float],
    local_parameters: LocalEpidemicParameters,
    initial_interval: float = 0.1,
) -> EpidemicProbabilityResult:
    """Call EpiPvr's branching-process epidemic-probability calculator."""
    self.require_installation()
    if vectors_per_plant <= 0:
        raise EpiPvrError("vectors_per_plant must be positive.")

    bridge = files("cropmix.epipvr").joinpath("resources", "epidemic_bridge.R")
    with tempfile.TemporaryDirectory(prefix="cropmix_epipvr_bp_") as temporary:
        output = Path(temporary) / "probabilities.csv"
        with as_file(bridge) as bridge_path:
            command = [
                self.rscript,
                str(bridge_path),
                str(output),
                str(vectors_per_plant),
                str(initial_interval),
                *(str(float(x)) for x in virus_parameters_per_day),
                str(local_parameters.dispersal_rate),
                str(local_parameters.roguing_rate),
                str(local_parameters.harvest_rate),
                str(local_parameters.vector_mortality_rate),
                str(local_parameters.plant_latent_progression_rate),
            ]
            completed = subprocess.run(command, capture_output=True, text=True, check=False)
        if completed.returncode != 0:
            raise EpiPvrError(
                "EpiPvr epidemic-probability calculation failed.\n"
                + completed.stdout
                + "\n"
                + completed.stderr
            )
        probabilities = pd.read_csv(output)["probability"].to_numpy(float)

    return EpidemicProbabilityResult(
        probabilities=np.asarray(probabilities, dtype=float),
        vectors_per_plant=vectors_per_plant,
    )

fit(experiment, *, options=None)

Fit EpiPvr without exposing an R object to the Python caller.

Source code in src/cropmix/epipvr/backend.py
 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
def fit(
    self,
    experiment: AccessPeriodExperiment,
    *,
    options: EpiPvrFitOptions | None = None,
) -> EpiPvrFit:
    """Fit EpiPvr without exposing an R object to the Python caller."""
    self.require_installation()
    options = EpiPvrFitOptions() if options is None else options

    bridge = files("cropmix.epipvr").joinpath("resources", "fit_bridge.R")
    with tempfile.TemporaryDirectory(prefix="cropmix_epipvr_") as temporary:
        workdir = Path(temporary)
        input_dir = workdir / "input"
        output_dir = workdir / "output"
        experiment.write_bundle(input_dir)
        output_dir.mkdir()

        with as_file(bridge) as bridge_path:
            command = [
                self.rscript,
                str(bridge_path),
                str(input_dir),
                str(output_dir),
                str(options.survival_upper_days),
                str(options.d_num_pts_pd),
                str(options.warmup),
                str(options.iterations),
                str(options.chains),
                str(options.parallel),
                str(options.seed),
            ]
            completed = subprocess.run(command, capture_output=True, text=True, check=False)

        if completed.returncode != 0:
            raise EpiPvrError(
                "EpiPvr fitting failed.\n\nSTDOUT:\n"
                + completed.stdout
                + "\n\nSTDERR:\n"
                + completed.stderr
            )

        posterior = pd.read_csv(output_dir / "posterior.csv")
        summary = pd.read_csv(output_dir / "summary.csv")
        diagnostics_frame = pd.read_csv(output_dir / "diagnostics.csv")
        diagnostics = dict(
            zip(diagnostics_frame["key"].astype(str), diagnostics_frame["value"].astype(str), strict=False)
        )
        bayes_path = output_dir / "bayes_r2.csv"
        bayes_r2 = pd.read_csv(bayes_path) if bayes_path.exists() else pd.DataFrame()

    return EpiPvrFit(
        transmission_type=experiment.transmission_type,
        posterior_hourly=posterior,
        summary_table=summary,
        diagnostics=diagnostics,
        bayes_r2=bayes_r2,
        stdout=completed.stdout,
        stderr=completed.stderr,
    )

Python-side EpiPvr posterior and diagnostics.

Source code in src/cropmix/epipvr/models.py
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
@dataclass
class EpiPvrFit:
    """Python-side EpiPvr posterior and diagnostics."""

    transmission_type: EpiPvrMode
    posterior_hourly: pd.DataFrame
    summary_table: pd.DataFrame
    diagnostics: dict[str, str]
    bayes_r2: pd.DataFrame
    stdout: str = ""
    stderr: str = ""

    def posterior(self, *, unit: Literal["per_hour", "per_day"] = "per_day") -> pd.DataFrame:
        frame = self.posterior_hourly.copy()
        rate_columns = [
            column
            for column in (
                "acquisition_rate",
                "inoculation_rate",
                "vector_clearance_rate",
                "vector_latent_progression_rate",
            )
            if column in frame.columns
        ]
        if unit == "per_day":
            frame[rate_columns] = frame[rate_columns] * 24.0
        elif unit != "per_hour":
            raise ValidationError("unit must be 'per_hour' or 'per_day'.")
        return frame

    def parameter_summary(self, *, unit: Literal["per_hour", "per_day"] = "per_day") -> pd.DataFrame:
        posterior = self.posterior(unit=unit)
        rate_columns = [c for c in posterior.columns if c.endswith("_rate")]
        rows = []
        for column in rate_columns:
            values = posterior[column].to_numpy(float)
            rows.append(
                {
                    "parameter": column,
                    "mean": float(np.mean(values)),
                    "median": float(np.median(values)),
                    "sd": float(np.std(values, ddof=1)),
                    "q05": float(np.quantile(values, 0.05)),
                    "q95": float(np.quantile(values, 0.95)),
                    "unit": unit,
                }
            )
        return pd.DataFrame(rows)

    def median_host_transmission(self) -> HostTransmission:
        posterior = self.posterior(unit="per_day")
        return HostTransmission(
            acquisition_rate=float(posterior["acquisition_rate"].median()),
            inoculation_rate=float(posterior["inoculation_rate"].median()),
        )

    def median_pathogen_parameters(self) -> PathogenParameters:
        posterior = self.posterior(unit="per_day")
        latent = None
        if "vector_latent_progression_rate" in posterior:
            latent = float(posterior["vector_latent_progression_rate"].median())
        return PathogenParameters(
            vector_clearance_rate=float(posterior["vector_clearance_rate"].median()),
            transmission_mode=self.transmission_type,
            vector_latent_progression_rate=latent,
        )

    def convergence_report(
        self,
        *,
        max_rhat: float = 1.01,
        min_ess_per_chain: float = 100.0,
    ) -> dict[str, object]:
        virus_names = {"al[1]", "be[1]", "mu[1]", "lat[1]"}
        summary = self.summary_table
        if "variable" in summary.columns:
            virus_summary = summary[summary["variable"].astype(str).isin(virus_names)]
        else:
            virus_summary = summary
        rhat_ok = True
        ess_ok = True
        max_seen_rhat = np.nan
        min_seen_ess = np.nan
        if "rhat" in virus_summary.columns and len(virus_summary):
            values = pd.to_numeric(virus_summary["rhat"], errors="coerce").dropna()
            if len(values):
                max_seen_rhat = float(values.max())
                rhat_ok = max_seen_rhat <= max_rhat
        if "ess_bulk" in virus_summary.columns and len(virus_summary):
            values = pd.to_numeric(virus_summary["ess_bulk"], errors="coerce").dropna()
            if len(values):
                min_seen_ess = float(values.min())
                # Chain count can be inferred from posterior metadata when exported.
                chains = int(self.posterior_hourly.get("chain", pd.Series([1])).nunique())
                ess_ok = min_seen_ess >= min_ess_per_chain * max(chains, 1)

        divergent = int(float(self.diagnostics.get("divergent_transitions", "0")))
        treedepth_value = self.diagnostics.get("max_treedepth_exceeded", "FALSE").upper()
        treedepth_ok = treedepth_value not in {"TRUE", "T", "1"}
        return {
            "usable": bool(rhat_ok and ess_ok and divergent == 0 and treedepth_ok),
            "rhat_ok": rhat_ok,
            "ess_ok": ess_ok,
            "divergent_transitions": divergent,
            "treedepth_ok": treedepth_ok,
            "max_rhat": max_seen_rhat,
            "min_ess_bulk": min_seen_ess,
        }

    def require_usable(self, **kwargs) -> None:
        report = self.convergence_report(**kwargs)
        if not report["usable"]:
            raise EpiPvrError(f"EpiPvr diagnostics are not satisfactory: {report}")