API reference

amr_hub_abm package.

agent

Module to represent an agent in the AMR Hub ABM simulation.

Agent dataclass

Representation of an agent in the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/agent.py

@dataclass
class Agent:
    """Representation of an agent in the AMR Hub ABM simulation."""

    idx: int
    location: Location
    heading_rad: float
    space: list[Building]
    rng_generator: Generator

    interaction_radius: float = field(default=0.01)
    tasks: list[Task] = field(default_factory=list)
    agent_type: AgentType = field(default=AgentType.GENERIC)
    infection_status: InfectionStatus = field(default=InfectionStatus.SUSCEPTIBLE)
    infection_details: dict = field(default_factory=dict)

    movement_speed: float = field(default=0.1)  # units per time step

    trajectory_length: int = field(default=0)
    trajectory: Record = field(init=False)

    stationary: bool = field(default=False, init=False)

    @property
    def heading_degrees(self) -> float:
        """Get the agent's heading in degrees."""
        return math.degrees(self.heading_rad)

    @heading_degrees.setter
    def heading_degrees(self, value: float) -> None:
        self.heading_rad = math.radians(value) % (2 * math.pi)

    def __post_init__(self) -> None:
        """Post-initialization to log agent creation."""
        # Ensure heading is between 0 and 360 degrees
        self.heading_rad = self.heading_rad % (2 * math.pi)

        logger.debug(
            "Created Agent id %s of type %s at location %s with heading %s",
            self.idx,
            self.agent_type,
            self.location,
            self.heading_rad,
        )

        if self.trajectory_length < 0:
            msg = "trajectory_length must be non-negative."
            raise ValueError(msg)

        if self.trajectory_length > 0:
            self.trajectory = Record(total_time=self.trajectory_length)

    def get_room(self, coords: tuple[float, float] | None = None) -> Room | None:
        """Get the room the agent is currently located in, if any."""
        if coords is None:
            coords = (self.location.x, self.location.y)

        for building in self.space:
            if building.name != self.location.building:
                continue
            for floor in building.floors:
                if floor.floor_number != self.location.floor:
                    continue
                room = floor.find_room_by_location(coords)
                if room:
                    return room
        logger.info(
            "Agent id %s is not located in any room. Location: %s",
            self.idx,
            self.location,
        )
        return None

    def check_if_location_reached(self, target_location: Location) -> bool:
        """Check if the agent has reached the target location."""
        if self.location.building != target_location.building:
            return False
        if self.location.floor != target_location.floor:
            return False

        distance = self.location.distance_to(target_location)
        return distance <= self.interaction_radius

    def move_to_location(self, new_location: Location) -> None:
        """Move the agent to a new location and log the movement."""
        msg = f"Moving Agent id {self.idx} from {self.location} to {new_location}"
        logger.info(msg)
        self.location = new_location

    def plot_agent(self, ax: Axes, *, show_tags: bool = True) -> None:
        """Plot the agent on the given axes."""
        ax.plot(
            self.location.x,
            self.location.y,
            marker="o",
            markersize=5,
            color=ROLE_COLOUR_MAP[self.agent_type],
        )

        if show_tags:
            ax.text(
                self.location.x + 0.05,
                self.location.y + 0.05,
                f"{self.agent_type.value} {self.idx}",
                fontsize=8,
                ha="left",
                va="bottom",
            )

    def plot_trajectory(self, ax: Axes) -> None:
        """Plot the agent's trajectory on the given axes."""
        if self.trajectory_length == 0:
            msg = "Cannot plot trajectory for agent with trajectory_length of 0."
            raise ValueError(msg)

        ax.plot(
            self.trajectory.position[:, 0],
            self.trajectory.position[:, 1],
            linestyle="-",
            linewidth=1.5,
            color=ROLE_COLOUR_MAP[self.agent_type],
            alpha=0.7,
        )

    def __repr__(self) -> str:
        """Return a string representation of the agent."""
        return (
            f"Agent(idx={self.idx}, {self.location}, "
            f"{math.degrees(self.heading_rad):.2f}°, "
            f"{self.interaction_radius}, {self.agent_type.value}, "
            f"{self.infection_status.value})"
        )

    def add_task(  # noqa: PLR0912
        self,
        time: int,
        location: Location,
        event_type: str,
        additional_info: dict | None = None,
    ) -> None:
        """Add a task to the agent's task list and log the addition."""
        if event_type not in TASK_TYPES:
            msg = f"Invalid task type: {event_type}. Must be one of {TASK_TYPES}."
            raise SimulationModeError(msg)
        task_type = TaskType[event_type.upper()]
        task: Task

        if task_type == TaskType.ATTEND_PATIENT:
            if additional_info is None or "patient" not in additional_info:
                msg = "Patient ID must be provided for attend_patient tasks."
                raise SimulationModeError(msg)

            patient = additional_info["patient"]
            if not isinstance(patient, Agent):
                msg = "Patient must be an instance of Agent."
                raise SimulationModeError(msg)

            task = TaskAttendPatient(
                time_needed=15,
                time_due=time,
                patient=patient,
            )

        elif task_type == TaskType.DOOR_ACCESS:
            if location.building is None or location.floor is None:
                msg = "Building and floor must be provided for door access tasks."
                raise SimulationModeError(msg)

            if (
                additional_info is None
                or "door" not in additional_info
                or not isinstance(additional_info["door"], Door)
            ):
                msg = "Door must be provided in additional_info for door access tasks."
                raise SimulationModeError(msg)

            task = TaskDoorAccess(
                door=additional_info["door"],
                destination_room=additional_info["destination"],
                time_needed=1,
                time_due=time,
                building=location.building,
                floor=location.floor,
            )

        elif task_type == TaskType.WORKSTATION:
            task = TaskWorkstation(
                workstation_location=location,
                time_needed=30,
                time_due=time,
            )

        elif task_type == TaskType.OCCUPY_CONTENT:
            if not additional_info or not isinstance(additional_info, dict):
                msg = "additional_info must be a dictionary for occupy_content tasks."
                raise SimulationModeError(msg)

            if "content_type" not in additional_info:
                msg = "Content type must be provided in additional_info for "
                msg += "occupy_content tasks."
                raise SimulationModeError(msg)

            if "room" not in additional_info or not isinstance(
                additional_info["room"], Room
            ):
                msg = (
                    "Room must be provided in additional_info for occupy_content tasks."
                )
                raise SimulationModeError(msg)

            task = TaskOccupyContent(
                content_type=additional_info["content_type"],
                room=additional_info["room"],
                time_needed=10,
                time_due=time,
            )

        else:
            msg = f"Task type {task_type.name} not implemented yet."
            raise NotImplementedError(msg)

        self.tasks.append(task)

    def head_to_point(self, point: tuple[float, float]) -> None:
        """Set the agent's heading to face a specific point."""
        delta_x = point[0] - self.location.x
        delta_y = point[1] - self.location.y

        self.heading_rad = math.atan2(delta_y, delta_x) % (2 * math.pi)

    @staticmethod
    def propose_new_coordinates(
        coordinates: tuple[float, float],
        heading_rad: float,
        movement_speed: float,
        stochasticity: float,
        rng_generator: Generator,
    ) -> tuple[float, float]:
        """Propose a new location for agent movement."""
        delta_x = movement_speed * math.cos(heading_rad)
        delta_y = movement_speed * math.sin(heading_rad)

        delta_x = (1 + rng_generator.normal(0, stochasticity)) * delta_x
        delta_y = (1 + rng_generator.normal(0, stochasticity)) * delta_y

        new_x = coordinates[0] + delta_x
        new_y = coordinates[1] + delta_y

        return new_x, new_y

    def try_move_one_step(
        self,
        stochasticity: float,
        max_attempts: int = 5,
    ) -> tuple[float, float]:
        """Return valid coordinates for a single movement step."""
        for attempt in range(1, max_attempts + 1):
            new_x, new_y = self.propose_new_coordinates(
                (self.location.x, self.location.y),
                self.heading_rad,
                self.movement_speed,
                stochasticity,
                self.rng_generator,
            )

            room = self.get_room((new_x, new_y))
            if room is None:
                logger.info(
                    "Attempt %s: location (%s, %s) is not located in any room.",
                    attempt,
                    new_x,
                    new_y,
                )
                continue

            walls = room.walls
            if not walls:
                msg = (
                    f"Room {room.name} has no walls defined, "
                    "cannot check for wall intersections."
                )
                raise SimulationModeError(msg)

            if Location.check_intersection_with_walls(
                new_x,
                new_y,
                self.interaction_radius,
                walls,
            ):
                logger.info(
                    "Attempt %s: Agent id %s cannot move to (%s, %s): "
                    "wall intersection.",
                    attempt,
                    self.idx,
                    new_x,
                    new_y,
                )
                continue

            return new_x, new_y

        logger.info(
            "Maximum attempts %s exceeded for moving one step. "
            "Agent id %s moving to proposed coordinates (%s, %s) despite "
            "wall intersection.",
            max_attempts,
            self.idx,
            self.location.x,
            self.location.y,
        )

        return new_x, new_y

    def move_one_step(self, stochasticity: float = 0.2) -> None:
        """Move the agent one step in the direction of its heading."""
        new_x, new_y = self.try_move_one_step(stochasticity)
        self.move_to_location(replace(self.location, x=new_x, y=new_y))

    def select_task_based_on_progress(
        self, progress: TaskProgress, *, allow_multiple: bool = False
    ) -> Task | None:
        """Select a task based on its progress."""
        tasks = [task for task in self.tasks if task.progress == progress]
        if not tasks:
            return None
        if len(tasks) > 1 and not allow_multiple:
            msg = f"Agent {self.idx} has multiple tasks"
            msg += f" with progress {progress.value}."
            logger.error(msg)
            raise RuntimeError(msg)
        return min(tasks, key=lambda t: (t.time_due, t.priority.value))

    def perform_in_progress_task(self, current_time: int) -> bool:
        """Perform an in-progress task and return True if a task was performed."""
        task = self.select_task_based_on_progress(TaskProgress.IN_PROGRESS)
        if task is None:
            return False
        task.update_progress(current_time=current_time, agent=self)
        return True

    def perform_moving_to_task_location(self, current_time: int) -> bool:
        """Move the agent towards the location of its next task."""
        next_task = self.select_task_based_on_progress(TaskProgress.MOVING_TO_LOCATION)
        if next_task is None:
            return False
        next_task.update_progress(current_time=current_time, agent=self)
        return True

    def perform_suspended_task(self, current_time: int) -> bool:
        """Perform a suspended task and return True if a task was performed."""
        task = self.select_task_based_on_progress(
            TaskProgress.SUSPENDED, allow_multiple=True
        )
        if task is None:
            return False
        task.update_progress(current_time=current_time, agent=self)
        return True

    def perform_to_be_started_task(self, current_time: int) -> bool:
        """Perform a to-be-started task and return True if a task was performed."""
        task = self.select_task_based_on_progress(
            TaskProgress.NOT_STARTED, allow_multiple=True
        )
        if task is None:
            return False
        if isinstance(task, TaskOccupyContent):
            task.assign_content()

        task_move_time = (
            task.time_due
            - task.time_needed
            - self.estimate_time_to_reach_location(task.location)
        )
        logger.info(
            "Agent id %s next task move time: %s, current time: %s",
            self.idx,
            task_move_time,
            current_time,
        )

        if current_time < task_move_time:
            self.attempt_task_insertion(
                next_task=task,
                next_task_move_time=task_move_time,
                current_time=current_time,
            )
            return False

        task.update_progress(current_time=current_time, agent=self)
        return True

    def record_state(self, current_time: int) -> None:
        """Push a record of the agent's current state to the trajectory."""
        if current_time >= self.trajectory_length:
            msg = f"Current time {current_time} "
            msg += f"exceeds trajectory length {self.trajectory_length}."
            raise ValueError(msg)

        building_idx_list = [
            b.idx for b in self.space if b.name == self.location.building
        ]
        if not building_idx_list:
            msg = f"Building {self.location.building} not found in agent's space."
            raise ValueError(msg)
        if len(building_idx_list) > 1:
            msg = f"Multiple buildings with name {self.location.building} found."
            raise ValueError(msg)
        building_idx = building_idx_list[0]

        self.trajectory.push(
            time=current_time,
            building_idx=building_idx,
            floor=self.location.floor,
            pos_x=self.location.x,
            pos_y=self.location.y,
            heading=self.heading_rad,
            infection_status=self.infection_status,
        )

    def perform_task(self, current_time: int, *, record: bool = False) -> None:
        """Perform the agent's current task if it's due."""
        if record:
            logger.info(
                "Recording state for Agent id %s at time %s: location=%s",
                self.idx,
                current_time,
                self.location,
            )
            self.record_state(current_time=current_time)

        if logger.isEnabledFor(logging.INFO):
            task_list_values = [task.task_type.value for task in self.tasks]
            task_progress_values = [task.progress.value for task in self.tasks]
            msg = f"Time {current_time} Task list: {task_list_values}"
            logger.info(msg)
            msg = f"Time {current_time} Task list: {task_progress_values}"
            logger.info(msg)

        if not self.tasks:
            return
        logger.debug(
            "Agent id %s has %s tasks to perform.",
            self.idx,
            len(self.tasks),
        )

        if self.perform_in_progress_task(current_time=current_time):
            return
        logger.debug(
            "No in-progress tasks for Agent id %s.",
            self.idx,
        )

        if self.perform_moving_to_task_location(current_time=current_time):
            return
        logger.debug(
            "No tasks to move to for Agent id %s.",
            self.idx,
        )

        if self.perform_suspended_task(current_time=current_time):
            return
        logger.debug(
            "No suspended tasks for Agent id %s.",
            self.idx,
        )

        if self.perform_to_be_started_task(current_time=current_time):
            return
        logger.debug(
            "No to-be-started tasks for Agent id %s.",
            self.idx,
        )

    def estimate_time_to_reach_location(self, target_location: Location) -> float:
        """Estimate the time required to reach a target location."""
        distance = self.location.distance_to(target_location)
        return distance / self.movement_speed

    def attempt_task_insertion(
        self, next_task: Task, next_task_move_time: float, current_time: int
    ) -> None:
        """Attempt to insert a task to occupy an empty chair."""
        if isinstance(next_task, TaskOccupyContent):
            return
        if self.stationary:
            return

        room = self.get_room()
        if room is None:
            logger.info(
                "Agent id %s is not located in any room. Cannot check for "
                "empty chairs for task %s.",
                self.idx,
                next_task.task_type.name,
            )
            return

        empty_chairs = [
            content
            for content in room.contents
            if content.content_type == ContentType.CHAIR and content.occupier_id is None
        ]

        logger.info(
            "Agent id %s found %s empty chairs in room %s for task %s.",
            self.idx,
            len(empty_chairs),
            room.name,
            next_task.task_type.name,
        )

        if empty_chairs:
            chair = empty_chairs[0]
            estimated_time_to_chair = self.estimate_time_to_reach_location(
                chair.location
            )
            if current_time + estimated_time_to_chair < next_task_move_time:
                self.add_task(
                    time=current_time,
                    location=chair.location,
                    event_type="occupy_content",
                    additional_info={
                        "content_type": ContentType.CHAIR,
                        "room": room,
                    },
                )
                logger.info(
                    """
                    Current time: %s
                    Estimated time to chair: %s
                    Next task move time: %s.
                    Agent id %s inserted occupy_content task for chair at location %s
                    to be performed before next task move time %s.
                    """,
                    current_time,
                    estimated_time_to_chair,
                    next_task_move_time,
                    self.idx,
                    chair.location,
                    next_task_move_time,
                )
                logger.warning(
                    """
                    Length of task list %s at time %s.
                    """,
                    len(self.tasks),
                    current_time,
                )

