Toy Problem Overview
The first part of the project focusses on a toy problem to illustrate the core concepts of the AMR-Hub framework. We make minimal but informed assumptions about the for the forms of input data and try to construct a simple and flexible data-structure to represent the problem.
The Overall Problem Setup
In general, a standard modelling workflow can be represented as follows:
flowchart LR
A[📥 Inputs] --> B[🧠Model]
B --> C[📊 Predictions]
C --> D[✅ Verification and Prediction Assessment]
D --> A
%% Optional annotation
D -->|Feedback & Updates| BIn the AMR-Hub framework, we expect each of these compoments to be broken down as follows:
Inputs (📥)
- Architectural Data: Building layouts
- Movement Data: Sparse and Continuous time data
- Disease Data: Infection characteristics, Resistance profiles
- Model Assumptions and Parameters: Transmission rates, Movement patterns, and other relevant parameters
Model (🧠)
- Representation of the environment (e.g., building layout, rooms, corridors)
- Representation of agents (e.g., individuals, their states, and behaviors)
- Representation of tasks (e.g., cleaning, maintenance, patient care)
- Simulation engine to model movement, interactions, and disease spread
- Possible changes to the environment or agent behaviors over time
Predictions (📊)
- Agent movement patterns
- Task completion rates
- Infection spread over time
- Environmental contamination levels
Verification and Prediction Assessment (✅)
- Evaluation of task completion against actual records
- Comparison of predicted movement patterns with observed data
- Assessment of infection spread predictions against real outbreak data
- Analysis of environmental contamination predictions with sensor data
The Toy Problem Setup (Upto now) 🧩
The toy problem aims to simulate a part of the above workflow with simplified assumptions and data structures. The focus is on creating a minimal yet functional representation of the problem to demonstrate the capabilities of the AMR-Hub framework.
There are certains parts of the problem completely left out for now, such as the disease spread, verification and prediction assessment part, and environmental contamination predictions. The focus is on building a flexible and extensible framework using core Python data-structures and classes, which can be easily expanded to include more complex features in the future.
Focus is also on creating a modular design where different components (e.g., environment representation, agent behaviors, task management) can be developed and tested independently before integrating them into a complete simulation model.
Info
The Toy Problem: Given a simple building layout with a Ward, a Corridor, and a Staff Room, simulate the movement of a health care worker who attends to patients in the Ward and takes breaks in the Staff Room, while completing assigned tasks over a specified time period.
The Input Data Format 📥
The toy-problem uses three input files in YAML and CSV formats to define the parameters of the model, the environment, and the movement data of the agents. These files are located in the tests/inputs/ directory.
The model parameters are defined in the simulation_config.yml file:
mode: data driven # Simulation mode: "data driven" or "rule based"
location_data:
format: duckdb
path: tests/inputs/location_timeseries.duckdb
table: location_timeseries
schema_version: 1
buildings_path: tests/inputs/buildings.yml # Path to buildings geojson file
agent_movement_speed: 0.001 # Agent movement speed in units per time step
agent_stochasticity: 5.0 # Degrees of randomness applied to agent heading each step
agent_interaction_radius: 0.01 # Distance within which an agent is considered to have reached a target
agent_max_movement_attempts: 5 # Max attempts to find a movement step that avoids wall intersections
time_needed_attend_patient: 15 # Timesteps needed for a healthcare worker to attend to a patient
time_needed_door_access: 1 # Timesteps needed for a healthcare worker to access a door (e.g., open/close)
time_needed_workstation: 30 # Timesteps needed for a healthcare worker to use a workstation
time_needed_occupy_content: 10 # Timesteps needed for a healthcare worker to occupy content (e.g. a bed, a chair, etc.)
start_time: 2024-01-01 00:00:00
end_time: 2024-01-02 00:00:00
length_of_timestep_in_seconds: 1 # Length of each timestep in seconds (e.g., 900 seconds = 15 minutes)
So, in addition to the simulation related parameters, we also point to the other two input files here.
The environment layout is defined in the buildings.yml file:
building:
name: Sample Hospital
address: 123 Health St, Wellness City
floors:
- level: 0
rooms:
- name: Staff Room
walls:
- [0, 0, 0, 3]
- [0, 3, 3, 3]
- [4, 3, 5, 3]
- [5, 3, 5, 0]
- [5, 0, 0, 0]
doors:
- [3, 3, 4, 3]
contents:
- type: "workstation"
position: [0.5, 0.5]
- type: "workstation"
position: [0.5, 1.5]
- type: "workstation"
position: [0.5, 2.5]
- type: "chair"
position: [4.5, 0.5]
- type: "chair"
position: [4.5, 1.5]
- type: "chair"
position: [4.5, 2.5]
- name: Ward
walls:
- [0, 4, 0, 7]
- [0, 7, 5, 7]
- [5, 7, 5, 4]
- [5, 4, 3, 4]
- [2, 4, 0, 4]
doors:
- [2, 4, 3, 4]
contents:
- type: "bed"
position: [1, 5]
- type: "bed"
position: [1, 6]
- type: "bed"
position: [2, 5]
- type: "bed"
position: [2, 6]
- name: Corridor
walls:
- [0, 3, 0, 4]
- [0, 4, 2, 4]
- [3, 4, 5, 4]
- [5, 4, 5, 3]
- [5, 3, 4, 3]
- [3, 3, 0, 3]
doors:
- [2, 4, 3, 4]
- [4, 3, 3, 3]
contents:
- type: "chair"
position: [2, 3.5]
- type: "chair"
position: [4, 3.5]
- type: "chair"
position: [1, 3.5]
- type: "chair"
position: [3, 3.5]
This file describes the building layout, including rooms, their types, and connections. This structure of building --> floors --> rooms --> contents is also checked in code for consistency. In future, we would like to convert the CAD files of building layouts into this YAML format automatically.
The movement data of the agents is defined in the location_timeseries.csv file:
hcw_id,timestamp,location,event_type,patient_id,door_id,content_type
1,2024-01-01 01:00:00,Sample Hospital:0:Staff Room,workstation,-,-,-
1,2024-01-01 05:00:00,Sample Hospital:0:Staff Room,door_access,-,1,-
1,2024-01-01 06:00:00,Sample Hospital:0:Corridor,door_access,-,0,-
1,2024-01-01 07:00:00,Sample Hospital:0:Ward,attend_patient,1,-,-
1,2024-01-01 08:30:00,Sample Hospital:0:Ward,door_access,-,0,-
1,2024-01-01 09:30:00,Sample Hospital:0:Corridor,door_access,-,1,-
1,2024-01-01 10:00:00,Sample Hospital:0:Staff Room,door_access,-,1,-
1,2024-01-01 11:30:00,Sample Hospital:0:Corridor,door_access,-,0,-
1,2024-01-01 12:00:00,Sample Hospital:0:Ward,attend_patient,2,-,-
1,2024-01-01 13:00:00,Sample Hospital:0:Ward,attend_patient,2,-,-
1,2024-01-01 14:00:00,Sample Hospital:0:Ward,attend_patient,1,-,-
1,2024-01-01 14:30:00,Sample Hospital:0:Ward,door_access,-,0,-
1,2024-01-01 15:30:00,Sample Hospital:0:Corridor,door_access,-,1,-
1,2024-01-01 16:00:00,Sample Hospital:0:Staff Room,workstation,-,-,-
1,2024-01-01 16:30:00,Sample Hospital:0:Staff Room,door_access,-,1,-
1,2024-01-01 17:00:00,Sample Hospital:0:Corridor,door_access,-,0,-
1,2024-01-01 18:00:00,Sample Hospital:0:Ward,attend_patient,1,-,-
1,2024-01-01 19:00:00,Sample Hospital:0:Ward,attend_patient,3,-,-
1,2024-01-01 19:30:00,Sample Hospital:0:Ward,door_access,-,0,-
1,2024-01-01 20:00:00,Sample Hospital:0:Corridor,door_access,-,1,-
1,2024-01-01 21:00:00,Sample Hospital:0:Staff Room,occupy_content,-,-,2
2,2024-01-01 01:00:00,Sample Hospital:0:Staff Room,workstation,-,-,-
2,2024-01-01 05:00:00,Sample Hospital:0:Staff Room,door_access,-,1,-
2,2024-01-01 06:00:00,Sample Hospital:0:Corridor,door_access,-,0,-
2,2024-01-01 07:00:00,Sample Hospital:0:Ward,attend_patient,1,-,-
2,2024-01-01 08:00:00,Sample Hospital:0:Ward,door_access,-,0,-
2,2024-01-01 09:00:00,Sample Hospital:0:Corridor,door_access,-,1,-
2,2024-01-01 10:00:00,Sample Hospital:0:Staff Room,door_access,-,1,-
2,2024-01-01 11:00:00,Sample Hospital:0:Corridor,door_access,-,0,-
2,2024-01-01 12:00:00,Sample Hospital:0:Ward,attend_patient,2,-,-
2,2024-01-01 13:00:00,Sample Hospital:0:Ward,attend_patient,2,-,-
2,2024-01-01 14:00:00,Sample Hospital:0:Ward,attend_patient,1,-,-
2,2024-01-01 14:00:00,Sample Hospital:0:Ward,door_access,-,0,-
2,2024-01-01 15:00:00,Sample Hospital:0:Corridor,door_access,-,1,-
2,2024-01-01 16:00:00,Sample Hospital:0:Staff Room,workstation,-,-,-
2,2024-01-01 16:00:00,Sample Hospital:0:Staff Room,door_access,-,1,-
2,2024-01-01 17:00:00,Sample Hospital:0:Corridor,door_access,-,0,-
2,2024-01-01 18:00:00,Sample Hospital:0:Ward,attend_patient,1,-,-
2,2024-01-01 19:00:00,Sample Hospital:0:Ward,attend_patient,3,-,-
2,2024-01-01 19:00:00,Sample Hospital:0:Ward,door_access,-,0,-
2,2024-01-01 20:00:00,Sample Hospital:0:Corridor,door_access,-,1,-
2,2024-01-01 21:00:00,Sample Hospital:0:Staff Room,occupy_content,-,-,2
This CSV file contains timestamped location data for the agents, which will be used to simulate their movement within the building. In the code, this is used also to create agents and initiate their task-lists (explained later).
The Model Structure ðŸ§
The core forward model is structured around a few key classes and functions that represent the main components of the simulation. The overall flow of data and interactions between these components can be visualized as follows:
@dataclass
class Simulation:
"""
Representation of the AMR Hub ABM simulation.
This class encapsulates the entire state and behavior of the simulation, including
the space (buildings, floors, rooms), the agents, and the logic for advancing
the simulation through time steps. It also includes methods for plotting the current
state of the simulation and recording agent states to files.
Parameters
----------
name : str
The name of the simulation.
description : str
A brief description of the simulation.
mode : SimulationMode
The mode of the simulation (SPATIAL or TOPOLOGICAL).
space : list[Building]
The simulation space represented as a list of Building instances.
agents : list[Agent]
The list of Agent instances in the simulation.
total_simulation_time : int
The total number of time steps in the simulation.
rng_generator : np.random.Generator
Random number generator for reproducibility.
agent_max_movement_attempts : int, optional
Maximum number of attempts an agent may make to find a movement step
that avoids wall intersections, by default 5.
"""
# ------------------------------------------------------------------------------
name: str
description: str
mode: SimulationMode
space: list[Building]
agents: list[Agent]
total_simulation_time: int
rng_generator: np.random.Generator
time: int = field(default=0, init=False)
# NG Added Flag for GPU Acceleration
use_gpu: bool = field(default=False)
# Unified engine property (handles both CPU and GPU cleanly)
spatial_engine: Any = field(default=None, init=False)
agent_max_movement_attempts: int = field(default=5)
# ------------------------------------------------------------------------------
def __post_init__(self) -> None:
"""Initialize the appropriate spatial engine (CPU or GPU)."""
if self.use_gpu:
# Pass the required interface arguments to the GPU engine
self.spatial_engine = GPUSpatialQuery(
space=self.space,
max_movement_attempts=self.agent_max_movement_attempts,
)
else:
self.spatial_engine = SpatialQuery(
space=self.space,
max_movement_attempts=self.agent_max_movement_attempts,
)
self._agent_store = None
# ------------------------------------------------------------------------------
# ------------------------------------------------------------------------------
def step(self, plot_path: Path | None = None, *, record: bool = False) -> None:
"""Advance the simulation by one time step."""
if self.time >= self.total_simulation_time:
msg = "Simulation has already reached its total simulation time."
raise TimeError(msg)
# randomize agent order each step to avoid bias
self.rng_generator.shuffle(self.agents)
# --------------------------------------------------------------------------
# 1. CPU Evaluates Logic and set targets but if GPU then no movement
# --------------------------------------------------------------------------
# Always run the task state machine so agents transition correctly.
# If engine is type gpu then polymorphic move_one_step does nothing
for agent in self.agents:
agent.perform_task(
current_time=self.time, engine=self.spatial_engine, record=record
)
# --------------------------------------------------------------------------
# 2. GPU Handover to do "move_one_step"
# --------------------------------------------------------------------------
if self.use_gpu:
# GPU computes collision, stochastics, and proximity in one parallel batch
self.spatial_engine.step_physics(self.agents)
# --------------------------------------------------------------------------
if plot_path is not None:
self.plot_current_state(directory_path=plot_path)
self.time += 1
# ------------------------------------------------------------------------------
def plot_current_state(
self, directory_path: Path, *, trajectory: bool = False
) -> None:
"""
Plot the current state of the simulation.
This method iterates through each building in the simulation space and creates
a plot of the building's layout along with the positions of agents. If the
`trajectory` flag is set to True, it also plots the trajectories of agents up to
the current time step. The plots are saved to the specified directory.
Parameters
----------
directory_path : Path
The directory where the plots will be saved. The method will create the
directory if it does not exist.
trajectory : bool, optional
Whether to plot agent trajectories up to the current time step,
by default False.
"""
if directory_path.suffix != "":
msg = f"The path {directory_path} is not a directory."
raise NotADirectoryError(msg)
directory_path.mkdir(parents=True, exist_ok=True)
for building in self.space:
axes: list[Axes] = [plt.subplots(nrows=len(building.floors), ncols=1)[1]] # type: ignore # noqa: PGH003
building.plot_building(axes=axes, agents=self.agents, trajectory=trajectory)
simulation_name = f"Simulation: {self.name}"
if trajectory:
simulation_name += " | Agent Trajectories"
filename = f"{building.name}_trajectories.png"
else:
simulation_name += f" | Time: {self.time}/{self.total_simulation_time}"
filename = f"{building.name}_time_{self.time}.png"
plt.suptitle(simulation_name)
plt.savefig(directory_path / filename)
plt.close()
def plot_live(
self,
figures: list,
*,
pause: float = 0.05,
trajectory: bool = False,
) -> None:
"""
Update existing figures in place with current agent positions.
This method is designed for live plotting during the simulation run. It takes a
list of figures and updates them with the current agent positions.
If the `trajectory` flag is set to True, it also updates the plots with agent
trajectories up to the current time step. The method uses `plt.pause` to create
a brief pause after updating the plots to allow for visualization.
Parameters
----------
figures : list
A list of tuples containing (building, figure, axes) for each building in
the simulation.
pause : float, optional
The amount of time to pause after updating the plots, by default 0.05
seconds.
trajectory : bool, optional
Whether to include agent trajectories in the live plot, by default False.
"""
for building, fig, axes in figures:
for ax in axes:
ax.clear()
building.plot_building(
axes=axes,
agents=self.agents,
trajectory=trajectory,
)
title = (
f"Simulation: {self.name} | "
f"Time: {self.time}/{self.total_simulation_time}"
)
if trajectory:
title += " | Trajectories"
fig.suptitle(title)
fig.canvas.draw_idle()
fig.canvas.flush_events()
plt.pause(pause)
def setup_live_plot(self) -> list:
"""
Create reusable figures for live plotting. Call once before the run loop.
This method initializes Matplotlib figures for each building in the simulation
space. It creates a subplot for each floor of the building and returns a list of
tuples containing the building, figure, and axes. These figures can then be
updated in place during the simulation run using the `plot_live` method.
Returns
-------
list
A list of tuples, where each tuple contains (building, figure, axes) for
each building in the simulation.
"""
plt.ion()
figures = []
for building in self.space:
fig, axes = plt.subplots(
nrows=len(building.floors),
ncols=1,
figsize=(10, 6 * len(building.floors)),
)
axes = [axes] if len(building.floors) == 1 else list(axes)
figures.append((building, fig, axes))
plt.show(block=False)
return figures
def __repr__(self) -> str:
"""Representation of the simulation."""
header = f"Simulation: {self.name}\nDescription: {self.description}\n"
header += f"Mode: {self.mode.value}\n"
header += f"Total Simulation Time: {self.total_simulation_time}\n"
header += f"Current Time: {self.time}\n"
header += f"Number of Buildings: {len(self.space)}\n"
header += f"Number of Agents: {len(self.agents)}\n"
buildings_repr = "\n".join([repr(building) for building in self.space])
agents_repr = "\n".join([repr(agent) for agent in self.agents])
return f"{header}\nBuildings:\n{buildings_repr}\n\nAgents:\n{agents_repr}"
@property
def rooms(self) -> list[Room]:
"""Get all rooms in the simulation space."""
all_rooms: list[Room] = []
for building in self.space:
for floor in building.floors:
all_rooms.extend(floor.rooms)
return all_rooms
def record_agent_states(self, file_path: Path) -> None:
"""
Record the states of all agents at the current time step to a CSV file.
This method saves the current state of each agent, including their position,
heading, infection status, and other relevant attributes, to a CSV file at the
specified path. The CSV file will have a header row and will be overwritten if
it already exists.
Parameters
----------
file_path : Path
The path to the CSV file where agent states will be recorded.
"""
for agent in self.agents:
agent_filename = (
file_path.parent
/ f"agent_{agent.agent_type.value}_{agent.idx}_trajectory.csv"
)
np.savetxt(
agent_filename,
np.column_stack(
[
np.arange(len(agent.trajectory.position)),
agent.trajectory.building,
agent.trajectory.floor,
agent.trajectory.position,
agent.trajectory.heading.T.flatten(),
agent.trajectory.infection_status,
]
),
delimiter=",",
header="time,building,floor,x,y,heading,infection_status",
comments="",
)
def plot_agent_trajectories(self, output_file: Path) -> None:
"""
Plot the trajectories of all agents from a recorded CSV file.
Parameters
----------
output_file : Path
The path to the CSV file containing the agent trajectories.
"""
self.plot_current_state(directory_path=output_file.parent, trajectory=True)
Building
The space in which the agents operate and consists of a list of Building objects, which in turn contain Floor and Room objects. The Room is a key component, as it holds information about the room type, its connections to other rooms via Door objects, and any Component objects (yet to be implemented) present in the room (e.g., beds, desks).
Door
The Door class represents the connections between rooms within the environment. Each door has attributes such as door ID, connected rooms, and door type (e.g., standard, fire exit). The door management system allows for defining how agents can move between different rooms based on the building layout.
The Door class inherits from the DetachedDoor class, which encapsulates the basic properties of a door without any room connections. This separation allows for more flexible door management and easier handling of door attributes.
Agent
The Agent class represents the individuals moving within the environment. Each agent has attributes such as ID, type (e.g., healthcare worker), current location, and a list of assigned Task objects. The agent's behavior is defined by methods defined in the Task list that allow it to move between rooms, perform the tasks, and update its state based on the simulation parameters.
Task
The Task class encapsulates the actions that agents need to perform within the environment. Each task has attributes such as task type (e.g., attend patient, take break), associated room, start and end times, and status (e.g., pending, in-progress, completed). The task management system allows for scheduling, execution, and tracking of tasks assigned to agents.
Simulation Execution
Once the space and agents are set up, the simulation can be executed using the Simulation class. The simulation runs for a specified duration, updating the state of agents and tasks at each time step based on the defined parameters and interactions.
For each time step,
- Each health care worker agent checks their assigned tasks.
- The agent estimates the time required to reach the task location based on their movement speed and current position.
- If the agent can reach the task location within the current time step, they move to the location and perform the task.
- The task status is updated to "in-progress" or "completed" based on the agent's actions.
Results and Outputs 📊
Currently, the output and visualisation components are under development. As of now, we can plot the floor layout of the building and the positions of the agents at different time steps using Matplotlib.