Skip to content

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:

  • a single entry dict
  • a list of entry dicts
  • a file path (str or Path) to a .json or .csv file containing the schedule
required
start_date str | date | None

Simulation start date in "YYYY-MM-DD" format or as a datetime.date object. Required when any when value is a date string.

None

Raises:

Type Description
ValueError

If the source path has an unsupported suffix.

ValueError

If any schedule entry omits the required who or where field. Use "*" explicitly to target all states or all nodes.

ValueError

If any what value names an intervention class that has not been registered with Campaign.register.

ValueError

If when values mix integer ticks and date strings (across entries or within a single list).

ValueError

If date-valued when entries are present but start_date is not provided.

ValueError

If a when date is earlier than start_date.

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. entry.tick is the absolute simulation tick at which to dispatch (or None for every-tick firing); entry.what must be a registered intervention class name.

required

Raises:

Type Description
TypeError

If entry is not a ScheduledEntry.

ValueError

If entry.what is not a registered intervention.

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 Intervention to register. The class name becomes the value expected in what fields.

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; None means all states.

required
where list[int] | None

Target node IDs; None means all nodes.

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 None (all).

where list[int] | None

Target node IDs, or None (all).

params dict[str, Any]

Arbitrary parameters forwarded to execute.

notes str

Free-text annotation forwarded to execute.

tick int | None

Tick on which to fire, or None to fire on every tick.