heading_degrees property writable

Get the agent's heading in degrees.

__post_init__()

Post-initialization to log agent creation.

Source code in src/amr_hub_abm/agent.py

def __post_init__(self) -> None:
    """Post-initialization to log agent creation."""
    # Ensure heading is between 0 and 360 degrees
    self.heading_rad = self.heading_rad % (2 * math.pi)

    logger.debug(
        "Created Agent id %s of type %s at location %s with heading %s",
        self.idx,
        self.agent_type,
        self.location,
        self.heading_rad,
    )

    if self.trajectory_length < 0:
        msg = "trajectory_length must be non-negative."
        raise ValueError(msg)

    if self.trajectory_length > 0:
        self.trajectory = Record(total_time=self.trajectory_length)

__repr__()

Return a string representation of the agent.

Source code in src/amr_hub_abm/agent.py

def __repr__(self) -> str:
    """Return a string representation of the agent."""
    return (
        f"Agent(idx={self.idx}, {self.location}, "
        f"{math.degrees(self.heading_rad):.2f}°, "
        f"{self.interaction_radius}, {self.agent_type.value}, "
        f"{self.infection_status.value})"
    )

add_task(time, location, event_type, additional_info=None)

Add a task to the agent's task list and log the addition.

Source code in src/amr_hub_abm/agent.py

def add_task(  # noqa: PLR0912
    self,
    time: int,
    location: Location,
    event_type: str,
    additional_info: dict | None = None,
) -> None:
    """Add a task to the agent's task list and log the addition."""
    if event_type not in TASK_TYPES:
        msg = f"Invalid task type: {event_type}. Must be one of {TASK_TYPES}."
        raise SimulationModeError(msg)
    task_type = TaskType[event_type.upper()]
    task: Task

    if task_type == TaskType.ATTEND_PATIENT:
        if additional_info is None or "patient" not in additional_info:
            msg = "Patient ID must be provided for attend_patient tasks."
            raise SimulationModeError(msg)

        patient = additional_info["patient"]
        if not isinstance(patient, Agent):
            msg = "Patient must be an instance of Agent."
            raise SimulationModeError(msg)

        task = TaskAttendPatient(
            time_needed=15,
            time_due=time,
            patient=patient,
        )

    elif task_type == TaskType.DOOR_ACCESS:
        if location.building is None or location.floor is None:
            msg = "Building and floor must be provided for door access tasks."
            raise SimulationModeError(msg)

        if (
            additional_info is None
            or "door" not in additional_info
            or not isinstance(additional_info["door"], Door)
        ):
            msg = "Door must be provided in additional_info for door access tasks."
            raise SimulationModeError(msg)

        task = TaskDoorAccess(
            door=additional_info["door"],
            destination_room=additional_info["destination"],
            time_needed=1,
            time_due=time,
            building=location.building,
            floor=location.floor,
        )

    elif task_type == TaskType.WORKSTATION:
        task = TaskWorkstation(
            workstation_location=location,
            time_needed=30,
            time_due=time,
        )

    elif task_type == TaskType.OCCUPY_CONTENT:
        if not additional_info or not isinstance(additional_info, dict):
            msg = "additional_info must be a dictionary for occupy_content tasks."
            raise SimulationModeError(msg)

        if "content_type" not in additional_info:
            msg = "Content type must be provided in additional_info for "
            msg += "occupy_content tasks."
            raise SimulationModeError(msg)

        if "room" not in additional_info or not isinstance(
            additional_info["room"], Room
        ):
            msg = (
                "Room must be provided in additional_info for occupy_content tasks."
            )
            raise SimulationModeError(msg)

        task = TaskOccupyContent(
            content_type=additional_info["content_type"],
            room=additional_info["room"],
            time_needed=10,
            time_due=time,
        )

    else:
        msg = f"Task type {task_type.name} not implemented yet."
        raise NotImplementedError(msg)

    self.tasks.append(task)

attempt_task_insertion(next_task, next_task_move_time, current_time)

Attempt to insert a task to occupy an empty chair.

Source code in src/amr_hub_abm/agent.py

def attempt_task_insertion(
    self, next_task: Task, next_task_move_time: float, current_time: int
) -> None:
    """Attempt to insert a task to occupy an empty chair."""
    if isinstance(next_task, TaskOccupyContent):
        return
    if self.stationary:
        return

    room = self.get_room()
    if room is None:
        logger.info(
            "Agent id %s is not located in any room. Cannot check for "
            "empty chairs for task %s.",
            self.idx,
            next_task.task_type.name,
        )
        return

    empty_chairs = [
        content
        for content in room.contents
        if content.content_type == ContentType.CHAIR and content.occupier_id is None
    ]

    logger.info(
        "Agent id %s found %s empty chairs in room %s for task %s.",
        self.idx,
        len(empty_chairs),
        room.name,
        next_task.task_type.name,
    )

    if empty_chairs:
        chair = empty_chairs[0]
        estimated_time_to_chair = self.estimate_time_to_reach_location(
            chair.location
        )
        if current_time + estimated_time_to_chair < next_task_move_time:
            self.add_task(
                time=current_time,
                location=chair.location,
                event_type="occupy_content",
                additional_info={
                    "content_type": ContentType.CHAIR,
                    "room": room,
                },
            )
            logger.info(
                """
                Current time: %s
                Estimated time to chair: %s
                Next task move time: %s.
                Agent id %s inserted occupy_content task for chair at location %s
                to be performed before next task move time %s.
                """,
                current_time,
                estimated_time_to_chair,
                next_task_move_time,
                self.idx,
                chair.location,
                next_task_move_time,
            )
            logger.warning(
                """
                Length of task list %s at time %s.
                """,
                len(self.tasks),
                current_time,
            )

check_if_location_reached(target_location)

Check if the agent has reached the target location.

Source code in src/amr_hub_abm/agent.py

def check_if_location_reached(self, target_location: Location) -> bool:
    """Check if the agent has reached the target location."""
    if self.location.building != target_location.building:
        return False
    if self.location.floor != target_location.floor:
        return False

    distance = self.location.distance_to(target_location)
    return distance <= self.interaction_radius

estimate_time_to_reach_location(target_location)

Estimate the time required to reach a target location.

Source code in src/amr_hub_abm/agent.py

def estimate_time_to_reach_location(self, target_location: Location) -> float:
    """Estimate the time required to reach a target location."""
    distance = self.location.distance_to(target_location)
    return distance / self.movement_speed

get_room(coords=None)

Get the room the agent is currently located in, if any.

Source code in src/amr_hub_abm/agent.py

def get_room(self, coords: tuple[float, float] | None = None) -> Room | None:
    """Get the room the agent is currently located in, if any."""
    if coords is None:
        coords = (self.location.x, self.location.y)

    for building in self.space:
        if building.name != self.location.building:
            continue
        for floor in building.floors:
            if floor.floor_number != self.location.floor:
                continue
            room = floor.find_room_by_location(coords)
            if room:
                return room
    logger.info(
        "Agent id %s is not located in any room. Location: %s",
        self.idx,
        self.location,
    )
    return None

head_to_point(point)

Set the agent's heading to face a specific point.

Source code in src/amr_hub_abm/agent.py

def head_to_point(self, point: tuple[float, float]) -> None:
    """Set the agent's heading to face a specific point."""
    delta_x = point[0] - self.location.x
    delta_y = point[1] - self.location.y

    self.heading_rad = math.atan2(delta_y, delta_x) % (2 * math.pi)

move_one_step(stochasticity=0.2)

Move the agent one step in the direction of its heading.

Source code in src/amr_hub_abm/agent.py

def move_one_step(self, stochasticity: float = 0.2) -> None:
    """Move the agent one step in the direction of its heading."""
    new_x, new_y = self.try_move_one_step(stochasticity)
    self.move_to_location(replace(self.location, x=new_x, y=new_y))

move_to_location(new_location)

Move the agent to a new location and log the movement.

Source code in src/amr_hub_abm/agent.py

def move_to_location(self, new_location: Location) -> None:
    """Move the agent to a new location and log the movement."""
    msg = f"Moving Agent id {self.idx} from {self.location} to {new_location}"
    logger.info(msg)
    self.location = new_location

perform_in_progress_task(current_time)

Perform an in-progress task and return True if a task was performed.

Source code in src/amr_hub_abm/agent.py

def perform_in_progress_task(self, current_time: int) -> bool:
    """Perform an in-progress task and return True if a task was performed."""
    task = self.select_task_based_on_progress(TaskProgress.IN_PROGRESS)
    if task is None:
        return False
    task.update_progress(current_time=current_time, agent=self)
    return True

perform_moving_to_task_location(current_time)

Move the agent towards the location of its next task.

Source code in src/amr_hub_abm/agent.py

def perform_moving_to_task_location(self, current_time: int) -> bool:
    """Move the agent towards the location of its next task."""
    next_task = self.select_task_based_on_progress(TaskProgress.MOVING_TO_LOCATION)
    if next_task is None:
        return False
    next_task.update_progress(current_time=current_time, agent=self)
    return True

perform_suspended_task(current_time)

Perform a suspended task and return True if a task was performed.

Source code in src/amr_hub_abm/agent.py

def perform_suspended_task(self, current_time: int) -> bool:
    """Perform a suspended task and return True if a task was performed."""
    task = self.select_task_based_on_progress(
        TaskProgress.SUSPENDED, allow_multiple=True
    )
    if task is None:
        return False
    task.update_progress(current_time=current_time, agent=self)
    return True

perform_task(current_time, *, record=False)

Perform the agent's current task if it's due.

Source code in src/amr_hub_abm/agent.py

def perform_task(self, current_time: int, *, record: bool = False) -> None:
    """Perform the agent's current task if it's due."""
    if record:
        logger.info(
            "Recording state for Agent id %s at time %s: location=%s",
            self.idx,
            current_time,
            self.location,
        )
        self.record_state(current_time=current_time)

    if logger.isEnabledFor(logging.INFO):
        task_list_values = [task.task_type.value for task in self.tasks]
        task_progress_values = [task.progress.value for task in self.tasks]
        msg = f"Time {current_time} Task list: {task_list_values}"
        logger.info(msg)
        msg = f"Time {current_time} Task list: {task_progress_values}"
        logger.info(msg)

    if not self.tasks:
        return
    logger.debug(
        "Agent id %s has %s tasks to perform.",
        self.idx,
        len(self.tasks),
    )

    if self.perform_in_progress_task(current_time=current_time):
        return
    logger.debug(
        "No in-progress tasks for Agent id %s.",
        self.idx,
    )

    if self.perform_moving_to_task_location(current_time=current_time):
        return
    logger.debug(
        "No tasks to move to for Agent id %s.",
        self.idx,
    )

    if self.perform_suspended_task(current_time=current_time):
        return
    logger.debug(
        "No suspended tasks for Agent id %s.",
        self.idx,
    )

    if self.perform_to_be_started_task(current_time=current_time):
        return
    logger.debug(
        "No to-be-started tasks for Agent id %s.",
        self.idx,
    )

perform_to_be_started_task(current_time)

Perform a to-be-started task and return True if a task was performed.

Source code in src/amr_hub_abm/agent.py

def perform_to_be_started_task(self, current_time: int) -> bool:
    """Perform a to-be-started task and return True if a task was performed."""
    task = self.select_task_based_on_progress(
        TaskProgress.NOT_STARTED, allow_multiple=True
    )
    if task is None:
        return False
    if isinstance(task, TaskOccupyContent):
        task.assign_content()

    task_move_time = (
        task.time_due
        - task.time_needed
        - self.estimate_time_to_reach_location(task.location)
    )
    logger.info(
        "Agent id %s next task move time: %s, current time: %s",
        self.idx,
        task_move_time,
        current_time,
    )

    if current_time < task_move_time:
        self.attempt_task_insertion(
            next_task=task,
            next_task_move_time=task_move_time,
            current_time=current_time,
        )
        return False

    task.update_progress(current_time=current_time, agent=self)
    return True

plot_agent(ax, *, show_tags=True)

Plot the agent on the given axes.

Source code in src/amr_hub_abm/agent.py

def plot_agent(self, ax: Axes, *, show_tags: bool = True) -> None:
    """Plot the agent on the given axes."""
    ax.plot(
        self.location.x,
        self.location.y,
        marker="o",
        markersize=5,
        color=ROLE_COLOUR_MAP[self.agent_type],
    )

    if show_tags:
        ax.text(
            self.location.x + 0.05,
            self.location.y + 0.05,
            f"{self.agent_type.value} {self.idx}",
            fontsize=8,
            ha="left",
            va="bottom",
        )

plot_trajectory(ax)

Plot the agent's trajectory on the given axes.

Source code in src/amr_hub_abm/agent.py

def plot_trajectory(self, ax: Axes) -> None:
    """Plot the agent's trajectory on the given axes."""
    if self.trajectory_length == 0:
        msg = "Cannot plot trajectory for agent with trajectory_length of 0."
        raise ValueError(msg)

    ax.plot(
        self.trajectory.position[:, 0],
        self.trajectory.position[:, 1],
        linestyle="-",
        linewidth=1.5,
        color=ROLE_COLOUR_MAP[self.agent_type],
        alpha=0.7,
    )

propose_new_coordinates(coordinates, heading_rad, movement_speed, stochasticity, rng_generator) staticmethod

Propose a new location for agent movement.

Source code in src/amr_hub_abm/agent.py

@staticmethod
def propose_new_coordinates(
    coordinates: tuple[float, float],
    heading_rad: float,
    movement_speed: float,
    stochasticity: float,
    rng_generator: Generator,
) -> tuple[float, float]:
    """Propose a new location for agent movement."""
    delta_x = movement_speed * math.cos(heading_rad)
    delta_y = movement_speed * math.sin(heading_rad)

    delta_x = (1 + rng_generator.normal(0, stochasticity)) * delta_x
    delta_y = (1 + rng_generator.normal(0, stochasticity)) * delta_y

    new_x = coordinates[0] + delta_x
    new_y = coordinates[1] + delta_y

    return new_x, new_y

record_state(current_time)

Push a record of the agent's current state to the trajectory.

Source code in src/amr_hub_abm/agent.py

def record_state(self, current_time: int) -> None:
    """Push a record of the agent's current state to the trajectory."""
    if current_time >= self.trajectory_length:
        msg = f"Current time {current_time} "
        msg += f"exceeds trajectory length {self.trajectory_length}."
        raise ValueError(msg)

    building_idx_list = [
        b.idx for b in self.space if b.name == self.location.building
    ]
    if not building_idx_list:
        msg = f"Building {self.location.building} not found in agent's space."
        raise ValueError(msg)
    if len(building_idx_list) > 1:
        msg = f"Multiple buildings with name {self.location.building} found."
        raise ValueError(msg)
    building_idx = building_idx_list[0]

    self.trajectory.push(
        time=current_time,
        building_idx=building_idx,
        floor=self.location.floor,
        pos_x=self.location.x,
        pos_y=self.location.y,
        heading=self.heading_rad,
        infection_status=self.infection_status,
    )

select_task_based_on_progress(progress, *, allow_multiple=False)

Select a task based on its progress.

Source code in src/amr_hub_abm/agent.py

def select_task_based_on_progress(
    self, progress: TaskProgress, *, allow_multiple: bool = False
) -> Task | None:
    """Select a task based on its progress."""
    tasks = [task for task in self.tasks if task.progress == progress]
    if not tasks:
        return None
    if len(tasks) > 1 and not allow_multiple:
        msg = f"Agent {self.idx} has multiple tasks"
        msg += f" with progress {progress.value}."
        logger.error(msg)
        raise RuntimeError(msg)
    return min(tasks, key=lambda t: (t.time_due, t.priority.value))

