Simulation Factory

Module for creating simulation instances.

create_simulation(config, *, use_gpu=False)

Create a simulation instance from a configuration file.

Parameters:

Name	Type	Description	Default
config	SimulationConfig	The simulation configuration object.	required
use_gpu	bool	Flag to enable GPU acceleration, by default False	False

Returns:

Type	Description
Simulation	An instance of the Simulation class.

Raises:

Type	Description
InvalidDefinitionError	If the config file is missing one or more required agent kinematics keys (see AgentKinematicsConfig).

Source code in src/amr_hub_abm/simulation_factory.py

def create_simulation(
    config: SimulationConfig,
    *,
    use_gpu: bool = False,
) -> Simulation:
    """
    Create a simulation instance from a configuration file.

    Parameters
    ----------
    config : SimulationConfig
        The simulation configuration object.
    use_gpu : bool, optional
        Flag to enable GPU acceleration, by default False

    Returns
    -------
    Simulation
        An instance of the Simulation class.

    Raises
    ------
    InvalidDefinitionError
        If the config file is missing one or more required agent kinematics
        keys (see ``AgentKinematicsConfig``).

    """
    agent_kinematics = config.agent_kinematics
    task_durations = config.task_durations
    rng_generator = np.random.default_rng()

    buildings_path_value = config.config_data["buildings_path"]
    if not isinstance(buildings_path_value, (str, Path)):
        msg = "'buildings_path' must be a path string."
        raise InvalidDefinitionError(msg)
    buildings_path = Path(buildings_path_value)
    msg = f"Buildings path from config: {buildings_path}"
    logger.debug(msg)
    space_reader = SpaceInputReader(buildings_path, rng_generator)
    logger.debug("Buildings loaded successfully.")
    logger.debug(space_reader.buildings)

    start_time = pd.to_datetime(str(config.config_data["start_time"]))
    end_time = pd.to_datetime(str(config.config_data["end_time"]))
    total_seconds = (end_time - start_time).total_seconds()
    time_step_value = config.config_data["length_of_timestep_in_seconds"]
    if (
        isinstance(time_step_value, bool)
        or not isinstance(time_step_value, int)
        or time_step_value <= 0
    ):
        msg = "'length_of_timestep_in_seconds' must be a positive integer."
        raise InvalidDefinitionError(msg)
    time_step_length_seconds = time_step_value
    total_steps = int(total_seconds // time_step_length_seconds)
    logger.info("Total simulation time steps: %d", total_steps)

    timeseries_data = read_location_timeseries(config.location_data)
    validation_report = validate_location_timeseries(
        data=timeseries_data,
        rooms=space_reader.rooms,
        start_time=start_time,
        end_time=end_time,
        task_durations=task_durations,
    )
    for warning in validation_report.warnings:
        logger.warning("Location data validation warning: %s", warning)
    validation_report.raise_for_errors()

    agents = parse_location_timeseries(
        timeseries_data=timeseries_data,
        rooms=space_reader.rooms,
        start_time=start_time,
        total_time_steps=total_steps,
        time_scaling_factor=time_step_length_seconds,
        rng_generator=rng_generator,
        agent_kinematics=agent_kinematics,
        task_durations=task_durations,
    )

    msg = f"Parsed {len(agents)} agents from location time series."
    logger.info(msg)
    logger.info("Simulation creation complete.")

    return Simulation(
        name="AMR Hub ABM Simulation",
        description="A simulation instance created from configuration.",
        mode=SimulationMode.SPATIAL,
        space=space_reader.buildings,
        agents=agents,
        total_simulation_time=total_steps,
        rng_generator=rng_generator,
        use_gpu=use_gpu,
        agent_max_movement_attempts=agent_kinematics.max_movement_attempts,
    )

get_random_location(room, building, floor)

Get a random location within a room.

Parameters:

Name	Type	Description	Default
room	Room	The Room instance to get a random location from.	required
building	str	The name of the building the room is in.	required
floor	int	The floor number the room is on.	required

Returns:

Type	Description
Location	A Location instance with random coordinates within the room.

Source code in src/amr_hub_abm/simulation_factory.py

def get_random_location(room: Room, building: str, floor: int) -> Location:
    """
    Get a random location within a room.

    Parameters
    ----------
    room : Room
        The Room instance to get a random location from.
    building : str
        The name of the building the room is in.
    floor : int
        The floor number the room is on.

    Returns
    -------
    Location
        A Location instance with random coordinates within the room.

    """
    point = room.get_random_point()
    return Location(
        building=building,
        floor=floor,
        x=point[0],
        y=point[1],
    )

parse_location_string(location_str)

Parse a location string into its components.

Parameters:

Name	Type	Description	Default
location_str	str	The location string in the format "BuildingName:x,y".	required

Returns:

Type	Description
tuple[str, int, str]	A tuple containing the building name, floor number, and room name.

Source code in src/amr_hub_abm/simulation_factory.py

def parse_location_string(location_str: str) -> tuple[str, int, str]:
    """
    Parse a location string into its components.

    Parameters
    ----------
    location_str : str
        The location string in the format "BuildingName:x,y".

    Returns
    -------
    tuple[str, int, str]
        A tuple containing the building name, floor number, and room name.

    """
    building_part, floor, room = location_str.split(":")
    return building_part, int(floor), room

parse_location_timeseries(timeseries_data, rooms, start_time, total_time_steps, time_scaling_factor, rng_generator, agent_kinematics, task_durations)

Parse validated location-event data for agents.

Parameters:

Name	Type	Description	Default
timeseries_data	DataFrame	DataFrame containing the location time series.	required
rooms	list[Room]	List of Room instances representing the simulation space.	required
start_time	Timestamp	The start time of the simulation.	required
total_time_steps	int	The total number of time steps in the simulation.	required
time_scaling_factor	int	The duration of each time step in seconds.	required
rng_generator	Generator	Random number generator for reproducibility.	required
agent_kinematics	AgentKinematicsConfig	Agent kinematics parameters (speed, stochasticity, interaction radius) read from the simulation config file.	required

Returns:

Type	Description
list[Agent]	A list of Agent instances with populated location time series.

Source code in src/amr_hub_abm/simulation_factory.py

def parse_location_timeseries(  # noqa: PLR0913, PLR0915, PLR0912
    timeseries_data: pd.DataFrame,
    rooms: list[Room],
    start_time: pd.Timestamp,
    total_time_steps: int,
    time_scaling_factor: int,
    rng_generator: np.random.Generator,
    agent_kinematics: AgentKinematicsConfig,
    task_durations: TaskDurationConfig,
) -> list[Agent]:
    """
    Parse validated location-event data for agents.

    Parameters
    ----------
    timeseries_data : pd.DataFrame
        DataFrame containing the location time series.
    rooms : list[Room]
        List of Room instances representing the simulation space.
    start_time : pd.Timestamp
        The start time of the simulation.
    total_time_steps : int
        The total number of time steps in the simulation.
    time_scaling_factor : int
        The duration of each time step in seconds.
    rng_generator : np.random.Generator
        Random number generator for reproducibility.
    agent_kinematics : AgentKinematicsConfig
        Agent kinematics parameters (speed, stochasticity, interaction radius)
        read from the simulation config file.

    Returns
    -------
    list[Agent]
        A list of Agent instances with populated location time series.

    """
    hcw_dict: dict[int, Agent] = {}
    patient_dict: dict[int, Agent] = {}

    for _, row in timeseries_data.iterrows():
        hcw_id = int(row["hcw_id"])
        timestamp = pd.Timestamp(row["timestamp"])
        location_str = str(row["location"])
        patient_id = _optional_int(row["patient_id"])
        event_type = str(row["event_type"])
        door_id = _optional_int(row["door_id"])

        timestep_index = timestamp_to_timestep(
            timestamp, start_time, time_scaling_factor
        )
        building, floor, room_str = parse_location_string(location_str)
        additional_info: dict[Any, Any] = {}

        room = next(
            (
                r
                for r in rooms
                if r.name == room_str and r.building == building and r.floor == floor
            ),
            None,
        )

        if room is None:
            msg = f"Room not found: {room_str} in building {building} on floor {floor}"
            raise SimulationModeError(msg)

        if event_type == "attend_patient" and patient_id is None:
            msg = f"Patient ID must be provided for 'attend' events. Row: {row}"
            raise SimulationModeError(msg)

        if patient_id is not None:
            update_patient(
                patient_id=patient_id,
                space_tuple=(building, floor, room),
                patient_dict=patient_dict,
                total_time_steps=total_time_steps,
                rng_generator=rng_generator,
                agent_kinematics=agent_kinematics,
            )
            patient = patient_dict[patient_id]

        if event_type == "attend_patient" and patient_id is not None:
            additional_info["patient"] = patient
            location = patient.location

        elif event_type == "door_access":
            if door_id is None:
                msg = f"Door ID must be provided for 'door_access' events. Row: {row}"
                raise SimulationModeError(msg)

            door = next(
                (d for d in room.doors if d.door_id == door_id),
                None,
            )

            if door is None:
                msg = f"Door ID {door_id} not found in room {room.name}. Row: {row}"
                msg += f" Available doors: {[d.door_id for d in room.doors]}"
                raise SimulationModeError(msg)

            midpoint = door.line.interpolate(0.5, normalized=True)
            point = (midpoint.x, midpoint.y)
            additional_info["door"] = door

            destination = next(
                idx for idx in door.connecting_rooms if idx != room.room_id
            )
            additional_info["destination"] = destination

            location = Location(
                building=building,
                floor=floor,
                x=point[0],
                y=point[1],
            )

        elif event_type == "workstation":
            workstations = [
                c for c in room.contents if c.content_type == ContentType.WORKSTATION
            ]
            if not workstations:
                msg = f"No workstation found in room {room.name} for 'workstation'"
                msg += f" event. Row: {row}. Selecting random location in room instead."
                logger.error(msg)
                possible_locations = [room.get_random_point()]

            own_workstations = [
                w
                for w in workstations
                if w.owner_id == (hcw_id, AgentType.HEALTHCARE_WORKER)
            ]
            if own_workstations:
                possible_locations = [
                    (w.location.x, w.location.y) for w in own_workstations
                ]
            elif workstations:
                available_workstations = [w for w in workstations if not w.owned]
                if not available_workstations:
                    msg = f"No available workstations found in room {room.name}"
                    msg += f" for HCW {hcw_id}."
                    msg += " Selecting random location in room instead."
                    logger.error(msg)
                    possible_locations = [room.get_random_point()]
                else:
                    workstation = available_workstations[0]
                    workstation.owner_id = (hcw_id, AgentType.HEALTHCARE_WORKER)
                    possible_locations = [
                        (workstation.location.x, workstation.location.y)
                    ]
            else:
                msg = f"No workstations found in {room.name} for 'workstation' event."
                msg += f" Row: {row}. Selecting random location in room instead."
                logger.error(msg)
                possible_locations = [room.get_random_point()]

            location = Location(
                building=building,
                floor=floor,
                x=possible_locations[0][0],
                y=possible_locations[0][1],
            )

        elif event_type == "occupy_content":
            content_type = _optional_int(row["content_type"])
            if content_type is None:
                msg = "Content type must be provided for 'occupy_content' events. "
                msg += f"Row: {row}"
                raise SimulationModeError(msg)

            # Dummy location for task assignment; actual location will be determined by
            # the content's location when the task is executed

            location = Location(building=building, floor=floor, x=0, y=0)
            additional_info["content_type"] = ContentType(content_type)
            additional_info["room"] = room

        else:
            msg = f"Unknown event type: {event_type} in row: {row}"
            raise SimulationModeError(msg)

        update_hcw(
            hcw_id=hcw_id,
            space_tuple=(building, floor, room),
            event_tuple=(location, timestep_index, event_type),
            hcw_dict=hcw_dict,
            additional_info=additional_info or None,
            total_time_steps=total_time_steps,
            rng_generator=rng_generator,
            agent_kinematics=agent_kinematics,
            task_durations=task_durations,
        )

    return list(hcw_dict.values()) + list(patient_dict.values())

timestamp_to_timestep(timestamp, start_time, time_scaling_factor)

Convert a timestamp to a simulation time step index.

Parameters:

Name	Type	Description	Default
timestamp	Timestamp	The timestamp to convert.	required
start_time	Timestamp	The simulation start time.	required
time_scaling_factor	int	The duration of timestep in seconds.	required

Returns:

Type	Description
int	The corresponding time step index.

Source code in src/amr_hub_abm/simulation_factory.py

def timestamp_to_timestep(
    timestamp: pd.Timestamp,
    start_time: pd.Timestamp,
    time_scaling_factor: int,
) -> int:
    """
    Convert a timestamp to a simulation time step index.

    Parameters
    ----------
    timestamp : pd.Timestamp
        The timestamp to convert.
    start_time : pd.Timestamp
        The simulation start time.
    time_scaling_factor : int
        The duration of timestep in seconds.

    Returns
    -------
    int
        The corresponding time step index.

    """
    delta = timestamp - start_time
    return int(delta.total_seconds() // time_scaling_factor)

update_hcw(hcw_id, space_tuple, event_tuple, hcw_dict, total_time_steps, rng_generator, agent_kinematics, task_durations, additional_info=None)

Update healthcare worker information from data.

Parameters:

Name	Type	Description	Default
hcw_id	int	The ID of the healthcare worker.	required
space_tuple	tuple[str, int, Room]	A tuple containing the building name, floor number, and Room instance.	required
event_tuple	tuple[Location, int, str]	A tuple containing the event location, time step index, and event type.	required
hcw_dict	dict[int, Agent]	A dictionary mapping healthcare worker IDs to Agent instances.	required
total_time_steps	int	The total number of time steps in the simulation.	required
rng_generator	Generator	Random number generator for reproducibility.	required
agent_kinematics	AgentKinematicsConfig	Agent kinematics parameters (speed, stochasticity, interaction radius) read from the simulation config file.	required
additional_info	dict | None	Additional information related to the event, by default None.	None

Source code in src/amr_hub_abm/simulation_factory.py

def update_hcw(  # noqa: PLR0913
    hcw_id: int,
    space_tuple: tuple[str, int, Room],
    event_tuple: tuple[Location, int, str],
    hcw_dict: dict[int, Agent],
    total_time_steps: int,
    rng_generator: np.random.Generator,
    agent_kinematics: AgentKinematicsConfig,
    task_durations: TaskDurationConfig,
    additional_info: dict | None = None,
) -> None:
    """
    Update healthcare worker information from data.

    Parameters
    ----------
    hcw_id : int
        The ID of the healthcare worker.
    space_tuple : tuple[str, int, Room]
        A tuple containing the building name, floor number, and Room instance.
    event_tuple : tuple[Location, int, str]
        A tuple containing the event location, time step index, and event type.
    hcw_dict : dict[int, Agent]
        A dictionary mapping healthcare worker IDs to Agent instances.
    total_time_steps : int
        The total number of time steps in the simulation.
    rng_generator : np.random.Generator
        Random number generator for reproducibility.
    agent_kinematics : AgentKinematicsConfig
        Agent kinematics parameters (speed, stochasticity, interaction radius)
        read from the simulation config file.
    additional_info : dict | None, optional
        Additional information related to the event, by default None.

    """
    building, floor, room = space_tuple
    location, timestep_index, event_type = event_tuple
    time_needed = task_durations.task_duration_mapping.get(event_type)
    if time_needed is None:
        msg = f"Unknown event type: {event_type}. Allowed types: "
        msg += f"{task_durations.task_duration_mapping.keys()}"
        raise SimulationModeError(msg)

    if hcw_id not in hcw_dict:
        available_chairs = [
            c
            for c in room.contents
            if c.content_type == ContentType.CHAIR and c.available
        ]

        if not available_chairs:
            msg = f"No available chairs found in room {room.name} for HCW {hcw_id}."
            msg += " Selecting random location in room instead."
            logger.error(msg)
            hcw_location = get_random_location(room, building, floor)
        else:
            chair = available_chairs[0]
            chair.occupier_id = (hcw_id, AgentType.HEALTHCARE_WORKER)
            hcw_location = chair.location

        # FIX: Removed `rooms=rooms`
        hcw_dict[hcw_id] = Agent(
            idx=hcw_id,
            location=hcw_location,
            heading_rad=0.0,
            agent_type=AgentType.HEALTHCARE_WORKER,
            trajectory_length=total_time_steps,
            rng_generator=rng_generator,
            movement_speed=agent_kinematics.movement_speed,
            stochasticity=agent_kinematics.stochasticity,
            interaction_radius=agent_kinematics.interaction_radius,
        )

    hcw_dict[hcw_id].add_task(
        time=timestep_index,
        time_needed=time_needed,
        location=location,
        event_type=event_type,
        additional_info=additional_info,
    )

update_patient(patient_id, space_tuple, patient_dict, total_time_steps, rng_generator, agent_kinematics)

Update patient information from data.

Parameters:

Name	Type	Description	Default
patient_id	int	The ID of the patient.	required
space_tuple	tuple[str, int, Room]	A tuple containing the building name, floor number, and Room instance.	required
patient_dict	dict[int, Agent]	A dictionary mapping patient IDs to Agent instances.	required
total_time_steps	int	The total number of time steps in the simulation.	required
rng_generator	Generator	Random number generator for reproducibility.	required
agent_kinematics	AgentKinematicsConfig	Agent kinematics parameters (speed, stochasticity, interaction radius) read from the simulation config file.	required

Source code in src/amr_hub_abm/simulation_factory.py

def update_patient(  # noqa: PLR0913
    patient_id: int,
    space_tuple: tuple[str, int, Room],
    patient_dict: dict[int, Agent],
    total_time_steps: int,
    rng_generator: np.random.Generator,
    agent_kinematics: AgentKinematicsConfig,
) -> None:
    """
    Update patient information from data.

    Parameters
    ----------
    patient_id : int
        The ID of the patient.
    space_tuple : tuple[str, int, Room]
        A tuple containing the building name, floor number, and Room instance.
    patient_dict : dict[int, Agent]
        A dictionary mapping patient IDs to Agent instances.
    total_time_steps : int
        The total number of time steps in the simulation.
    rng_generator : np.random.Generator
        Random number generator for reproducibility.
    agent_kinematics : AgentKinematicsConfig
        Agent kinematics parameters (speed, stochasticity, interaction radius)
        read from the simulation config file.

    """
    building, floor, room = space_tuple

    if patient_id is not None and patient_id not in patient_dict:
        available_beds = [
            b
            for b in room.contents
            if b.content_type == ContentType.BED and b.available
        ]
        if not available_beds:
            msg = (
                f"No available beds found in room {room.name} for patient {patient_id}."
            )
            msg += " Selecting random location in room instead."
            logger.error(msg)
            location = get_random_location(room, building, floor)
        else:
            bed = available_beds[0]
            bed.occupier_id = (patient_id, AgentType.PATIENT)
            location = bed.location

        # FIX: Removed `rooms=rooms`
        patient_dict[patient_id] = Agent(
            idx=patient_id,
            location=location,
            heading_rad=0.0,
            agent_type=AgentType.PATIENT,
            trajectory_length=total_time_steps,
            rng_generator=rng_generator,
            movement_speed=agent_kinematics.movement_speed,
            stochasticity=agent_kinematics.stochasticity,
            interaction_radius=agent_kinematics.interaction_radius,
        )