Skip to content

Main Run Function

Module to run the AMR Hub ABM simulation.

run_steps(simulation, plot_path, *, record, figures=None, trajectory=False)

Run the simulation loop until completion, optionally plotting live.

Parameters:

Name Type Description Default
simulation Simulation

The simulation instance to run.

required
plot_path Path | None

Directory to save plots if plotting is enabled, otherwise None.

required
record bool

Whether to record agent states to a CSV file at the end of simulation.

required
figures list | None

List of Matplotlib figure objects for live plotting, by default None

None
trajectory bool

Whether to plot agent trajectories at the end of simulation, by default False

False
Source code in src/amr_hub_abm/run.py
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
def run_steps(
    simulation: Simulation,
    plot_path: Path | None,
    *,
    record: bool,
    figures: list | None = None,
    trajectory: bool = False,  # NEW
) -> None:
    """
    Run the simulation loop until completion, optionally plotting live.

    Parameters
    ----------
    simulation : Simulation
        The simulation instance to run.
    plot_path : Path | None
        Directory to save plots if plotting is enabled, otherwise None.
    record : bool
        Whether to record agent states to a CSV file at the end of simulation.
    figures : list | None, optional
        List of Matplotlib figure objects for live plotting, by default None
    trajectory : bool, optional
        Whether to plot agent trajectories at the end of simulation, by default False

    """
    while simulation.time < simulation.total_simulation_time:
        simulation.step(plot_path=plot_path, record=record)

        if figures is not None and simulation.time % 100 == 0:
            simulation.plot_live(figures, trajectory=trajectory)

simulate(*, plot=False, record=False, live=False, plot_trajectory=False, use_gpu=False, seed_infections=False)

Simulate the AMR Hub ABM based on a configuration file.

Parameters:

Name Type Description Default
plot bool

Whether to save plots of the simulation at regular intervals, by default False

False
record bool

Whether to record agent states to a CSV file, by default False

False
live bool

Whether to display a live plot of the simulation, by default False

False
plot_trajectory bool

Whether to plot agent trajectories at the end of simulation, by default False

False
seed_infections bool

Whether to seed initial infections for demonstration purposes, by default False

False
Source code in src/amr_hub_abm/run.py
 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
def simulate(  # noqa: PLR0913
    *,
    plot: bool = False,
    record: bool = False,
    live: bool = False,
    plot_trajectory: bool = False,
    use_gpu: bool = False,
    seed_infections: bool = False,
) -> None:
    """
    Simulate the AMR Hub ABM based on a configuration file.

    Parameters
    ----------
    plot : bool, optional
        Whether to save plots of the simulation at regular intervals, by default False
    record : bool, optional
        Whether to record agent states to a CSV file, by default False
    live : bool, optional
        Whether to display a live plot of the simulation, by default False
    plot_trajectory : bool, optional
        Whether to plot agent trajectories at the end of simulation, by default False
    seed_infections : bool, optional
        Whether to seed initial infections for demonstration purposes, by default False

    """
    config = sim_config
    simulation = create_simulation(config, use_gpu=use_gpu)

    if use_gpu:
        logger.info("GPU Acceleration Enabled: Routing physics to NVIDIA Warp")
        # Note: Plotting is now fully supported in GPU mode via unified spatial_engine
    else:
        logger.info("CPU Mode Enabled: Using legacy Python movement logic")

    # Manual setting infections should be replaced @yidilozdemir@arindamsaha1507
    if seed_infections:
        simulation.agents[0].infection_status = InfectionStatus.INFECTED
        simulation.agents[1].infection_status = InfectionStatus.EXPOSED

    if plot:
        output_dir = Path("../simulation_outputs")
        output_dir.mkdir(parents=True, exist_ok=True)

    # Debug info Loop over each agent
    for agent in simulation.agents:
        msg = f"Agent {agent.agent_type, agent.idx} task list"
        logger.info(msg)
        msg = f"{[task.task_type.value for task in agent.tasks]}"
        logger.info(msg)

    logger.info("Simulation starting...")
    plot_path = Path("../simulation_outputs") if plot else None

    # --------------------------------------------------------------------------
    # This is the core sim where agent activity is simulated between recorded events
    run_steps(simulation, plot_path, record=record)
    # --------------------------------------------------------------------------

    if plot_trajectory:
        record = True
        plot_path = Path("../simulation_outputs")

    if record:
        logger.info("Recording agent states to CSV...")
        record_path = Path("simulation_outputs/agent_states.csv")

        # NG Fix Force the operating system to create the folder
        record_path.parent.mkdir(parents=True, exist_ok=True)
        simulation.record_agent_states(record_path)

        if plot_trajectory:
            logger.info("Plotting agent trajectories...")
            if plot_path is None:
                msg = "Plot path must be provided to plot agent trajectories."
                raise ValueError(msg)
            simulation.plot_agent_trajectories(record_path)

    # Writes Results to Disk (Updated to use unified spatial_engine)
    if getattr(simulation, "use_gpu", False) and hasattr(
        simulation.spatial_engine, "export_data"
    ):
        logger.info("Exporting GPU Telemetry to disk")
        simulation.spatial_engine.export_data(output_dir="simulation_outputs")

    logger.info("Simulation completed successfully...")

    if live:
        plt.ioff()  # turn interactive mode off
        plt.show()  # final blocking show so window stays up after sim ends