try_move_one_step(stochasticity, max_attempts=5)

Return valid coordinates for a single movement step.

Source code in src/amr_hub_abm/agent.py

def try_move_one_step(
    self,
    stochasticity: float,
    max_attempts: int = 5,
) -> tuple[float, float]:
    """Return valid coordinates for a single movement step."""
    for attempt in range(1, max_attempts + 1):
        new_x, new_y = self.propose_new_coordinates(
            (self.location.x, self.location.y),
            self.heading_rad,
            self.movement_speed,
            stochasticity,
            self.rng_generator,
        )

        room = self.get_room((new_x, new_y))
        if room is None:
            logger.info(
                "Attempt %s: location (%s, %s) is not located in any room.",
                attempt,
                new_x,
                new_y,
            )
            continue

        walls = room.walls
        if not walls:
            msg = (
                f"Room {room.name} has no walls defined, "
                "cannot check for wall intersections."
            )
            raise SimulationModeError(msg)

        if Location.check_intersection_with_walls(
            new_x,
            new_y,
            self.interaction_radius,
            walls,
        ):
            logger.info(
                "Attempt %s: Agent id %s cannot move to (%s, %s): "
                "wall intersection.",
                attempt,
                self.idx,
                new_x,
                new_y,
            )
            continue

        return new_x, new_y

    logger.info(
        "Maximum attempts %s exceeded for moving one step. "
        "Agent id %s moving to proposed coordinates (%s, %s) despite "
        "wall intersection.",
        max_attempts,
        self.idx,
        self.location.x,
        self.location.y,
    )

    return new_x, new_y

AgentType

Bases: IntEnum

Enumeration of possible agent types.

Source code in src/amr_hub_abm/agent.py

class AgentType(IntEnum):
    """Enumeration of possible agent types."""

    GENERIC = 0
    PATIENT = 1
    HEALTHCARE_WORKER = 2

InfectionStatus

Bases: IntEnum

Enumeration of possible infection statuses.

Source code in src/amr_hub_abm/agent.py

class InfectionStatus(IntEnum):
    """Enumeration of possible infection statuses."""

    SUSCEPTIBLE = 0
    EXPOSED = 1
    INFECTED = 2
    RECOVERED = 3

Record dataclass

Representation of a record of an agent's state at a given time step.

Source code in src/amr_hub_abm/agent.py

@dataclass(slots=True)
class Record:
    """Representation of a record of an agent's state at a given time step."""

    total_time: int

    building: npt.NDArray[np.int8] = field(init=False)
    floor: npt.NDArray[np.int8] = field(init=False)
    position: npt.NDArray[np.float64] = field(init=False)
    heading: npt.NDArray[np.float64] = field(init=False)
    infection_status: npt.NDArray[np.int8] = field(init=False)

    def __post_init__(self) -> None:
        """Post-initialization to setup the record arrays."""
        self.building = np.empty(self.total_time, dtype=np.int8)
        self.floor = np.empty(self.total_time, dtype=np.int8)
        self.position = np.empty((self.total_time, 2), dtype=np.float64)
        self.heading = np.empty((self.total_time, 1), dtype=np.float64)
        self.infection_status = np.empty(self.total_time, dtype=np.int8)

    def push(  # noqa: PLR0913
        self,
        time: int,
        building_idx: int,
        floor: int,
        pos_x: float,
        pos_y: float,
        heading: float,
        infection_status: InfectionStatus,
    ) -> None:
        """Push a new record of the agent's state at a given time step."""
        if time >= self.total_time:
            msg = f"Time {time} exceeds total_time {self.total_time} for record."
            raise ValueError(msg)

        self.building[time] = building_idx
        self.floor[time] = floor
        self.heading[time] = heading
        self.position[time] = [pos_x, pos_y]
        self.infection_status[time] = infection_status.value

__post_init__()

Post-initialization to setup the record arrays.

Source code in src/amr_hub_abm/agent.py

def __post_init__(self) -> None:
    """Post-initialization to setup the record arrays."""
    self.building = np.empty(self.total_time, dtype=np.int8)
    self.floor = np.empty(self.total_time, dtype=np.int8)
    self.position = np.empty((self.total_time, 2), dtype=np.float64)
    self.heading = np.empty((self.total_time, 1), dtype=np.float64)
    self.infection_status = np.empty(self.total_time, dtype=np.int8)

push(time, building_idx, floor, pos_x, pos_y, heading, infection_status)

Push a new record of the agent's state at a given time step.

Source code in src/amr_hub_abm/agent.py

def push(  # noqa: PLR0913
    self,
    time: int,
    building_idx: int,
    floor: int,
    pos_x: float,
    pos_y: float,
    heading: float,
    infection_status: InfectionStatus,
) -> None:
    """Push a new record of the agent's state at a given time step."""
    if time >= self.total_time:
        msg = f"Time {time} exceeds total_time {self.total_time} for record."
        raise ValueError(msg)

    self.building[time] = building_idx
    self.floor[time] = floor
    self.heading[time] = heading
    self.position[time] = [pos_x, pos_y]
    self.infection_status[time] = infection_status.value

exceptions

Module defining all custom exceptions for the AMR Hub ABM simulation.

InvalidDefinitionError

Bases: Exception

Exception raised when a general invalid definition is encountered.

Source code in src/amr_hub_abm/exceptions.py

class InvalidDefinitionError(Exception):
    """Exception raised when a general invalid definition is encountered."""

    def __init__(self, message: str) -> None:
        """Initialize the InvalidDefinitionError."""
        super().__init__(f"Invalid definition: {message}.")

__init__(message)

Initialize the InvalidDefinitionError.

Source code in src/amr_hub_abm/exceptions.py

def __init__(self, message: str) -> None:
    """Initialize the InvalidDefinitionError."""
    super().__init__(f"Invalid definition: {message}.")

InvalidDistanceError

Bases: Exception

Exception raised when an invalid distance calculation is attempted.

Source code in src/amr_hub_abm/exceptions.py

class InvalidDistanceError(Exception):
    """Exception raised when an invalid distance calculation is attempted."""

    def __init__(self, locations: tuple, building: bool) -> None:  # noqa: FBT001
        """Initialize the InvalidDistanceError."""
        initial_string = "Invalid distance calculation between"

        if building:
            super().__init__(
                f"{initial_string} buildings: {locations[0]} and {locations[1]}."
            )
        else:
            super().__init__(
                f"{initial_string} floors: {locations[0]} and {locations[1]}."
            )

__init__(locations, building)

Initialize the InvalidDistanceError.

Source code in src/amr_hub_abm/exceptions.py

def __init__(self, locations: tuple, building: bool) -> None:  # noqa: FBT001
    """Initialize the InvalidDistanceError."""
    initial_string = "Invalid distance calculation between"

    if building:
        super().__init__(
            f"{initial_string} buildings: {locations[0]} and {locations[1]}."
        )
    else:
        super().__init__(
            f"{initial_string} floors: {locations[0]} and {locations[1]}."
        )

InvalidDoorError

Bases: Exception

Exception raised when a door is defined with invalid parameters.

Source code in src/amr_hub_abm/exceptions.py

class InvalidDoorError(Exception):
    """Exception raised when a door is defined with invalid parameters."""

    def __init__(self, message: str) -> None:
        """Initialize the InvalidDoorError."""
        super().__init__(f"Invalid door definition: {message}.")

__init__(message)

Initialize the InvalidDoorError.

Source code in src/amr_hub_abm/exceptions.py

def __init__(self, message: str) -> None:
    """Initialize the InvalidDoorError."""
    super().__init__(f"Invalid door definition: {message}.")

InvalidRoomError

Bases: Exception

Exception raised when a room is defined with invalid parameters.

Source code in src/amr_hub_abm/exceptions.py

class InvalidRoomError(Exception):
    """Exception raised when a room is defined with invalid parameters."""

    def __init__(self, message: str) -> None:
        """Initialize the InvalidRoomError."""
        super().__init__(f"Invalid room definition: {message}.")

__init__(message)

Initialize the InvalidRoomError.

Source code in src/amr_hub_abm/exceptions.py

def __init__(self, message: str) -> None:
    """Initialize the InvalidRoomError."""
    super().__init__(f"Invalid room definition: {message}.")

SimulationModeError

Bases: Exception

Exception raised when an operation is invalid for the current simulation mode.

Source code in src/amr_hub_abm/exceptions.py

class SimulationModeError(Exception):
    """Exception raised when an operation is invalid for the current simulation mode."""

    def __init__(self, message: str) -> None:
        """Initialize the SimulationModeError."""
        super().__init__(f"Simulation mode error: {message}.")

__init__(message)

Initialize the SimulationModeError.

Source code in src/amr_hub_abm/exceptions.py

def __init__(self, message: str) -> None:
    """Initialize the SimulationModeError."""
    super().__init__(f"Simulation mode error: {message}.")

TimeError

Bases: Exception

Exception raised when a negative time value is encountered.

Source code in src/amr_hub_abm/exceptions.py

class TimeError(Exception):
    """Exception raised when a negative time value is encountered."""

    def __init__(self, message: str) -> None:
        """Initialize the TimeError."""
        super().__init__(f"Invalid time value encountered: {message}.")

__init__(message)

Initialize the TimeError.

Source code in src/amr_hub_abm/exceptions.py

def __init__(self, message: str) -> None:
    """Initialize the TimeError."""
    super().__init__(f"Invalid time value encountered: {message}.")

read_space_input

Module to import space input data for the AMR Hub ABM simulation.

SpaceInputReader dataclass

Class to read space input data from a YAML file.

Source code in src/amr_hub_abm/read_space_input.py

@dataclass
class SpaceInputReader:
    """Class to read space input data from a YAML file."""

    input_path: Path
    rng_generator: Generator
    data: dict = field(init=False)

    door_list: list[Door] = field(init=False, default_factory=list)
    wall_list: list[Wall] = field(init=False, default_factory=list)

    room_door_dict: dict[str, list[DetachedDoor]] = field(
        init=False, default_factory=dict
    )

    rooms: list[Room] = field(init=False, default_factory=list)
    buildings: list[Building] = field(init=False, default_factory=list)

    topological: bool = field(init=False, default=False)

    def __post_init__(self) -> None:
        """Post-initialization to read and validate the YAML file."""
        self.validation()
        self.get_room_name_dict()
        self.create_doors_from_detatched_doors()
        self.create_rooms_from_data()

        for room in self.rooms:
            msg = f"Room '{room.name} (id {room.room_id})' created."
            logger.info(msg)

        self.buildings = self.organise_rooms_into_floors_and_buildings(self.rooms)

    def get_room_name_dict(self) -> None:
        """Extract room names and their corresponding detatched doors."""
        for building_data in [self.data["building"]]:
            for floor_data in building_data["floors"]:
                for room_data in floor_data["rooms"]:
                    self.topological = "area" in room_data
                    room_name = room_data["name"]
                    doors: list[DetachedDoor] = []

                    if not self.topological:
                        for door_data in room_data.get("doors", []):
                            door = DetachedDoor(
                                is_open=False,
                                access_control=(False, False),
                                start=(door_data[0], door_data[1]),
                                end=(door_data[2], door_data[3]),
                            )
                            doors.append(door)
                        self.room_door_dict[room_name] = doors

                    else:
                        for door_name in room_data.get("doors", []):
                            door = DetachedDoor(
                                is_open=False,
                                access_control=(False, False),
                                name=door_name,
                            )
                            doors.append(door)
                        self.room_door_dict[room_name] = doors

    def create_doors_from_detatched_doors(self) -> None:
        """Create Door instances from detatched doors."""
        detatced_door_list = [
            door for doors in self.room_door_dict.values() for door in doors
        ]
        unique_detatched_doors = set(detatced_door_list)
        sorted_unique_detatched_doors = sorted(
            unique_detatched_doors, key=lambda d: (d.start, d.end)
        )

        sorted_room_names = sorted(room_name for room_name in self.room_door_dict)

        for detatched_door in sorted_unique_detatched_doors:
            connecting_rooms = [
                sorted_room_names.index(room_name)
                for room_name, doors in self.room_door_dict.items()
                if detatched_door in doors
            ]

            if len(connecting_rooms) != 2:
                msg = (
                    f"Door at {detatched_door.start}-{detatched_door.end} must connect "
                    f"exactly two rooms. Found {len(connecting_rooms)}."
                )
                logger.error(msg)
                raise InvalidDoorError(msg)

            if self.topological:
                door = Door(
                    is_open=detatched_door.is_open,
                    access_control=detatched_door.access_control,
                    name=detatched_door.name,
                    connecting_rooms=(connecting_rooms[0], connecting_rooms[1]),
                    door_id=sorted_unique_detatched_doors.index(detatched_door),
                )
            else:
                door = Door(
                    is_open=detatched_door.is_open,
                    access_control=detatched_door.access_control,
                    start=detatched_door.start,
                    end=detatched_door.end,
                    connecting_rooms=(connecting_rooms[0], connecting_rooms[1]),
                    door_id=sorted_unique_detatched_doors.index(detatched_door),
                )

            self.door_list.append(door)

    @staticmethod
    def organise_rooms_into_floors_and_buildings(rooms: list[Room]) -> list[Building]:
        """Organize rooms into floors and buildings."""
        all_buildings = {room.building for room in rooms}
        buildings: list[Building] = []
        for building_name in all_buildings:
            building_rooms = [room for room in rooms if room.building == building_name]
            all_floors = {room.floor for room in building_rooms}
            floors: list[Floor] = []
            for floor_number in all_floors:
                floor_rooms = [
                    room for room in building_rooms if room.floor == floor_number
                ]
                floor = Floor(floor_number=floor_number, rooms=floor_rooms)
                msg = (
                    f"Floor {floor.floor_number} created with {len(floor.rooms)} rooms."
                )
                logger.info(msg)
                floors.append(floor)
            building = Building(name=building_name, floors=floors)
            msg = f"Building '{building.name}' created with {len(floors)} floors."
            logger.info(msg)
            buildings.append(building)

        return Building.sort_and_number_buildings(buildings)

    def create_rooms_from_data(self) -> None:
        """Create Room instances from the validated data."""
        sorted_room_names = sorted(room_name for room_name in self.room_door_dict)

        for index, name in enumerate(sorted_room_names):
            for building_data in [self.data["building"]]:
                for floor_data in building_data["floors"]:
                    floor_level = floor_data["level"]
                    for room_data in floor_data["rooms"]:
                        if room_data["name"] == name:
                            room = self.create_room(
                                room_data, index, building_data["name"], floor_level
                            )
                            self.rooms.append(room)

    def validation(self) -> None:
        """Validate the space input data from the YAML file."""
        with self.input_path.open("r", encoding="utf-8") as file:
            self.data = yaml.safe_load(file)

        msg = f"Loaded space input data from {self.input_path}"
        logger.info(msg)

        if "building" not in self.data:
            msg = "The input data must contain a 'building' key."
            logger.error(msg)
            raise KeyError(msg)

        building_data = self.data["building"]
        self.validate_building_data(building_data)

        floors_data = building_data["floors"]
        for floor_data in floors_data:
            self.validate_floor_data(floor_data)
            rooms_data = floor_data["rooms"]
            for room_data in rooms_data:
                self.validate_room_data(room_data)

    def create_room(
        self, room_data: dict, room_id: int, building_name: str, floor_level: int
    ) -> Room:
        """Create a Room instance from room data."""
        if self.topological:
            return self.create_topological_room(
                room_data, room_id, building_name, floor_level
            )
        return self.create_spatial_room(room_data, room_id, building_name, floor_level)

    def create_topological_room(
        self, room_data: dict, room_id: int, building_name: str, floor_level: int
    ) -> Room:
        """Create a topological Room instance from room data."""
        room_doors = [
            door for door in self.door_list if door.name in room_data["doors"]
        ]

        return Room(
            room_id=room_id,
            name=room_data["name"],
            building=building_name,
            floor=floor_level,
            walls=None,
            doors=room_doors,
            contents=room_data.get("contents", []),
            area=room_data["area"],
            rng_generator=self.rng_generator,
        )

    def create_spatial_room(
        self, room_data: dict, room_id: int, building_name: str, floor_level: int
    ) -> Room:
        """Create a spatial Room instance from room data."""
        room_doors = [
            door for door in self.door_list if room_id in door.connecting_rooms
        ]

        room_walls: list[Wall] = []
        for wall_data in room_data.get("walls", []):
            wall = Wall(
                start=(wall_data[0], wall_data[1]),
                end=(wall_data[2], wall_data[3]),
            )
            self.wall_list.append(wall)
            room_walls.append(wall)

        contents: list[Content] = []
        for content_data in room_data.get("contents", []):
            content_type = content_data["type"].upper()
            if content_type not in ContentType.__members__:
                msg = f"Invalid content type '{content_data['type']}' "
                msg += f"in room '{room_data['name']}'."
                logger.error(msg)
                raise InvalidDefinitionError(msg)

            content_location = Location(
                building=building_name,
                floor=floor_level,
                x=content_data["position"][0],
                y=content_data["position"][1],
            )
            content = Content(
                content_type=ContentType[content_data["type"].upper()],
                location=content_location,
            )
            contents.append(content)

        return Room(
            room_id=room_id,
            name=room_data["name"],
            building=building_name,
            floor=floor_level,
            walls=room_walls,
            doors=room_doors,
            contents=contents,
            rng_generator=self.rng_generator,
        )

    @staticmethod
    def validate_building_data(building_data: dict) -> None:
        """Validate the building data structure."""
        if "name" not in building_data:
            msg = "The 'building' data must contain a 'name' key."
            logger.error(msg)
            raise KeyError(msg)

        if "address" not in building_data:
            msg = "The 'building' data must contain an 'address' key."
            logger.error(msg)
            raise KeyError(msg)

        if "floors" not in building_data:
            msg = "The 'building' data must contain a 'floors' key."
            logger.error(msg)
            raise KeyError(msg)

    @staticmethod
    def validate_floor_data(floor_data: dict) -> None:
        """Validate the floor data structure."""
        if "level" not in floor_data:
            msg = "Each floor must have a 'level' defined."
            logger.error(msg)
            raise KeyError(msg)

        if "rooms" not in floor_data:
            msg = "Each floor must contain a 'rooms' key."
            logger.error(msg)
            raise KeyError(msg)

    @staticmethod
    def validate_room_data(room_data: dict) -> None:
        """Validate the room data structure."""
        if "name" not in room_data:
            msg = "Each room must have a 'name' defined."
            logger.error(msg)
            raise KeyError(msg)

        if "doors" not in room_data:
            msg = "Each room must have a 'doors' key."
            logger.error(msg)
            raise KeyError(msg)

        if "walls" not in room_data and "area" not in room_data:
            msg = "Each room must have either 'walls' or 'area' defined."
            logger.error(msg)
            raise KeyError(msg)

        topological = room_data.get("area", False)

        if topological and "walls" in room_data:
            msg = "A topological room cannot have walls defined."
            logger.error(msg)
            raise KeyError(msg)

        if not topological:
            walls: list[Wall] = []
            walls_data: list[list[float]] = room_data["walls"]
            for wall in walls_data:
                SpaceInputReader.check_tuple_length(wall, 4, "wall")
                walls.append(Wall(start=(wall[0], wall[1]), end=(wall[2], wall[3])))

            msg = f"Room '{room_data['name']}' walls validated successfully."
            logger.info(msg)

            doors_data: list[list[float]] = room_data["doors"]
            for door in doors_data:
                SpaceInputReader.check_tuple_length(door, 4, "door")

        else:
            doors_names: list[str] = room_data["doors"]
            for name in doors_names:
                if not isinstance(name, str):
                    msg = "In topological mode, doors must be defined by their names."
                    logger.error(msg)
                    raise InvalidDoorError(msg)

    @staticmethod
    def check_tuple_length(
        data_tuple: list[float], expected_length: int, data_type: str
    ) -> None:
        """Check if a data tuple has the expected length."""
        if data_type not in {"wall", "door"}:
            msg = f"data_type must be either 'wall' or 'door'. Got '{data_type}'."
            raise InvalidDefinitionError(msg)

        if len(data_tuple) != expected_length:
            msg = f"Each {data_type} must be defined by {expected_length} values."
            logger.error(msg)
            if data_type == "wall":
                raise InvalidRoomError(msg)
            raise InvalidDoorError(msg)

