laser.cohorts.campaign
campaign
Campaign-based intervention scheduling for cohort-based simulation.
Provides Campaign, a component that loads a schedule from a dict, list,
JSON file, or CSV file and dispatches named interventions at the right ticks,
nodes, and compartment states.
Each schedule entry specifies:
- who (required):
"*"(all states) or a list of state names, e.g.["S", "R"] - what (required): name of the registered intervention class
- when (optional, defaults to
"*"):"*"(every tick), an integer tick, a list of integer ticks, a"YYYY-MM-DD"date, or a list of date strings - where (required):
"*"(all nodes), a single node ID, or a list of node IDs - parameters (optional): arbitrary
{key: value}pairs forwarded to the intervention - notes (optional): free-text string forwarded to the intervention
who and where are required for every entry; omitting them raises
ValueError. Use "*" explicitly to target all states or all nodes.
Date-based when values require a start_date argument on the Campaign.
Integer ticks and date strings cannot be mixed in the same schedule.
Campaign
Campaign(model, source, start_date=None)
Intervention scheduling component.
Loads a campaign schedule and fires named interventions at the specified ticks, nodes, and compartment states.
Intervention classes are registered on the class-level registry with
Campaign.register and looked up by name at each tick.
Example
Campaign.register(Vaccination) schedule = [ ... {"who": "", "what": "Vaccination", "when": 30, ... "where": [0, 1], "parameters": {"coverage": 0.8}, "notes": ""}, ... {"who": ["S"], "what": "Vaccination", "when": [60, 90, 120], ... "where": "", "parameters": {"coverage": 0.6}, "notes": "boosters"}, ... ] campaign = Campaign(model, schedule) model.components = [..., campaign]
Initialize the Campaign component.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
Model
|
The parent model instance. |
required |
source
|
dict | list | str | Path
|
Campaign schedule. One of:
|
required |
start_date
|
str | date | None
|
Simulation start date in
|
None
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If the source path has an unsupported suffix. |
ValueError
|
If any schedule entry omits the required |
ValueError
|
If any |
ValueError
|
If |
ValueError
|
If date-valued |
ValueError
|
If a |
properties
property
properties
Return node properties required by this component and its interventions.
Iterates over the unique intervention class names referenced in the
schedule, instantiates each registered class with the model, and
accumulates their properties declarations into a set so duplicates
are dropped automatically.
Returns:
| Type | Description |
|---|---|
'list[PropertyType]'
|
list[PropertyType]: Union of all property declarations from interventions used in this campaign's schedule. Order is unspecified. |
states
property
states
Return compartment states required by this component and its interventions.
Iterates over the unique intervention class names referenced in the
schedule, instantiates each registered class with the model, and
accumulates their states declarations into a set so duplicates are
dropped automatically.
Returns:
| Type | Description |
|---|---|
'list[str]'
|
list[str]: Union of all state declarations from interventions used in this campaign's schedule. Order is unspecified. |
add_entry
add_entry(entry)
Schedule a ScheduledEntry for dispatch later in the running simulation.
Designed to be called at simulation time — typically from inside another
intervention's execute() to add a follow-up dispatch. The entry is
routed to self._every_tick if entry.tick is None, otherwise to
the appropriate self._at_tick bucket so that the next call to
Campaign.step will pick it up.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
entry
|
ScheduledEntry
|
Fully-formed entry. |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
If |
ValueError
|
If |
end_step
end_step(tick)
No-op end-of-step hook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tick
|
int
|
Current simulation tick (0-indexed). |
required |
register
classmethod
register(intervention_cls)
Register an intervention class using its __name__ as the schedule key.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
intervention_cls
|
type
|
Subclass of |
required |
setup
setup()
Build the tick-indexed schedule from parsed entries.
start_step
start_step(tick)
No-op start-of-step hook.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tick
|
int
|
Current simulation tick (0-indexed). |
required |
step
step(tick)
Dispatch all interventions scheduled for this tick.
Fires every-tick interventions first, then tick-specific ones, in
the order they appear in the schedule. All what names are
validated against the registry at construction time, so the lookup
here is unconditional.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tick
|
int
|
Current simulation tick (0-indexed). |
required |
Intervention
Intervention(model)
Base class for all campaign interventions.
Subclass this, implement execute, and register with
Campaign.register(MyClass).
Example
class Vaccination(Intervention): ... def execute(self, tick, who, where, params, notes): ... coverage = params.get("coverage", 0.5) ... # move fraction of S to R in the target nodes ... pass Campaign.register("Vaccination", Vaccination)
Initialize the Intervention.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
Model
|
The parent model instance. |
required |
properties
property
properties
Return node properties required by this intervention.
Override in subclasses to declare per-tick, per-node arrays for
recording intervention outputs. These properties are surfaced through
Campaign.properties so the model allocates them before the
simulation runs.
Returns:
| Type | Description |
|---|---|
list[PropertyType]
|
list[PropertyType]: Empty list; subclasses override as needed. |
states
property
states
Return compartment states required by this intervention.
Override in subclasses to declare any new states the intervention
needs (e.g. ["V"] for a vaccination intervention). These states
are surfaced through Campaign.states so the model allocates them
before the simulation runs.
Returns:
| Type | Description |
|---|---|
list[str]
|
list[str]: Empty list; subclasses override as needed. |
execute
execute(tick, who, where, params, notes)
Execute the intervention.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
tick
|
int
|
Current simulation tick (0-indexed). |
required |
who
|
list[str] | None
|
Target state names; |
required |
where
|
list[int] | None
|
Target node IDs; |
required |
params
|
dict[str, Any]
|
Arbitrary parameters from the schedule entry. |
required |
notes
|
str
|
Free-text annotation from the schedule entry. |
required |
Raises:
| Type | Description |
|---|---|
NotImplementedError
|
Subclasses must implement this method. |
ScheduledEntry
dataclass
ScheduledEntry(what, who, where, params, notes, tick)
A single fully-parsed schedule entry — one intervention dispatch.
Each instance corresponds to one (tick, intervention) pair after the
raw schedule has been validated and expanded — when lists are flattened
into individual entries, who/where are normalised to lists or
None, and date strings are converted to integer tick offsets.
Attributes:
| Name | Type | Description |
|---|---|---|
what |
str
|
Name of the registered intervention class. |
who |
list[str] | None
|
Target compartment states, or |
where |
list[int] | None
|
Target node IDs, or |
params |
dict[str, Any]
|
Arbitrary parameters forwarded to |
notes |
str
|
Free-text annotation forwarded to |
tick |
int | None
|
Tick on which to fire, or |