campaign

Simple campaign builder for EMOD simulations.

Import this module, add valid campaign events via add, and write the campaign file with save.

add(event, note=None)

Add a complete campaign event to the campaign builder.

The event is assumed to be valid and is not validated here.

Parameters:

Name	Type	Description	Default
event		A complete campaign event object. It must support finalize() and dict-style key assignment.	required
note	str	An optional human-readable note added to the event inside the output campaign.json file.	None

Source code in emod_api/campaign.py

def add(event, note: str = None):
    """Add a complete campaign event to the campaign builder.

    The event is assumed to be valid and is not validated here.

    Args:
        event: A complete campaign event object. It must support
            ``finalize()`` and dict-style key assignment.
        note: An optional human-readable note added to the event
            inside the output ``campaign.json`` file.
    """
    event.finalize()
    if note is not None:
        event["Note"] = note
    campaign_dict["Events"].append(event)

get_recv_trigger(trigger, old=use_old_adhoc_handling)

Register an individual-level event as listened to.

Tracks which individual events are used throughout the simulation so that validate_custom_individual_events can validate that every listened-to event has a corresponding broadcast.

Parameters:

Name	Type	Description	Default
trigger		The individual event name string.	required
old		Unused. Kept for backwards compatibility.	use_old_adhoc_handling

Returns:

Type	Description
	The event name, unchanged.

Source code in emod_api/campaign.py

def get_recv_trigger(trigger, old=use_old_adhoc_handling):
    """Register an individual-level event as listened to.

    Tracks which individual events are used throughout the simulation
    so that ``validate_custom_individual_events`` can validate that every
    listened-to event has a corresponding broadcast.

    Args:
        trigger: The individual event name string.
        old: Unused. Kept for backwards compatibility.

    Returns:
        The event name, unchanged.
    """
    if not trigger:
        raise ValueError("Event name must not be None or empty.")
    individual_events_listened.append(trigger)
    return trigger

get_schema()

Return the loaded schema JSON dictionary.

Returns:

Type	Description
	The parsed schema dictionary, or None if set_schema has
	not been called.

Source code in emod_api/campaign.py

def get_schema():
    """Return the loaded schema JSON dictionary.

    Returns:
        The parsed schema dictionary, or ``None`` if ``set_schema`` has
        not been called.
    """
    return _schema_json

get_send_trigger(trigger, old=use_old_adhoc_handling)

Register an individual-level event as broadcast.

Parameters:

Name	Type	Description	Default
trigger		The individual event name string.	required
old		Unused. Kept for backwards compatibility.	use_old_adhoc_handling

Returns:

Type	Description
	The event name, unchanged.

Source code in emod_api/campaign.py

def get_send_trigger(trigger, old=use_old_adhoc_handling):
    """Register an individual-level event as broadcast.

    Args:
        trigger: The individual event name string.
        old: Unused. Kept for backwards compatibility.

    Returns:
        The event name, unchanged.
    """
    if not trigger:
        raise ValueError("Event name must not be None or empty.")
    individual_events_broadcast.append(trigger)
    return trigger

reset()

Reset all campaign state to defaults.

Clears accumulated events, signal tracking lists, event mappings, and the schema cache.

Source code in emod_api/campaign.py

def reset():
    """Reset all campaign state to defaults.

    Clears accumulated events, signal tracking lists, event mappings,
    and the schema cache.
    """
    campaign_dict["Events"].clear()

    individual_events_listened.clear()
    individual_events_broadcast.clear()
    node_events_broadcast.clear()
    node_events_listened.clear()
    coordinator_events_broadcast.clear()
    coordinator_events_listened.clear()
    implicits.clear()
    individual_builtin_events.clear()
    node_builtin_events.clear()
    coordinator_builtin_events.clear()
    s2c.clear_schema_cache()

save(filename='campaign.json')

Save the accumulated campaign events to a JSON file.

Parameters:

Name	Type	Description	Default
filename	str	Output file path.	'campaign.json'

Returns:

Type	Description
	The filename that was written.

Source code in emod_api/campaign.py

def save(filename: str = "campaign.json"):
    """Save the accumulated campaign events to a JSON file.

    Args:
        filename: Output file path.

    Returns:
        The filename that was written.
    """
    with open(filename, "w") as camp_file:
        json.dump(campaign_dict, camp_file, sort_keys=True, indent=4)

    return filename

set_broadcast_coordinator_event(event)

Register a coordinator-level event as broadcast.

Tracks which coordinator events are used throughout the simulation so that validate_custom_coordinator_events can validate that every broadcast event has something listening to it.

Parameters:

Name	Type	Description	Default
event	str	The coordinator event name string.	required

Returns:

Type	Description
str	The event name, unchanged.

Source code in emod_api/campaign.py

def set_broadcast_coordinator_event(event: str) -> str:
    """Register a coordinator-level event as broadcast.

    Tracks which coordinator events are used throughout the simulation
    so that ``validate_custom_coordinator_events`` can validate that every
    broadcast event has something listening to it.

    Args:
        event: The coordinator event name string.

    Returns:
        The event name, unchanged.
    """
    if not event:
        raise ValueError("Event name must not be None or empty.")
    coordinator_events_broadcast.append(event)
    return event

set_broadcast_node_event(event)

Register a node-level event as broadcast.

Tracks which node events are used throughout the simulation so that validate_custom_node_events can validate that every broadcast event has something listening to it.

Parameters:

Name	Type	Description	Default
event	str	The node event name string.	required

Returns:

Type	Description
str	The event name, unchanged.

Source code in emod_api/campaign.py