__post_init__()

Post-initialization to read and validate the YAML file.

Source code in src/amr_hub_abm/read_space_input.py

def __post_init__(self) -> None:
    """Post-initialization to read and validate the YAML file."""
    self.validation()
    self.get_room_name_dict()
    self.create_doors_from_detatched_doors()
    self.create_rooms_from_data()

    for room in self.rooms:
        msg = f"Room '{room.name} (id {room.room_id})' created."
        logger.info(msg)

    self.buildings = self.organise_rooms_into_floors_and_buildings(self.rooms)

check_tuple_length(data_tuple, expected_length, data_type) staticmethod

Check if a data tuple has the expected length.

Source code in src/amr_hub_abm/read_space_input.py

@staticmethod
def check_tuple_length(
    data_tuple: list[float], expected_length: int, data_type: str
) -> None:
    """Check if a data tuple has the expected length."""
    if data_type not in {"wall", "door"}:
        msg = f"data_type must be either 'wall' or 'door'. Got '{data_type}'."
        raise InvalidDefinitionError(msg)

    if len(data_tuple) != expected_length:
        msg = f"Each {data_type} must be defined by {expected_length} values."
        logger.error(msg)
        if data_type == "wall":
            raise InvalidRoomError(msg)
        raise InvalidDoorError(msg)

create_doors_from_detatched_doors()

Create Door instances from detatched doors.

Source code in src/amr_hub_abm/read_space_input.py

def create_doors_from_detatched_doors(self) -> None:
    """Create Door instances from detatched doors."""
    detatced_door_list = [
        door for doors in self.room_door_dict.values() for door in doors
    ]
    unique_detatched_doors = set(detatced_door_list)
    sorted_unique_detatched_doors = sorted(
        unique_detatched_doors, key=lambda d: (d.start, d.end)
    )

    sorted_room_names = sorted(room_name for room_name in self.room_door_dict)

    for detatched_door in sorted_unique_detatched_doors:
        connecting_rooms = [
            sorted_room_names.index(room_name)
            for room_name, doors in self.room_door_dict.items()
            if detatched_door in doors
        ]

        if len(connecting_rooms) != 2:
            msg = (
                f"Door at {detatched_door.start}-{detatched_door.end} must connect "
                f"exactly two rooms. Found {len(connecting_rooms)}."
            )
            logger.error(msg)
            raise InvalidDoorError(msg)

        if self.topological:
            door = Door(
                is_open=detatched_door.is_open,
                access_control=detatched_door.access_control,
                name=detatched_door.name,
                connecting_rooms=(connecting_rooms[0], connecting_rooms[1]),
                door_id=sorted_unique_detatched_doors.index(detatched_door),
            )
        else:
            door = Door(
                is_open=detatched_door.is_open,
                access_control=detatched_door.access_control,
                start=detatched_door.start,
                end=detatched_door.end,
                connecting_rooms=(connecting_rooms[0], connecting_rooms[1]),
                door_id=sorted_unique_detatched_doors.index(detatched_door),
            )

        self.door_list.append(door)

create_room(room_data, room_id, building_name, floor_level)

Create a Room instance from room data.

Source code in src/amr_hub_abm/read_space_input.py

def create_room(
    self, room_data: dict, room_id: int, building_name: str, floor_level: int
) -> Room:
    """Create a Room instance from room data."""
    if self.topological:
        return self.create_topological_room(
            room_data, room_id, building_name, floor_level
        )
    return self.create_spatial_room(room_data, room_id, building_name, floor_level)

create_rooms_from_data()

Create Room instances from the validated data.

Source code in src/amr_hub_abm/read_space_input.py

def create_rooms_from_data(self) -> None:
    """Create Room instances from the validated data."""
    sorted_room_names = sorted(room_name for room_name in self.room_door_dict)

    for index, name in enumerate(sorted_room_names):
        for building_data in [self.data["building"]]:
            for floor_data in building_data["floors"]:
                floor_level = floor_data["level"]
                for room_data in floor_data["rooms"]:
                    if room_data["name"] == name:
                        room = self.create_room(
                            room_data, index, building_data["name"], floor_level
                        )
                        self.rooms.append(room)

create_spatial_room(room_data, room_id, building_name, floor_level)

Create a spatial Room instance from room data.

Source code in src/amr_hub_abm/read_space_input.py

def create_spatial_room(
    self, room_data: dict, room_id: int, building_name: str, floor_level: int
) -> Room:
    """Create a spatial Room instance from room data."""
    room_doors = [
        door for door in self.door_list if room_id in door.connecting_rooms
    ]

    room_walls: list[Wall] = []
    for wall_data in room_data.get("walls", []):
        wall = Wall(
            start=(wall_data[0], wall_data[1]),
            end=(wall_data[2], wall_data[3]),
        )
        self.wall_list.append(wall)
        room_walls.append(wall)

    contents: list[Content] = []
    for content_data in room_data.get("contents", []):
        content_type = content_data["type"].upper()
        if content_type not in ContentType.__members__:
            msg = f"Invalid content type '{content_data['type']}' "
            msg += f"in room '{room_data['name']}'."
            logger.error(msg)
            raise InvalidDefinitionError(msg)

        content_location = Location(
            building=building_name,
            floor=floor_level,
            x=content_data["position"][0],
            y=content_data["position"][1],
        )
        content = Content(
            content_type=ContentType[content_data["type"].upper()],
            location=content_location,
        )
        contents.append(content)

    return Room(
        room_id=room_id,
        name=room_data["name"],
        building=building_name,
        floor=floor_level,
        walls=room_walls,
        doors=room_doors,
        contents=contents,
        rng_generator=self.rng_generator,
    )

create_topological_room(room_data, room_id, building_name, floor_level)

Create a topological Room instance from room data.

Source code in src/amr_hub_abm/read_space_input.py

def create_topological_room(
    self, room_data: dict, room_id: int, building_name: str, floor_level: int
) -> Room:
    """Create a topological Room instance from room data."""
    room_doors = [
        door for door in self.door_list if door.name in room_data["doors"]
    ]

    return Room(
        room_id=room_id,
        name=room_data["name"],
        building=building_name,
        floor=floor_level,
        walls=None,
        doors=room_doors,
        contents=room_data.get("contents", []),
        area=room_data["area"],
        rng_generator=self.rng_generator,
    )

get_room_name_dict()

Extract room names and their corresponding detatched doors.

Source code in src/amr_hub_abm/read_space_input.py

def get_room_name_dict(self) -> None:
    """Extract room names and their corresponding detatched doors."""
    for building_data in [self.data["building"]]:
        for floor_data in building_data["floors"]:
            for room_data in floor_data["rooms"]:
                self.topological = "area" in room_data
                room_name = room_data["name"]
                doors: list[DetachedDoor] = []

                if not self.topological:
                    for door_data in room_data.get("doors", []):
                        door = DetachedDoor(
                            is_open=False,
                            access_control=(False, False),
                            start=(door_data[0], door_data[1]),
                            end=(door_data[2], door_data[3]),
                        )
                        doors.append(door)
                    self.room_door_dict[room_name] = doors

                else:
                    for door_name in room_data.get("doors", []):
                        door = DetachedDoor(
                            is_open=False,
                            access_control=(False, False),
                            name=door_name,
                        )
                        doors.append(door)
                    self.room_door_dict[room_name] = doors

organise_rooms_into_floors_and_buildings(rooms) staticmethod

Organize rooms into floors and buildings.

Source code in src/amr_hub_abm/read_space_input.py

@staticmethod
def organise_rooms_into_floors_and_buildings(rooms: list[Room]) -> list[Building]:
    """Organize rooms into floors and buildings."""
    all_buildings = {room.building for room in rooms}
    buildings: list[Building] = []
    for building_name in all_buildings:
        building_rooms = [room for room in rooms if room.building == building_name]
        all_floors = {room.floor for room in building_rooms}
        floors: list[Floor] = []
        for floor_number in all_floors:
            floor_rooms = [
                room for room in building_rooms if room.floor == floor_number
            ]
            floor = Floor(floor_number=floor_number, rooms=floor_rooms)
            msg = (
                f"Floor {floor.floor_number} created with {len(floor.rooms)} rooms."
            )
            logger.info(msg)
            floors.append(floor)
        building = Building(name=building_name, floors=floors)
        msg = f"Building '{building.name}' created with {len(floors)} floors."
        logger.info(msg)
        buildings.append(building)

    return Building.sort_and_number_buildings(buildings)

validate_building_data(building_data) staticmethod

Validate the building data structure.

Source code in src/amr_hub_abm/read_space_input.py

@staticmethod
def validate_building_data(building_data: dict) -> None:
    """Validate the building data structure."""
    if "name" not in building_data:
        msg = "The 'building' data must contain a 'name' key."
        logger.error(msg)
        raise KeyError(msg)

    if "address" not in building_data:
        msg = "The 'building' data must contain an 'address' key."
        logger.error(msg)
        raise KeyError(msg)

    if "floors" not in building_data:
        msg = "The 'building' data must contain a 'floors' key."
        logger.error(msg)
        raise KeyError(msg)

validate_floor_data(floor_data) staticmethod

Validate the floor data structure.

Source code in src/amr_hub_abm/read_space_input.py

@staticmethod
def validate_floor_data(floor_data: dict) -> None:
    """Validate the floor data structure."""
    if "level" not in floor_data:
        msg = "Each floor must have a 'level' defined."
        logger.error(msg)
        raise KeyError(msg)

    if "rooms" not in floor_data:
        msg = "Each floor must contain a 'rooms' key."
        logger.error(msg)
        raise KeyError(msg)

validate_room_data(room_data) staticmethod

Validate the room data structure.

Source code in src/amr_hub_abm/read_space_input.py

@staticmethod
def validate_room_data(room_data: dict) -> None:
    """Validate the room data structure."""
    if "name" not in room_data:
        msg = "Each room must have a 'name' defined."
        logger.error(msg)
        raise KeyError(msg)

    if "doors" not in room_data:
        msg = "Each room must have a 'doors' key."
        logger.error(msg)
        raise KeyError(msg)

    if "walls" not in room_data and "area" not in room_data:
        msg = "Each room must have either 'walls' or 'area' defined."
        logger.error(msg)
        raise KeyError(msg)

    topological = room_data.get("area", False)

    if topological and "walls" in room_data:
        msg = "A topological room cannot have walls defined."
        logger.error(msg)
        raise KeyError(msg)

    if not topological:
        walls: list[Wall] = []
        walls_data: list[list[float]] = room_data["walls"]
        for wall in walls_data:
            SpaceInputReader.check_tuple_length(wall, 4, "wall")
            walls.append(Wall(start=(wall[0], wall[1]), end=(wall[2], wall[3])))

        msg = f"Room '{room_data['name']}' walls validated successfully."
        logger.info(msg)

        doors_data: list[list[float]] = room_data["doors"]
        for door in doors_data:
            SpaceInputReader.check_tuple_length(door, 4, "door")

    else:
        doors_names: list[str] = room_data["doors"]
        for name in doors_names:
            if not isinstance(name, str):
                msg = "In topological mode, doors must be defined by their names."
                logger.error(msg)
                raise InvalidDoorError(msg)

validation()

Validate the space input data from the YAML file.

Source code in src/amr_hub_abm/read_space_input.py

def validation(self) -> None:
    """Validate the space input data from the YAML file."""
    with self.input_path.open("r", encoding="utf-8") as file:
        self.data = yaml.safe_load(file)

    msg = f"Loaded space input data from {self.input_path}"
    logger.info(msg)

    if "building" not in self.data:
        msg = "The input data must contain a 'building' key."
        logger.error(msg)
        raise KeyError(msg)

    building_data = self.data["building"]
    self.validate_building_data(building_data)

    floors_data = building_data["floors"]
    for floor_data in floors_data:
        self.validate_floor_data(floor_data)
        rooms_data = floor_data["rooms"]
        for room_data in rooms_data:
            self.validate_room_data(room_data)

run

Module to run the AMR Hub ABM simulation.

run_steps(simulation, plot_path, *, record)

Run the simulation steps until completion.

Source code in src/amr_hub_abm/run.py

def run_steps(
    simulation: Simulation,
    plot_path: Path | None,
    *,
    record: bool,
) -> None:
    """Run the simulation steps until completion."""
    while simulation.time < simulation.total_simulation_time:
        simulation.step(
            plot_path=plot_path,
            record=record,
        )

simulate(*, plot=False, record=False, plot_trajectory=False)

Simulate the AMR Hub ABM based on a configuration file.

Source code in src/amr_hub_abm/run.py

def simulate(
    *, plot: bool = False, record: bool = False, plot_trajectory: bool = False
) -> None:
    """Simulate the AMR Hub ABM based on a configuration file."""
    config_path = Path("tests/inputs/simulation_config.yml")
    simulation = create_simulation(config_path)
    if plot:
        output_dir = Path("../simulation_outputs")
        output_dir.mkdir(parents=True, exist_ok=True)

    logger.info([room.doors for room in simulation.space[0].floors[0].rooms])
    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 created successfully...")

    plot_path = Path("../simulation_outputs") if plot else None

    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")
        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)

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

simulation

The main simulation module for the AMR Hub ABM.

Simulation dataclass

Representation of the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/simulation.py

@dataclass
class Simulation:
    """Representation of the AMR Hub ABM simulation."""

    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)

    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)

        for agent in self.agents:
            agent.perform_task(current_time=self.time, record=record)

        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."""
        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]]
            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 __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."""
        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."""
        self.plot_current_state(directory_path=output_file.parent, trajectory=True)

rooms property

Get all rooms in the simulation space.

__repr__()

Representation of the simulation.

Source code in src/amr_hub_abm/simulation.py

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}"

plot_agent_trajectories(output_file)

Plot the trajectories of all agents from a recorded CSV file.

Source code in src/amr_hub_abm/simulation.py

def plot_agent_trajectories(self, output_file: Path) -> None:
    """Plot the trajectories of all agents from a recorded CSV file."""
    self.plot_current_state(directory_path=output_file.parent, trajectory=True)

plot_current_state(directory_path, *, trajectory=False)

Plot the current state of the simulation.

Source code in src/amr_hub_abm/simulation.py

def plot_current_state(
    self, directory_path: Path, *, trajectory: bool = False
) -> None:
    """Plot the current state of the simulation."""
    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]]
        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()

record_agent_states(file_path)

Record the states of all agents at the current time step to a CSV file.

Source code in src/amr_hub_abm/simulation.py

def record_agent_states(self, file_path: Path) -> None:
    """Record the states of all agents at the current time step to a CSV file."""
    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="",
        )

step(plot_path=None, *, record=False)

Advance the simulation by one time step.

Source code in src/amr_hub_abm/simulation.py

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)

    for agent in self.agents:
        agent.perform_task(current_time=self.time, record=record)

    if plot_path is not None:
        self.plot_current_state(directory_path=plot_path)

    self.time += 1

SimulationMode

Bases: IntEnum

Enumeration of simulation modes.

Source code in src/amr_hub_abm/simulation.py

class SimulationMode(IntEnum):
    """Enumeration of simulation modes."""

    SPATIAL = 0
    TOPOLOGICAL = 1

simulation_factory

Module for creating simulation instances.

create_simulation(config_file)

Create a simulation instance from a configuration file.

Parameters:

Name	Type	Description	Default
config_file	Path	Path to the configuration file.	required

Returns:

Name	Type	Description
Simulation	Simulation	An instance of the Simulation class.

Source code in src/amr_hub_abm/simulation_factory.py

def create_simulation(config_file: Path) -> Simulation:
    """
    Create a simulation instance from a configuration file.

    Args:
        config_file (Path): Path to the configuration file.

    Returns:
        Simulation: An instance of the Simulation class.

    """
    if not config_file.exists():
        msg = f"Configuration file not found: {config_file}"
        raise FileNotFoundError(msg)

    with config_file.open(encoding="utf-8") as file:
        config_data = yaml.safe_load(file)

    rng_generator = np.random.default_rng()

    buildings_path = Path(config_data["buildings_path"])
    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(config_data["start_time"])
    end_time = pd.to_datetime(config_data["end_time"])
    total_seconds = (end_time - start_time).total_seconds()
    time_step_length_seconds = config_data["length_of_timestep_in_seconds"]
    total_steps = int(total_seconds // time_step_length_seconds)
    logger.info("Total simulation time steps: %d", total_steps)

    timeseries_data = read_location_timeseries(
        file_path=Path(config_data["location_timeseries_path"])
    )

    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,
    )

    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,
    )

create_space_from_rooms(rooms)

Create a list of Building instances from a list of Room instances.

Source code in src/amr_hub_abm/simulation_factory.py

def create_space_from_rooms(rooms: list[Room]) -> list[Building]:
    """Create a list of Building instances from a list of Room instances."""
    building_dict: dict[str, Building] = {}

    for room in rooms:
        if room.building not in building_dict:
            building_dict[room.building] = Building(name=room.building, floors=[])

        if room.floor not in [
            f.floor_number for f in building_dict[room.building].floors
        ]:
            building_dict[room.building].floors.append(
                Floor(floor_number=room.floor, rooms=[room])
            )
        else:
            for floor in building_dict[room.building].floors:
                if floor.floor_number == room.floor:
                    floor.rooms.append(room)
                    break

    raw_buildings = list(building_dict.values())
    return Building.sort_and_number_buildings(raw_buildings)

get_random_location(room, building, floor)

Get a random location within a 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."""
    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
str	tuple[str, int, str]: A tuple containing the building name, floor number,
int	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.

    Args:
        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)

Parse a CSV file containing location time series data for agents.

Parameters:

Name	Type	Description	Default
timeseries_data	DataFrame	DataFrame containing the location time series.	required

Returns:

Type	Description
list[Agent]	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,
) -> list[Agent]:
    """
    Parse a CSV file containing location time series data for agents.

    Args:
        timeseries_data (pd.DataFrame): DataFrame containing the location time series.

    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 = row["timestamp"]
        location_str = row["location"]
        patient_id = int(row["patient_id"]) if row["patient_id"] != "-" else None
        event_type = row["event_type"]
        door_id = int(row["door_id"]) if row["door_id"] != "-" else None

        timestep = pd.to_datetime(timestamp)
        timestep_index = timestamp_to_timestep(
            timestep, 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:
            update_patient(
                patient_id=patient_id,
                space_tuple=(building, floor, room),
                patient_dict=patient_dict,
                total_time_steps=total_time_steps,
                space=create_space_from_rooms(rooms),
                rng_generator=rng_generator,
            )
            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":
            possible_locations = [
                c.position
                for c in room.contents
                if c.content_type == ContentType.WORKSTATION
            ]
            if not possible_locations:
                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()]

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

        elif event_type == "occupy_content":
            content_type = (
                int(row["content_type"]) if row["content_type"] != "-" else None
            )
            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"] = 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,
            space=create_space_from_rooms(rooms),
            rng_generator=rng_generator,
        )

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

read_location_timeseries(file_path)

Read a CSV file containing location time series data for agents.

Parameters:

Name	Type	Description	Default
file_path	Path	Path to the CSV file.	required

Returns:

Type	Description
DataFrame	pd.DataFrame: DataFrame containing the location time series data.

Source code in src/amr_hub_abm/simulation_factory.py

def read_location_timeseries(
    file_path: Path,
) -> pd.DataFrame:
    """
    Read a CSV file containing location time series data for agents.

    Args:
        file_path (Path): Path to the CSV file.

    Returns:
        pd.DataFrame: DataFrame containing the location time series data.

    """
    if not file_path.exists():
        msg = f"Location time series file not found: {file_path}"
        raise FileNotFoundError(msg)

    return pd.read_csv(file_path)

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:

Name	Type	Description
int	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.

    Args:
        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, space, rng_generator, additional_info=None)

Update healthcare worker information from data.

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,
    space: list[Building],
    rng_generator: np.random.Generator,
    additional_info: dict | None = None,
) -> None:
    """Update healthcare worker information from data."""
    building, floor, room = space_tuple
    location, timestep_index, event_type = event_tuple

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

        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

        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,
            space=space,
            rng_generator=rng_generator,
        )

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

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

Update patient information from data.

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,
    space: list[Building],
    rng_generator: np.random.Generator,
) -> None:
    """Update patient information from data."""
    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 not b.occupied
        ]
        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

        patient_dict[patient_id] = Agent(
            idx=patient_id,
            location=location,
            heading_rad=0.0,
            agent_type=AgentType.PATIENT,
            trajectory_length=total_time_steps,
            space=space,
            rng_generator=rng_generator,
        )

space

Space module for the AMR Hub ABM simulation.

building

Module containing building representation for the AMR Hub ABM simulation.

Building dataclass

Representation of a building in the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/space/building.py

@dataclass
class Building:
    """Representation of a building in the AMR Hub ABM simulation."""

    idx: int = field(init=False, repr=False)
    name: str
    floors: list[Floor]

    def plot_building(
        self,
        axes: list[Axes],
        agents: list[Agent] | None = None,
        *,
        trajectory: bool = False,
    ) -> None:
        """Plot the building layout."""
        for floor, ax in zip(self.floors, axes, strict=True):
            floor.plot(ax=ax, agents=agents, trajectory=trajectory)

    @staticmethod
    def sort_and_number_buildings(buildings: list[Building]) -> list[Building]:
        """Sort and number buildings by name."""
        sorted_buildings = sorted(buildings, key=lambda b: b.name)
        for i, building in enumerate(sorted_buildings):
            building.idx = i
        return sorted_buildings

plot_building(axes, agents=None, *, trajectory=False)

Plot the building layout.

Source code in src/amr_hub_abm/space/building.py

def plot_building(
    self,
    axes: list[Axes],
    agents: list[Agent] | None = None,
    *,
    trajectory: bool = False,
) -> None:
    """Plot the building layout."""
    for floor, ax in zip(self.floors, axes, strict=True):
        floor.plot(ax=ax, agents=agents, trajectory=trajectory)

sort_and_number_buildings(buildings) staticmethod

Sort and number buildings by name.

Source code in src/amr_hub_abm/space/building.py

@staticmethod
def sort_and_number_buildings(buildings: list[Building]) -> list[Building]:
    """Sort and number buildings by name."""
    sorted_buildings = sorted(buildings, key=lambda b: b.name)
    for i, building in enumerate(sorted_buildings):
        building.idx = i
    return sorted_buildings

content

Module defining room content types for the rooms of the AMR Hub ABM simulation.

Content dataclass

Enumeration of possible room contents.

Source code in src/amr_hub_abm/space/content.py

@dataclass
class Content:
    """Enumeration of possible room contents."""

    content_id: int = field(init=False)
    content_type: ContentType
    location: Location
    color: str = field(init=False)
    size: tuple[float, float] = field(init=False)
    occupier_id: tuple[int, AgentType] | None = field(default=None)

    marker_type: str = field(init=False, default="s")
    marker_size: int = field(init=False, default=100)

    def __post_init__(self) -> None:
        """Post-initialization to set content_id based on content_type and position."""
        self.content_id = hash((self.content_type, self.position))

        self.color = CONTENT_COLORS[self.content_type]
        self.size = CONTENT_SIZES[self.content_type]

    @property
    def length(self) -> float:
        """Get the length of the content based on its type."""
        return self.size[0]

    @property
    def width(self) -> float:
        """Get the width of the content based on its type."""
        return self.size[1]

    @property
    def polygon(self) -> shapely.geometry.Polygon:
        """Get the polygon representation of the content."""
        x, y = self.position
        length, width = self.size
        return shapely.geometry.box(
            x - length / 2, y - width / 2, x + length / 2, y + width / 2
        )

    @property
    def occupied(self) -> bool:
        """Check if the content is currently occupied by an agent."""
        return self.occupier_id is not None

    @property
    def position(self) -> tuple[float, float]:
        """Get the (x, y) position of the content."""
        return (self.location.x, self.location.y)

length property

Get the length of the content based on its type.

occupied property

Check if the content is currently occupied by an agent.

polygon property

Get the polygon representation of the content.

position property

Get the (x, y) position of the content.

width property

Get the width of the content based on its type.

__post_init__()

Post-initialization to set content_id based on content_type and position.

Source code in src/amr_hub_abm/space/content.py

def __post_init__(self) -> None:
    """Post-initialization to set content_id based on content_type and position."""
    self.content_id = hash((self.content_type, self.position))

    self.color = CONTENT_COLORS[self.content_type]
    self.size = CONTENT_SIZES[self.content_type]

ContentType

Bases: IntEnum

Enumeration of possible room content types.

Source code in src/amr_hub_abm/space/content.py

class ContentType(IntEnum):
    """Enumeration of possible room content types."""

    BED = 0
    WORKSTATION = 1
    CHAIR = 2

door

Module defining door representation for the AMR Hub ABM simulation.

DetachedDoor dataclass

Representation of a detached door in the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/space/door.py

@dataclass(kw_only=True, frozen=True)
class DetachedDoor:
    """Representation of a detached door in the AMR Hub ABM simulation."""

    is_open: bool
    access_control: tuple[bool, bool]
    name: str | None = field(default=None)
    start: tuple[float, float] | None = field(default=None)
    end: tuple[float, float] | None = field(default=None)

    def _identity_key(self) -> tuple[object, ...]:
        """Key used for equality + hashing. Ignores mutable state."""
        if self.name is not None:
            return ("name", self.name)
        # at this point start/end are both not None due to validation

        if self.start is None or self.end is None:
            msg = "Cannot create identity key from door without name or coordinates."
            raise InvalidDoorError(msg)
        return ("coords", self.start, self.end)

    def __eq__(self, other: object) -> bool:
        """Define equality comparison for DetachedDoor instances."""
        if not isinstance(other, DetachedDoor):
            return NotImplemented
        return self._identity_key() == other._identity_key()

    def __hash__(self) -> int:
        """Define hash for DetachedDoor instances."""
        return hash(self._identity_key())

    def check_for_start_end_consistency(self) -> None:
        """Check that start and end points are consistent."""
        if (self.start is None) != (self.end is None):
            msg = "Both start and end points must be None or both must be defined."
            raise InvalidDoorError(msg)

        if (self.start is not None and self.end is not None) and self.start == self.end:
            msg = "Door start and end points cannot be the same."
            raise InvalidDoorError(msg)

    def __post_init__(self) -> None:
        """Post-initialization to validate door coordinates."""
        if self.start is None and self.end is None:
            # If both start and end are None, we are in topological mode
            # and don't need to check coordinates. We just need to ensure that the door
            # has a name for identity purposes.
            if self.name is None:
                msg = "Door must have a name if start and end points are not defined."
                raise InvalidDoorError(msg)
            return

        self.check_for_start_end_consistency()

        if self.start is None or self.end is None:
            # This should never happen due to the consistency check, but we check again
            # to address mypy's anger issues.
            msg = "Both start and end points must be defined when in spatial mode."
            raise InvalidDoorError(msg)

        if self.start > self.end:
            temp = self.start
            object.__setattr__(self, "start", self.end)
            object.__setattr__(self, "end", temp)

__eq__(other)

Define equality comparison for DetachedDoor instances.

Source code in src/amr_hub_abm/space/door.py

def __eq__(self, other: object) -> bool:
    """Define equality comparison for DetachedDoor instances."""
    if not isinstance(other, DetachedDoor):
        return NotImplemented
    return self._identity_key() == other._identity_key()

__hash__()

Define hash for DetachedDoor instances.

Source code in src/amr_hub_abm/space/door.py

def __hash__(self) -> int:
    """Define hash for DetachedDoor instances."""
    return hash(self._identity_key())

__post_init__()

Post-initialization to validate door coordinates.

Source code in src/amr_hub_abm/space/door.py

def __post_init__(self) -> None:
    """Post-initialization to validate door coordinates."""
    if self.start is None and self.end is None:
        # If both start and end are None, we are in topological mode
        # and don't need to check coordinates. We just need to ensure that the door
        # has a name for identity purposes.
        if self.name is None:
            msg = "Door must have a name if start and end points are not defined."
            raise InvalidDoorError(msg)
        return

    self.check_for_start_end_consistency()

    if self.start is None or self.end is None:
        # This should never happen due to the consistency check, but we check again
        # to address mypy's anger issues.
        msg = "Both start and end points must be defined when in spatial mode."
        raise InvalidDoorError(msg)

    if self.start > self.end:
        temp = self.start
        object.__setattr__(self, "start", self.end)
        object.__setattr__(self, "end", temp)

check_for_start_end_consistency()

Check that start and end points are consistent.

Source code in src/amr_hub_abm/space/door.py

def check_for_start_end_consistency(self) -> None:
    """Check that start and end points are consistent."""
    if (self.start is None) != (self.end is None):
        msg = "Both start and end points must be None or both must be defined."
        raise InvalidDoorError(msg)

    if (self.start is not None and self.end is not None) and self.start == self.end:
        msg = "Door start and end points cannot be the same."
        raise InvalidDoorError(msg)

Door dataclass

Bases: DetachedDoor

Representation of a door in the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/space/door.py

@dataclass(eq=False, kw_only=True, frozen=True)
class Door(DetachedDoor):
    """Representation of a door in the AMR Hub ABM simulation."""

    connecting_rooms: tuple[int, int]
    door_id: int

    def __lt__(self, other: object) -> bool:
        """Define less-than comparison for Door instances."""
        if not isinstance(other, Door):
            return NotImplemented
        return self._identity_key() < other._identity_key()

    def __post_init__(self) -> None:
        """Post-initialization to validate door coordinates and create hash."""
        super().__post_init__()

    @property
    def line(self) -> shapely.geometry.LineString:
        """Get the line representation of the door."""
        if self.start is None or self.end is None:
            msg = "Door start and end must be defined when not in topological mode."
            raise InvalidDoorError(msg)
        return shapely.geometry.LineString([self.start, self.end])

line property

Get the line representation of the door.

__lt__(other)

Define less-than comparison for Door instances.

Source code in src/amr_hub_abm/space/door.py

def __lt__(self, other: object) -> bool:
    """Define less-than comparison for Door instances."""
    if not isinstance(other, Door):
        return NotImplemented
    return self._identity_key() < other._identity_key()

__post_init__()

Post-initialization to validate door coordinates and create hash.

Source code in src/amr_hub_abm/space/door.py

def __post_init__(self) -> None:
    """Post-initialization to validate door coordinates and create hash."""
    super().__post_init__()

floor

Module for Floor class.

Floor dataclass

Representation of a floor in a building.

Source code in src/amr_hub_abm/space/floor.py

@dataclass
class Floor:
    """Representation of a floor in a building."""

    floor_number: int
    rooms: list[Room]
    pseudo_rooms: list[Room] = field(init=False, default_factory=list)

    def __post_init__(self) -> None:
        """Post-initialization to validate floor attributes."""
        room_ids = [room.room_id for room in self.rooms]
        if len(room_ids) != len(set(room_ids)):
            msg = f"Duplicate room IDs found on floor {self.floor_number}."
            raise InvalidRoomError(msg)

    @property
    def room_ids(self) -> list[int]:
        """Get a list of room IDs on the floor."""
        return sorted([room.room_id for room in self.rooms])

    @property
    def room_names(self) -> list[str]:
        """Get a list of room names on the floor."""
        id_name_map = {room.room_id: room.name for room in self.rooms}
        return [id_name_map[room_id] for room_id in self.room_ids]

    @property
    def edge_set(self) -> set[tuple[int, int]]:
        """Get a list of all wall edges on the floor."""
        edges = set()
        for room in self.rooms:
            for door in room.doors:
                edges.add(door.connecting_rooms)
                edges.add((door.connecting_rooms[1], door.connecting_rooms[0]))
        return edges

    @property
    def adjacency_matrix(self) -> np.ndarray:
        """Get the adjacency matrix representing room connections on the floor."""
        n = len(self.rooms)
        room_id_to_index = {
            room_id: index for index, room_id in enumerate(self.room_ids)
        }
        adjacency_matrix = np.zeros((n, n), dtype=int)

        for edge in self.edge_set:
            room1, room2 = edge
            index1 = room_id_to_index[room1]
            index2 = room_id_to_index[room2]
            adjacency_matrix[index1, index2] = 1

        return adjacency_matrix

    def plot(
        self, ax: Axes, agents: list[Agent] | None = None, *, trajectory: bool = False
    ) -> None:
        """Plot the floor layout including rooms and doors."""
        for room in self.rooms:
            room.plot(ax=ax, agents=agents, trajectory=trajectory)

    def add_pseudo_rooms(self) -> None:
        """Add pseudo-rooms to the floor."""
        for existing_room in self.rooms:
            if not existing_room.walls:
                pseudo_room = Floor.create_spatial_room_from_pseudo_room(existing_room)
                self.pseudo_rooms.append(pseudo_room)

    @staticmethod
    def create_spatial_room_from_pseudo_room(room: Room) -> Room:
        """
        Create a spatial room from a pseudo-room based on area.

        Args:
            room (Room): The pseudo-room to convert.

        Returns:
            Room: A spatial room with walls and doors based on the pseudo-room's area.

        Raises:
            InvalidRoomError: If the pseudo-room does not have a valid positive area.

        Note:
            This method generates a rectangular room layout based on the area of the
            pseudo-room. The doors are positioned along one side of the rectangle. This
            function is not complete. Currently it only creates single simple
            rectangular rooms. Due to possibly complex topology, rectangular rooms may
            not always be a valid representation. Currently, no topology connections
            are considered. This function will be improved in future versions.

        """
        if not room.area or room.area <= 0:
            msg = "Pseudo-room must have a valid positive area."
            raise InvalidRoomError(msg)
        length = max(room.area**0.5, 2 * len(room.doors) + 2)
        width = room.area / length

        pseudo_doors = room.doors.copy()
        for count, door in enumerate(pseudo_doors):
            object.__setattr__(door, "start", (2 * count + 1, width))
            object.__setattr__(door, "end", (2 * count + 2, width))

        pseudo_walls = [
            Wall((0, width), (0, 0)),
            Wall((0, 0), (length, 0)),
            Wall((length, 0), (length, width)),
        ]

        doorside_walls = [
            Wall((2 * count, width), (2 * count + 1, width))
            for count in range(len(pseudo_doors))
        ]
        doorside_walls.append(Wall((len(pseudo_doors) * 2, width), (length, width)))
        pseudo_walls.extend(doorside_walls)

        return Room(
            room_id=room.room_id,
            name=room.name,
            building=room.building,
            floor=room.floor,
            walls=pseudo_walls,
            doors=pseudo_doors,
            contents=room.contents,
            rng_generator=np.random.default_rng(),
        )

    def find_room_by_location(self, location: tuple[float, float]) -> Room | None:
        """Find the room that contains the given location."""
        for room in self.rooms:
            if room.walls and room.contains_point(location):
                return room
        return None

adjacency_matrix property

Get the adjacency matrix representing room connections on the floor.

edge_set property

Get a list of all wall edges on the floor.

room_ids property

Get a list of room IDs on the floor.

room_names property

Get a list of room names on the floor.

__post_init__()

Post-initialization to validate floor attributes.

Source code in src/amr_hub_abm/space/floor.py

def __post_init__(self) -> None:
    """Post-initialization to validate floor attributes."""
    room_ids = [room.room_id for room in self.rooms]
    if len(room_ids) != len(set(room_ids)):
        msg = f"Duplicate room IDs found on floor {self.floor_number}."
        raise InvalidRoomError(msg)

add_pseudo_rooms()

Add pseudo-rooms to the floor.

Source code in src/amr_hub_abm/space/floor.py

def add_pseudo_rooms(self) -> None:
    """Add pseudo-rooms to the floor."""
    for existing_room in self.rooms:
        if not existing_room.walls:
            pseudo_room = Floor.create_spatial_room_from_pseudo_room(existing_room)
            self.pseudo_rooms.append(pseudo_room)

create_spatial_room_from_pseudo_room(room) staticmethod

Create a spatial room from a pseudo-room based on area.

Parameters:

Name	Type	Description	Default
room	Room	The pseudo-room to convert.	required

Returns:

Name	Type	Description
Room	Room	A spatial room with walls and doors based on the pseudo-room's area.

Raises:

Type	Description
InvalidRoomError	If the pseudo-room does not have a valid positive area.

Note

This method generates a rectangular room layout based on the area of the pseudo-room. The doors are positioned along one side of the rectangle. This function is not complete. Currently it only creates single simple rectangular rooms. Due to possibly complex topology, rectangular rooms may not always be a valid representation. Currently, no topology connections are considered. This function will be improved in future versions.

Source code in src/amr_hub_abm/space/floor.py

@staticmethod
def create_spatial_room_from_pseudo_room(room: Room) -> Room:
    """
    Create a spatial room from a pseudo-room based on area.

    Args:
        room (Room): The pseudo-room to convert.

    Returns:
        Room: A spatial room with walls and doors based on the pseudo-room's area.

    Raises:
        InvalidRoomError: If the pseudo-room does not have a valid positive area.

    Note:
        This method generates a rectangular room layout based on the area of the
        pseudo-room. The doors are positioned along one side of the rectangle. This
        function is not complete. Currently it only creates single simple
        rectangular rooms. Due to possibly complex topology, rectangular rooms may
        not always be a valid representation. Currently, no topology connections
        are considered. This function will be improved in future versions.

    """
    if not room.area or room.area <= 0:
        msg = "Pseudo-room must have a valid positive area."
        raise InvalidRoomError(msg)
    length = max(room.area**0.5, 2 * len(room.doors) + 2)
    width = room.area / length

    pseudo_doors = room.doors.copy()
    for count, door in enumerate(pseudo_doors):
        object.__setattr__(door, "start", (2 * count + 1, width))
        object.__setattr__(door, "end", (2 * count + 2, width))

    pseudo_walls = [
        Wall((0, width), (0, 0)),
        Wall((0, 0), (length, 0)),
        Wall((length, 0), (length, width)),
    ]

    doorside_walls = [
        Wall((2 * count, width), (2 * count + 1, width))
        for count in range(len(pseudo_doors))
    ]
    doorside_walls.append(Wall((len(pseudo_doors) * 2, width), (length, width)))
    pseudo_walls.extend(doorside_walls)

    return Room(
        room_id=room.room_id,
        name=room.name,
        building=room.building,
        floor=room.floor,
        walls=pseudo_walls,
        doors=pseudo_doors,
        contents=room.contents,
        rng_generator=np.random.default_rng(),
    )

find_room_by_location(location)

Find the room that contains the given location.

Source code in src/amr_hub_abm/space/floor.py

def find_room_by_location(self, location: tuple[float, float]) -> Room | None:
    """Find the room that contains the given location."""
    for room in self.rooms:
        if room.walls and room.contains_point(location):
            return room
    return None

plot(ax, agents=None, *, trajectory=False)

Plot the floor layout including rooms and doors.

Source code in src/amr_hub_abm/space/floor.py

def plot(
    self, ax: Axes, agents: list[Agent] | None = None, *, trajectory: bool = False
) -> None:
    """Plot the floor layout including rooms and doors."""
    for room in self.rooms:
        room.plot(ax=ax, agents=agents, trajectory=trajectory)

location

Module containing location representation for the AMR Hub ABM simulation.

Location dataclass

Representation of a location in the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/space/location.py

@dataclass
class Location:
    """Representation of a location in the AMR Hub ABM simulation."""

    x: float
    y: float
    floor: int
    building: str | None = None

    def move(self, new_x: float, new_y: float, new_floor: int) -> None:
        """Move the location to new coordinates."""
        self.x = new_x
        self.y = new_y
        self.floor = new_floor

    def distance_to(self, other: Location) -> float:
        """Calculate the Euclidean distance to another location."""
        if self.building != other.building:
            raise InvalidDistanceError((self.building, other.building), building=True)
        if self.floor != other.floor:
            raise InvalidDistanceError((self.floor, other.floor), building=False)

        return math.sqrt((self.x - other.x) ** 2 + (self.y - other.y) ** 2)

    def __repr__(self) -> str:
        """Return a string representation of the location."""
        return (
            f"Location(x={self.x:.2f}, y={self.y:.2f}, {self.floor}, {self.building})"
        )

    def which_room(self, rooms: list[Room]) -> Room | None:
        """Determine which room the location is in, if any."""
        for room in rooms:
            if room.building != self.building or room.floor != self.floor:
                continue
            if room.region.contains(shapely.geometry.Point(self.x, self.y)):
                return room
        return None

    def check_line_of_sight(self, other: Location, walls: list[Wall]) -> bool:
        """Check if there is a line of sight to another location, considering walls."""
        if self.building != other.building:
            return False

        if self.floor != other.floor:
            return False

        line_of_sight = shapely.geometry.LineString(
            [(self.x, self.y), (other.x, other.y)]
        )

        return not any(line_of_sight.crosses(wall.line) for wall in walls)

    @staticmethod
    def check_intersection_with_walls(
        loc_x: float, loc_y: float, interaction_radius: float, walls: list[Wall]
    ) -> bool:
        """Check if the agent intersects with any walls."""
        for wall in walls:
            if (
                wall.polygon.distance(shapely.geometry.Point(loc_x, loc_y))
                < interaction_radius
            ):
                return True
        return False

__repr__()

Return a string representation of the location.

Source code in src/amr_hub_abm/space/location.py

def __repr__(self) -> str:
    """Return a string representation of the location."""
    return (
        f"Location(x={self.x:.2f}, y={self.y:.2f}, {self.floor}, {self.building})"
    )

check_intersection_with_walls(loc_x, loc_y, interaction_radius, walls) staticmethod

Check if the agent intersects with any walls.

Source code in src/amr_hub_abm/space/location.py

@staticmethod
def check_intersection_with_walls(
    loc_x: float, loc_y: float, interaction_radius: float, walls: list[Wall]
) -> bool:
    """Check if the agent intersects with any walls."""
    for wall in walls:
        if (
            wall.polygon.distance(shapely.geometry.Point(loc_x, loc_y))
            < interaction_radius
        ):
            return True
    return False

check_line_of_sight(other, walls)

Check if there is a line of sight to another location, considering walls.

Source code in src/amr_hub_abm/space/location.py

def check_line_of_sight(self, other: Location, walls: list[Wall]) -> bool:
    """Check if there is a line of sight to another location, considering walls."""
    if self.building != other.building:
        return False

    if self.floor != other.floor:
        return False

    line_of_sight = shapely.geometry.LineString(
        [(self.x, self.y), (other.x, other.y)]
    )

    return not any(line_of_sight.crosses(wall.line) for wall in walls)

distance_to(other)

Calculate the Euclidean distance to another location.

Source code in src/amr_hub_abm/space/location.py

def distance_to(self, other: Location) -> float:
    """Calculate the Euclidean distance to another location."""
    if self.building != other.building:
        raise InvalidDistanceError((self.building, other.building), building=True)
    if self.floor != other.floor:
        raise InvalidDistanceError((self.floor, other.floor), building=False)

    return math.sqrt((self.x - other.x) ** 2 + (self.y - other.y) ** 2)

move(new_x, new_y, new_floor)

Move the location to new coordinates.

Source code in src/amr_hub_abm/space/location.py

def move(self, new_x: float, new_y: float, new_floor: int) -> None:
    """Move the location to new coordinates."""
    self.x = new_x
    self.y = new_y
    self.floor = new_floor

which_room(rooms)

Determine which room the location is in, if any.

Source code in src/amr_hub_abm/space/location.py

def which_room(self, rooms: list[Room]) -> Room | None:
    """Determine which room the location is in, if any."""
    for room in rooms:
        if room.building != self.building or room.floor != self.floor:
            continue
        if room.region.contains(shapely.geometry.Point(self.x, self.y)):
            return room
    return None

room

Module defining room-related classes for the AMR Hub ABM simulation.

Room dataclass

Representation of a room in the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/space/room.py

@dataclass
class Room:
    """Representation of a room in the AMR Hub ABM simulation."""

    room_id: int
    name: str
    building: str
    floor: int
    contents: list[Content]
    doors: list[Door]
    rng_generator: Generator
    walls: list[Wall] | None = field(default=None)
    area: float | None = field(default=None)
    region: shapely.geometry.Polygon = field(init=False)
    room_hash: str = field(init=False)

    def __post_init__(self) -> None:
        """Post-initialization to validate room attributes."""
        if not self.walls and not self.area:
            msg = "Either walls or area must be provided to define a room."
            raise SimulationModeError(msg)

        if self.walls and len(self.walls) < 3:
            msg = "A room must have at least 3 walls to form a closed region."
            raise InvalidRoomError(msg)

        if self.walls and self.area:
            msg = "Provide either walls or area, not both, to define a room."
            raise SimulationModeError(msg)

        if not self.area:
            self.area = self.form_region().area

        if self.area <= 0:
            msg = f"Room area must be positive. Got {self.area}."
            raise InvalidRoomError(msg)

        if self.walls:
            self.region = self.form_region()
        else:
            self.region = shapely.geometry.Polygon()
            logger.warning(
                "Room %s has no walls; region is set to an empty polygon.", self.name
            )

        self.room_hash = (
            self.create_polygon_hash() if self.walls else self.create_name_hash()
        )

        self.validate_contents()

    def __hash__(self) -> int:
        """Generate a hash for the room based on its unique hash string."""
        return hash(self.room_hash)

    def __eq__(self, other: object) -> bool:
        """Check equality of two rooms based on their unique hash strings."""
        if not isinstance(other, Room):
            return NotImplemented
        return self.room_hash == other.room_hash

    def create_polygon_hash(self) -> str:
        """Create a unique hash for the room based on its polygonal region."""
        if not self.walls:
            msg = "Cannot create polygon hash without walls."
            raise SimulationModeError(msg)

        return hashlib.sha256(shapely.ops.orient(self.region).wkb).hexdigest()

    def create_name_hash(self) -> str:
        """Create a unique hash for the room based on its name."""
        return hashlib.sha256(self.name.encode("utf-8")).hexdigest()

    def form_region(self) -> shapely.geometry.Polygon:
        """Get the polygonal region of the room based on its walls."""
        if self.walls is None:
            msg = "Cannot form region without walls."
            raise InvalidRoomError(msg)

        merged_lines = shapely.ops.linemerge(
            [wall.line for wall in self.walls] + [door.line for door in self.doors]
        )

        polygon = shapely.ops.polygonize(merged_lines)

        if len(polygon) == 0:
            msg = "The walls do not form a valid closed region."
            raise InvalidRoomError(msg)

        return polygon[0]

    def plot(
        self,
        ax: Axes,
        agents: list[Agent] | None = None,
        *,
        trajectory: bool = False,
        **kwargs: dict,
    ) -> None:
        """Plot the room on a given matplotlib axis."""
        if not self.walls:
            msg = "Cannot plot room without walls."
            raise SimulationModeError(msg)

        for wall in self.walls:
            wall.plot(ax, color="black")  # type: ignore  # noqa: PGH003

        for door in self.doors:
            x, y = door.line.xy
            ax.plot(
                x,
                y,
                color=kwargs.get("door_color", "brown"),
                linewidth=kwargs.get("door_width", 2),
            )

        for content in self.contents:
            ax.scatter(
                content.position[0],
                content.position[1],
                marker=content.marker_type,
                color=content.color,
                s=content.marker_size,
                label=f"{content.content_type.name} ({content.content_id})",
            )

        if agents is None:
            return

        for agent in agents:
            if (
                agent.location.building == self.building
                and agent.location.floor == self.floor
            ) and self.contains_point((agent.location.x, agent.location.y)):
                agent.plot_agent(ax)
                if trajectory:
                    agent.plot_trajectory(ax)

    def contains_point(self, point: tuple[float, float]) -> bool:
        """Check if a given point is inside the room."""
        if not self.walls:
            msg = "Cannot check point containment without walls."
            raise SimulationModeError(msg)

        return self.region.contains(shapely.geometry.Point(point))

    def get_random_point(self, max_attempts: int = 1000) -> tuple[float, float]:
        """Get a random point within the room."""
        if not self.walls:
            msg = "Cannot get random point without walls."
            raise SimulationModeError(msg)

        minx, miny, maxx, maxy = self.region.bounds

        for _ in range(max_attempts):
            # If required later... Improve efficiency using batching or spatial indexing
            random_point = shapely.geometry.Point(
                self.rng_generator.uniform(minx, maxx),
                self.rng_generator.uniform(miny, maxy),
            )
            if self.region.contains(
                random_point
            ) and not Location.check_intersection_with_walls(
                random_point.x, random_point.y, 0.1, self.walls
            ):
                return (random_point.x, random_point.y)

        msg = f"""
        Failed to find a random point within the room after {max_attempts} attempts.
        Consider increasing max_attempts or checking room geometry.
        """
        raise SimulationModeError(msg)

    def get_door_access_point(self) -> tuple[Door, tuple[float, float]]:
        """Get a point near one of the room's doors for access."""
        if not self.doors:
            msg = f"Room {self.name} has no doors for access."
            raise InvalidRoomError(msg)

        if len(self.doors) > 1:
            msg = f"Room {self.name} has multiple doors; \
            This functionality is not supported for now."
            raise InvalidRoomError(msg)

        door = self.doors[0]
        midpoint = door.line.interpolate(0.5, normalized=True)
        return (door, (midpoint.x, midpoint.y))

    def validate_contents(self) -> None:
        """Validate that all contents are located within the room."""
        if not self.walls:
            return

        for content in self.contents:
            if not self.contains_point(content.position):
                msg = f"Content {content.content_id} of type {content.content_type} "
                msg += f"is located at {content.position}, which is outside the room."
                raise InvalidRoomError(msg)