def set_broadcast_node_event(event: str) -> str:
    """Register a node-level event as broadcast.

    Tracks which node events are used throughout the simulation so
    that ``validate_custom_node_events`` can validate that every broadcast
    event has something listening to it.

    Args:
        event: The node event name string.

    Returns:
        The event name, unchanged.
    """
    if not event:
        raise ValueError("Event name must not be None or empty.")
    node_events_broadcast.append(event)
    return event

set_listened_coordinator_event(event)

Register a coordinator-level event as listened to.

Tracks which coordinator events are used throughout the simulation so that validate_custom_coordinator_events can validate that every listened-to event has a corresponding broadcast.

Parameters:

Name	Type	Description	Default
event	str	The coordinator event name string.	required

Returns:

Type	Description
str	The event name, unchanged.

Source code in emod_api/campaign.py

def set_listened_coordinator_event(event: str) -> str:
    """Register a coordinator-level event as listened to.

    Tracks which coordinator events are used throughout the simulation
    so that ``validate_custom_coordinator_events`` can validate that every
    listened-to event has a corresponding broadcast.

    Args:
        event: The coordinator event name string.

    Returns:
        The event name, unchanged.
    """
    if not event:
        raise ValueError("Event name must not be None or empty.")
    coordinator_events_listened.append(event)
    return event

set_listened_node_event(event)

Register a node-level event as listened to.

Tracks which node events are used throughout the simulation so that validate_custom_node_events can validate that every listened-to event has a corresponding broadcast.

Parameters:

Name	Type	Description	Default
event	str	The node event name string.	required

Returns:

Type	Description
str	The event name, unchanged.

Source code in emod_api/campaign.py

def set_listened_node_event(event: str) -> str:
    """Register a node-level event as listened to.

    Tracks which node events are used throughout the simulation so
    that ``validate_custom_node_events`` can validate that every listened-to
    event has a corresponding broadcast.

    Args:
        event: The node event name string.

    Returns:
        The event name, unchanged.
    """
    if not event:
        raise ValueError("Event name must not be None or empty.")
    node_events_listened.append(event)
    return event

set_schema(schema_path_in)

Set the schema file path and reset all campaign state.

This is essentially the "start building a campaign" entry point. It clears any previously accumulated events and loads the new schema. Also extracts built-in event lists for individual, node, and coordinator levels by recursively searching for ReportEventRecorder, ReportEventRecorderNode, and ReportEventRecorderCoordinator in the schema.

Parameters:

Name	Type	Description	Default
schema_path_in		Path to a schema.json file.	required

Source code in emod_api/campaign.py

def set_schema(schema_path_in):
    """Set the schema file path and reset all campaign state.

    This is essentially the "start building a campaign" entry point.
    It clears any previously accumulated events and loads the new schema.
    Also extracts built-in event lists for individual, node, and
    coordinator levels by recursively searching for
    ``ReportEventRecorder``, ``ReportEventRecorderNode``, and
    ``ReportEventRecorderCoordinator`` in the schema.

    Args:
        schema_path_in: Path to a ``schema.json`` file.
    """
    reset()
    global schema_path, _schema_json

    schema_path = schema_path_in
    with open(schema_path_in) as schema_file:
        _schema_json = json.load(schema_file)

    found = _find_builtin_events(_schema_json, "ReportEventRecorder", "Report_Event_Recorder_Events")
    if found:
        individual_builtin_events.extend(found)

    found = _find_builtin_events(_schema_json, "ReportEventRecorderNode", "Report_Node_Event_Recorder_Events")
    if found:
        node_builtin_events.extend(found)

    found = _find_builtin_events(_schema_json, "ReportEventRecorderCoordinator", "Report_Coordinator_Event_Recorder_Events")
    if found:
        coordinator_builtin_events.extend(found)

validate_custom_coordinator_events()

Validate and return deduplicated custom coordinator-level events.

Returns:

Type	Description
	A list of unique coordinator event name strings that are broadcast
	in the campaign.

Raises:

Type	Description
ValueError	If any coordinator events are listened to but never broadcast.

Source code in emod_api/campaign.py

def validate_custom_coordinator_events():
    """Validate and return deduplicated custom coordinator-level events.

    Returns:
        A list of unique coordinator event name strings that are broadcast
        in the campaign.

    Raises:
        ValueError: If any coordinator events are listened to but
            never broadcast.
    """
    return _validate_custom_events(coordinator_events_listened, coordinator_events_broadcast, coordinator_builtin_events, "coordinator")

validate_custom_individual_events()

Validate and return deduplicated custom individual-level events.

Returns:

Type	Description
	A list of unique individual event name strings that are broadcast
	in the campaign.

Raises:

Type	Description
ValueError	If any individual events are listened to but never broadcast.

Source code in emod_api/campaign.py

def validate_custom_individual_events():
    """Validate and return deduplicated custom individual-level events.

    Returns:
        A list of unique individual event name strings that are broadcast
        in the campaign.

    Raises:
        ValueError: If any individual events are listened to but
            never broadcast.
    """
    return _validate_custom_events(individual_events_listened, individual_events_broadcast, individual_builtin_events, "individual")

validate_custom_node_events()

Validate and return deduplicated custom node-level events.

Returns:

Type	Description
	A list of unique node event name strings that are broadcast
	in the campaign.

Raises:

Type	Description
ValueError	If any node events are listened to but never broadcast.

Source code in emod_api/campaign.py

def validate_custom_node_events():
    """Validate and return deduplicated custom node-level events.

    Returns:
        A list of unique node event name strings that are broadcast
        in the campaign.

    Raises:
        ValueError: If any node events are listened to but
            never broadcast.
    """
    return _validate_custom_events(node_events_listened, node_events_broadcast, node_builtin_events, "node")