__eq__(other)

Check equality of two rooms based on their unique hash strings.

Source code in src/amr_hub_abm/space/room.py

def __eq__(self, other: object) -> bool:
    """Check equality of two rooms based on their unique hash strings."""
    if not isinstance(other, Room):
        return NotImplemented
    return self.room_hash == other.room_hash

__hash__()

Generate a hash for the room based on its unique hash string.

Source code in src/amr_hub_abm/space/room.py

def __hash__(self) -> int:
    """Generate a hash for the room based on its unique hash string."""
    return hash(self.room_hash)

__post_init__()

Post-initialization to validate room attributes.

Source code in src/amr_hub_abm/space/room.py

def __post_init__(self) -> None:
    """Post-initialization to validate room attributes."""
    if not self.walls and not self.area:
        msg = "Either walls or area must be provided to define a room."
        raise SimulationModeError(msg)

    if self.walls and len(self.walls) < 3:
        msg = "A room must have at least 3 walls to form a closed region."
        raise InvalidRoomError(msg)

    if self.walls and self.area:
        msg = "Provide either walls or area, not both, to define a room."
        raise SimulationModeError(msg)

    if not self.area:
        self.area = self.form_region().area

    if self.area <= 0:
        msg = f"Room area must be positive. Got {self.area}."
        raise InvalidRoomError(msg)

    if self.walls:
        self.region = self.form_region()
    else:
        self.region = shapely.geometry.Polygon()
        logger.warning(
            "Room %s has no walls; region is set to an empty polygon.", self.name
        )

    self.room_hash = (
        self.create_polygon_hash() if self.walls else self.create_name_hash()
    )

    self.validate_contents()

contains_point(point)

Check if a given point is inside the room.

Source code in src/amr_hub_abm/space/room.py

def contains_point(self, point: tuple[float, float]) -> bool:
    """Check if a given point is inside the room."""
    if not self.walls:
        msg = "Cannot check point containment without walls."
        raise SimulationModeError(msg)

    return self.region.contains(shapely.geometry.Point(point))

create_name_hash()

Create a unique hash for the room based on its name.

Source code in src/amr_hub_abm/space/room.py

def create_name_hash(self) -> str:
    """Create a unique hash for the room based on its name."""
    return hashlib.sha256(self.name.encode("utf-8")).hexdigest()

create_polygon_hash()

Create a unique hash for the room based on its polygonal region.

Source code in src/amr_hub_abm/space/room.py

def create_polygon_hash(self) -> str:
    """Create a unique hash for the room based on its polygonal region."""
    if not self.walls:
        msg = "Cannot create polygon hash without walls."
        raise SimulationModeError(msg)

    return hashlib.sha256(shapely.ops.orient(self.region).wkb).hexdigest()

form_region()

Get the polygonal region of the room based on its walls.

Source code in src/amr_hub_abm/space/room.py

def form_region(self) -> shapely.geometry.Polygon:
    """Get the polygonal region of the room based on its walls."""
    if self.walls is None:
        msg = "Cannot form region without walls."
        raise InvalidRoomError(msg)

    merged_lines = shapely.ops.linemerge(
        [wall.line for wall in self.walls] + [door.line for door in self.doors]
    )

    polygon = shapely.ops.polygonize(merged_lines)

    if len(polygon) == 0:
        msg = "The walls do not form a valid closed region."
        raise InvalidRoomError(msg)

    return polygon[0]

get_door_access_point()

Get a point near one of the room's doors for access.

Source code in src/amr_hub_abm/space/room.py

def get_door_access_point(self) -> tuple[Door, tuple[float, float]]:
    """Get a point near one of the room's doors for access."""
    if not self.doors:
        msg = f"Room {self.name} has no doors for access."
        raise InvalidRoomError(msg)

    if len(self.doors) > 1:
        msg = f"Room {self.name} has multiple doors; \
        This functionality is not supported for now."
        raise InvalidRoomError(msg)

    door = self.doors[0]
    midpoint = door.line.interpolate(0.5, normalized=True)
    return (door, (midpoint.x, midpoint.y))

get_random_point(max_attempts=1000)

Get a random point within the room.

Source code in src/amr_hub_abm/space/room.py

def get_random_point(self, max_attempts: int = 1000) -> tuple[float, float]:
    """Get a random point within the room."""
    if not self.walls:
        msg = "Cannot get random point without walls."
        raise SimulationModeError(msg)

    minx, miny, maxx, maxy = self.region.bounds

    for _ in range(max_attempts):
        # If required later... Improve efficiency using batching or spatial indexing
        random_point = shapely.geometry.Point(
            self.rng_generator.uniform(minx, maxx),
            self.rng_generator.uniform(miny, maxy),
        )
        if self.region.contains(
            random_point
        ) and not Location.check_intersection_with_walls(
            random_point.x, random_point.y, 0.1, self.walls
        ):
            return (random_point.x, random_point.y)

    msg = f"""
    Failed to find a random point within the room after {max_attempts} attempts.
    Consider increasing max_attempts or checking room geometry.
    """
    raise SimulationModeError(msg)

plot(ax, agents=None, *, trajectory=False, **kwargs)

Plot the room on a given matplotlib axis.

Source code in src/amr_hub_abm/space/room.py

def plot(
    self,
    ax: Axes,
    agents: list[Agent] | None = None,
    *,
    trajectory: bool = False,
    **kwargs: dict,
) -> None:
    """Plot the room on a given matplotlib axis."""
    if not self.walls:
        msg = "Cannot plot room without walls."
        raise SimulationModeError(msg)

    for wall in self.walls:
        wall.plot(ax, color="black")  # type: ignore  # noqa: PGH003

    for door in self.doors:
        x, y = door.line.xy
        ax.plot(
            x,
            y,
            color=kwargs.get("door_color", "brown"),
            linewidth=kwargs.get("door_width", 2),
        )

    for content in self.contents:
        ax.scatter(
            content.position[0],
            content.position[1],
            marker=content.marker_type,
            color=content.color,
            s=content.marker_size,
            label=f"{content.content_type.name} ({content.content_id})",
        )

    if agents is None:
        return

    for agent in agents:
        if (
            agent.location.building == self.building
            and agent.location.floor == self.floor
        ) and self.contains_point((agent.location.x, agent.location.y)):
            agent.plot_agent(ax)
            if trajectory:
                agent.plot_trajectory(ax)

validate_contents()

Validate that all contents are located within the room.

Source code in src/amr_hub_abm/space/room.py

def validate_contents(self) -> None:
    """Validate that all contents are located within the room."""
    if not self.walls:
        return

    for content in self.contents:
        if not self.contains_point(content.position):
            msg = f"Content {content.content_id} of type {content.content_type} "
            msg += f"is located at {content.position}, which is outside the room."
            raise InvalidRoomError(msg)

wall

Module defining wall representation for the AMR Hub ABM simulation.

Wall dataclass

Representation of a wall in the AMR Hub ABM simulation.

Source code in src/amr_hub_abm/space/wall.py

@dataclass
class Wall:
    """Representation of a wall in the AMR Hub ABM simulation."""

    start: tuple[float, float]
    end: tuple[float, float]
    thickness: float = 0.2

    @property
    def line(self) -> shapely.geometry.LineString:
        """Get the line representation of the wall."""
        return shapely.geometry.LineString([self.start, self.end])

    @property
    def polygon(self) -> shapely.geometry.Polygon:
        """Get the polygon representation of the wall based on its thickness."""
        line = self.line
        return line.buffer(self.thickness / 2, cap_style="square")

    def plot(self, ax: Axes, **kwargs: dict) -> None:
        """Plot the wall on a given matplotlib axis."""
        x, y = self.polygon.exterior.xy
        ax.fill(x, y, **kwargs)  # pyright: ignore[reportArgumentType]

line property

Get the line representation of the wall.

polygon property

Get the polygon representation of the wall based on its thickness.

plot(ax, **kwargs)

Plot the wall on a given matplotlib axis.

Source code in src/amr_hub_abm/space/wall.py

def plot(self, ax: Axes, **kwargs: dict) -> None:
    """Plot the wall on a given matplotlib axis."""
    x, y = self.polygon.exterior.xy
    ax.fill(x, y, **kwargs)  # pyright: ignore[reportArgumentType]

task

Module for AMR Hub ABM tasks.

Task dataclass

Representation of a task assigned to an agent.

Source code in src/amr_hub_abm/task.py

@dataclass
class Task:
    """Representation of a task assigned to an agent."""

    task_type: ClassVar[TaskType] = TaskType.GENERIC

    time_needed: int
    time_due: int

    progress: TaskProgress = field(default=TaskProgress.NOT_STARTED, kw_only=True)
    priority: TaskPriority = field(default=TaskPriority.MEDIUM, kw_only=True)

    location: Location = field(init=False)

    time_started: int = field(init=False)
    time_completed: int = field(init=False)

    def time_spent(self, current_time: int) -> int:
        """Calculate the time spent on the task so far."""
        if self.progress == TaskProgress.COMPLETED:
            return self.time_completed - self.time_started

        if self.progress == TaskProgress.IN_PROGRESS:
            return current_time - self.time_started

        return 0

    def __post_init__(self) -> None:
        """Post-initialization to validate task attributes."""
        if self.time_needed < 0:
            msg = "Time needed for a task cannot be negative."
            raise TimeError(msg)

        if self.time_due < 0:
            msg = "Time due for a task cannot be negative."
            raise TimeError(msg)

    def update_progress(self, current_time: int, agent: Agent) -> None:
        """Update the progress of the task based on time spent."""
        if self.progress == TaskProgress.COMPLETED:
            return

        time_spent = self.time_spent(current_time=current_time)

        if time_spent >= self.time_needed:
            self.progress = TaskProgress.COMPLETED
            logger.info(
                "Task %s completed for Agent id %s at time %d.",
                self.task_type.name,
                agent.idx,
                current_time,
            )
            self.time_completed = current_time
            if isinstance(self, TaskOccupyContent):
                add_agent_occupancy(agent, self.content, current_time=current_time)
            return

        if self.progress == TaskProgress.IN_PROGRESS:
            logger.info(
                "Agent id %s performing task %s at location %s.",
                agent.idx,
                self.task_type,
                self.location,
            )
            return

        if not agent.check_if_location_reached(self.location):
            self.progress = TaskProgress.MOVING_TO_LOCATION
            if isinstance(self, TaskDoorAccess):
                self.modify_location(agent)
            remove_agent_occupancy(agent, current_time=current_time)
            logger.info(
                "Agent id %s moving to task location %s.", agent.idx, self.location
            )
            agent.head_to_point((self.location.x, self.location.y))
            agent.move_one_step()
            return
        self.progress = TaskProgress.IN_PROGRESS
        self.time_started = current_time

        if self.progress == TaskProgress.MOVING_TO_LOCATION:
            self.progress = TaskProgress.IN_PROGRESS
            self.time_started = current_time

    def __repr__(self) -> str:
        """Representation of the task."""
        return (
            f"Task(type={self.task_type}, priority={self.priority}, "
            f"progress={self.progress}, time_needed={self.time_needed}, "
            f"time_due={self.time_due})"
        )

__post_init__()

Post-initialization to validate task attributes.

Source code in src/amr_hub_abm/task.py

def __post_init__(self) -> None:
    """Post-initialization to validate task attributes."""
    if self.time_needed < 0:
        msg = "Time needed for a task cannot be negative."
        raise TimeError(msg)

    if self.time_due < 0:
        msg = "Time due for a task cannot be negative."
        raise TimeError(msg)

__repr__()

Representation of the task.

Source code in src/amr_hub_abm/task.py

def __repr__(self) -> str:
    """Representation of the task."""
    return (
        f"Task(type={self.task_type}, priority={self.priority}, "
        f"progress={self.progress}, time_needed={self.time_needed}, "
        f"time_due={self.time_due})"
    )

time_spent(current_time)

Calculate the time spent on the task so far.

Source code in src/amr_hub_abm/task.py

def time_spent(self, current_time: int) -> int:
    """Calculate the time spent on the task so far."""
    if self.progress == TaskProgress.COMPLETED:
        return self.time_completed - self.time_started

    if self.progress == TaskProgress.IN_PROGRESS:
        return current_time - self.time_started

    return 0

update_progress(current_time, agent)

Update the progress of the task based on time spent.

Source code in src/amr_hub_abm/task.py

def update_progress(self, current_time: int, agent: Agent) -> None:
    """Update the progress of the task based on time spent."""
    if self.progress == TaskProgress.COMPLETED:
        return

    time_spent = self.time_spent(current_time=current_time)

    if time_spent >= self.time_needed:
        self.progress = TaskProgress.COMPLETED
        logger.info(
            "Task %s completed for Agent id %s at time %d.",
            self.task_type.name,
            agent.idx,
            current_time,
        )
        self.time_completed = current_time
        if isinstance(self, TaskOccupyContent):
            add_agent_occupancy(agent, self.content, current_time=current_time)
        return

    if self.progress == TaskProgress.IN_PROGRESS:
        logger.info(
            "Agent id %s performing task %s at location %s.",
            agent.idx,
            self.task_type,
            self.location,
        )
        return

    if not agent.check_if_location_reached(self.location):
        self.progress = TaskProgress.MOVING_TO_LOCATION
        if isinstance(self, TaskDoorAccess):
            self.modify_location(agent)
        remove_agent_occupancy(agent, current_time=current_time)
        logger.info(
            "Agent id %s moving to task location %s.", agent.idx, self.location
        )
        agent.head_to_point((self.location.x, self.location.y))
        agent.move_one_step()
        return
    self.progress = TaskProgress.IN_PROGRESS
    self.time_started = current_time

    if self.progress == TaskProgress.MOVING_TO_LOCATION:
        self.progress = TaskProgress.IN_PROGRESS
        self.time_started = current_time

TaskAttendPatient dataclass

Bases: Task

Representation of an 'attend patient' task.

Source code in src/amr_hub_abm/task.py

@dataclass
class TaskAttendPatient(Task):
    """Representation of an 'attend patient' task."""

    task_type: ClassVar[TaskType] = TaskType.ATTEND_PATIENT
    patient: Agent

    def __post_init__(self) -> None:
        """Post-initialization to set the task location."""
        super().__post_init__()
        self.location = self.patient.location

__post_init__()

Post-initialization to set the task location.

Source code in src/amr_hub_abm/task.py

def __post_init__(self) -> None:
    """Post-initialization to set the task location."""
    super().__post_init__()
    self.location = self.patient.location

TaskDoorAccess dataclass

Bases: Task

Representation of a 'door access' task.

Source code in src/amr_hub_abm/task.py

@dataclass
class TaskDoorAccess(Task):
    """Representation of a 'door access' task."""

    task_type: ClassVar[TaskType] = TaskType.DOOR_ACCESS
    door: Door
    building: str
    floor: int
    destination_room: int
    buffer_distance: float = 0.05

    def __post_init__(self) -> None:
        """Post-initialization to set the task location."""
        super().__post_init__()

        if self.door.start is None or self.door.end is None:
            msg = "Door must have defined start and end points to set task location."
            raise SimulationModeError(msg)

        self.location = Location(
            building=self.building,
            floor=self.floor,
            x=(self.door.start[0] + self.door.end[0]) / 2,
            y=(self.door.start[1] + self.door.end[1]) / 2,
        )

    def modify_location(self, agent: Agent) -> None:
        """Modify the location of the task to account for buffer."""
        if self.door.start is None:
            msg = "Door coords needed"
            raise SimulationModeError(msg)

        if self.door.end is None:
            msg = "Door coords needed"
            raise SimulationModeError(msg)

        proposed_location1 = Location(
            building=self.building,
            floor=self.floor,
            x=(self.door.start[0] + self.door.end[0]) / 2,
            y=(self.door.start[1] + self.door.end[1]) / 2 + self.buffer_distance,
        )

        proposed_location2 = Location(
            building=self.building,
            floor=self.floor,
            x=(self.door.start[0] + self.door.end[0]) / 2,
            y=(self.door.start[1] + self.door.end[1]) / 2 - self.buffer_distance,
        )

        proposed_location1_room = agent.get_room(
            (proposed_location1.x, proposed_location1.y)
        )

        if proposed_location1_room is None:
            msg = "Proposed location 1 does not correspond to a valid room."
            raise SimulationModeError(msg)

        if proposed_location1_room.room_id == self.destination_room:
            self.location = proposed_location1
        else:
            self.location = proposed_location2

__post_init__()

Post-initialization to set the task location.

Source code in src/amr_hub_abm/task.py

def __post_init__(self) -> None:
    """Post-initialization to set the task location."""
    super().__post_init__()

    if self.door.start is None or self.door.end is None:
        msg = "Door must have defined start and end points to set task location."
        raise SimulationModeError(msg)

    self.location = Location(
        building=self.building,
        floor=self.floor,
        x=(self.door.start[0] + self.door.end[0]) / 2,
        y=(self.door.start[1] + self.door.end[1]) / 2,
    )

modify_location(agent)

Modify the location of the task to account for buffer.

Source code in src/amr_hub_abm/task.py

def modify_location(self, agent: Agent) -> None:
    """Modify the location of the task to account for buffer."""
    if self.door.start is None:
        msg = "Door coords needed"
        raise SimulationModeError(msg)

    if self.door.end is None:
        msg = "Door coords needed"
        raise SimulationModeError(msg)

    proposed_location1 = Location(
        building=self.building,
        floor=self.floor,
        x=(self.door.start[0] + self.door.end[0]) / 2,
        y=(self.door.start[1] + self.door.end[1]) / 2 + self.buffer_distance,
    )

    proposed_location2 = Location(
        building=self.building,
        floor=self.floor,
        x=(self.door.start[0] + self.door.end[0]) / 2,
        y=(self.door.start[1] + self.door.end[1]) / 2 - self.buffer_distance,
    )

    proposed_location1_room = agent.get_room(
        (proposed_location1.x, proposed_location1.y)
    )

    if proposed_location1_room is None:
        msg = "Proposed location 1 does not correspond to a valid room."
        raise SimulationModeError(msg)

    if proposed_location1_room.room_id == self.destination_room:
        self.location = proposed_location1
    else:
        self.location = proposed_location2

TaskGotoLocation dataclass

Bases: Task

Representation of a 'goto location' task.

Source code in src/amr_hub_abm/task.py

@dataclass
class TaskGotoLocation(Task):
    """Representation of a 'goto location' task."""

    task_type: ClassVar[TaskType] = TaskType.GOTO_LOCATION
    destination_location: Location

    def __post_init__(self) -> None:
        """Post-initialization to set the task location."""
        super().__post_init__()
        self.location = self.destination_location

__post_init__()

Post-initialization to set the task location.

Source code in src/amr_hub_abm/task.py

def __post_init__(self) -> None:
    """Post-initialization to set the task location."""
    super().__post_init__()
    self.location = self.destination_location

TaskOccupyContent dataclass

Bases: Task

Representation of an 'occupy content' task.

Source code in src/amr_hub_abm/task.py

@dataclass
class TaskOccupyContent(Task):
    """Representation of an 'occupy content' task."""

    task_type: ClassVar[TaskType] = TaskType.OCCUPY_CONTENT
    content_type: int
    room: Room
    content: Content = field(init=False)

    def __post_init__(self) -> None:
        """Post-initialization to set the task location."""
        super().__post_init__()

    def assign_content(self) -> None:
        """Assign the content to be occupied based on the content type and room."""
        content = next(
            (c for c in self.room.contents if c.content_type == self.content_type), None
        )

        if content is None:
            msg = (
                f"No content of type {self.content_type} found in {self.room.name} "
                f"for 'occupy_content' task."
            )
            raise SimulationModeError(msg)

        self.content = content

        self.location = Location(
            building=self.content.location.building,
            floor=self.content.location.floor,
            x=self.content.location.x,
            y=self.content.location.y,
        )

__post_init__()

Post-initialization to set the task location.

Source code in src/amr_hub_abm/task.py

def __post_init__(self) -> None:
    """Post-initialization to set the task location."""
    super().__post_init__()

assign_content()

Assign the content to be occupied based on the content type and room.

Source code in src/amr_hub_abm/task.py

def assign_content(self) -> None:
    """Assign the content to be occupied based on the content type and room."""
    content = next(
        (c for c in self.room.contents if c.content_type == self.content_type), None
    )

    if content is None:
        msg = (
            f"No content of type {self.content_type} found in {self.room.name} "
            f"for 'occupy_content' task."
        )
        raise SimulationModeError(msg)

    self.content = content

    self.location = Location(
        building=self.content.location.building,
        floor=self.content.location.floor,
        x=self.content.location.x,
        y=self.content.location.y,
    )

TaskPriority

Bases: IntEnum

Enumeration of possible task priority levels.

Source code in src/amr_hub_abm/task.py

class TaskPriority(IntEnum):
    """Enumeration of possible task priority levels."""

    LOW = 1
    MEDIUM = 2
    HIGH = 3

TaskProgress

Bases: IntEnum

Enumeration of possible task progress states.

Source code in src/amr_hub_abm/task.py

class TaskProgress(IntEnum):
    """Enumeration of possible task progress states."""

    NOT_STARTED = 0
    MOVING_TO_LOCATION = 1
    SUSPENDED = 2
    IN_PROGRESS = 3
    COMPLETED = 4

TaskType

Bases: IntEnum

Enumeration of possible task types.

Source code in src/amr_hub_abm/task.py

class TaskType(IntEnum):
    """Enumeration of possible task types."""

    GENERIC = 0
    OFFICE_WORK = 1
    NURSE_ROUND = 2
    ATTEND_PATIENT = 3
    GOTO_LOCATION = 4
    GOTO_AGENT = 5
    ATTEND_BELL = 6
    STAY_IN_BED = 7
    STAY_IN_ROOM = 8
    INTERACT_WITH_AGENT = 9
    DOOR_ACCESS = 10
    WORKSTATION = 11
    OCCUPY_CONTENT = 12

TaskWorkstation dataclass

Bases: Task

Representation of a 'workstation' task.

Source code in src/amr_hub_abm/task.py

@dataclass
class TaskWorkstation(Task):
    """Representation of a 'workstation' task."""

    task_type: ClassVar[TaskType] = TaskType.WORKSTATION
    workstation_location: Location

    def __post_init__(self) -> None:
        """Post-initialization to set the task location."""
        super().__post_init__()
        self.location = self.workstation_location

__post_init__()

Post-initialization to set the task location.

Source code in src/amr_hub_abm/task.py

def __post_init__(self) -> None:
    """Post-initialization to set the task location."""
    super().__post_init__()
    self.location = self.workstation_location

add_agent_occupancy(agent, content, current_time)

Add the agent's occupancy to the specified content.

Source code in src/amr_hub_abm/task.py

def add_agent_occupancy(agent: Agent, content: Content, current_time: int) -> None:
    """Add the agent's occupancy to the specified content."""
    content.occupier_id = (agent.idx, agent.agent_type)
    agent.stationary = True

    room = agent.get_room()
    room_name = "unknown" if room is None else room.name

    logger.info(
        """
        Agent id %s added occupancy to content id %s of type %s
        in room %s at time %d.
        """,
        agent.idx,
        content.content_id,
        content.content_type,
        room_name,
        current_time,
    )

remove_agent_occupancy(agent, current_time)

Remove the agent's occupancy from any content they are currently occupying.

Source code in src/amr_hub_abm/task.py

def remove_agent_occupancy(agent: Agent, current_time: int) -> None:
    """Remove the agent's occupancy from any content they are currently occupying."""
    room = agent.get_room()
    if room is None:
        return
    for content in room.contents:
        if content.occupier_id == (agent.idx, agent.agent_type):
            content.occupier_id = None
            agent.stationary = False
            logger.info(
                """
                Agent id %s removed occupancy from content id %s of type %s
                in room %s at time %d.
                """,
                agent.idx,
                content.content_id,
                content.content_type,
                room.name,
                current_time,
            )
            return