Task Dataclass

Task module for the AMR Hub ABM framework.

task

Module for AMR Hub ABM tasks.

Design notes

• Task is a plain dataclass with an explicit location field. Subclasses exist only where there is genuine behaviour (door buffering, content lookup), not just to set a location.
• Subclass-specific behaviour is expressed through lifecycle hooks (on_start_moving, on_started, on_completed) instead of isinstance checks in the base class.
• update_progress is a small explicit state machine: one handler per TaskProgress state.
• create_task is a factory that owns per-type defaults (durations etc.), so Agent.add_task no longer needs a long if/elif chain.
• The spatial engine (SpatialQuery) is threaded through as an explicit engine parameter rather than being accessed via the agent. This keeps the task layer decoupled from the agent's attributes and allows swapping CPU/GPU backends transparently.

Task dataclass

Representation of a task assigned to an agent.

A task is a small state machine driven by update_progress. Subclasses customise behaviour via the lifecycle hooks prepare, on_start_moving, on_started and on_completed rather than by being special-cased in the base class.

Parameters:

Name	Type	Description	Default
time_needed	int	The time required to complete the task.	required
time_due	int	The time by which the task should be completed.	required
location	Location | None	Where the task is performed. Simple task types (goto, workstation, attend patient) pass this directly; subclasses that derive their own location (door access, occupy content) may leave it None and set it in __post_init__ / prepare.	None
progress	TaskProgress	Current progress. Defaults to NOT_STARTED.	NOT_STARTED
priority	TaskPriority	Priority level. Defaults to MEDIUM.	MEDIUM

Source code in src/amr_hub_abm/task/task.py

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

    A task is a small state machine driven by ``update_progress``. Subclasses
    customise behaviour via the lifecycle hooks ``prepare``,
    ``on_start_moving``, ``on_started`` and ``on_completed`` rather than by
    being special-cased in the base class.

    Parameters
    ----------
    time_needed : int
        The time required to complete the task.
    time_due : int
        The time by which the task should be completed.
    location : Location | None
        Where the task is performed. Simple task types (goto, workstation,
        attend patient) pass this directly; subclasses that derive their own
        location (door access, occupy content) may leave it None and set it
        in ``__post_init__`` / ``prepare``.
    progress : TaskProgress, optional
        Current progress. Defaults to NOT_STARTED.
    priority : TaskPriority, optional
        Priority level. Defaults to MEDIUM.

    """

    time_needed: int
    time_due: int
    location: Location | None = None

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

    time_started: int | None = field(init=False, default=None)
    time_completed: int | None = field(init=False, default=None)

    def __post_init__(self) -> None:
        """Validate timing 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)

    # -- lifecycle hooks (no-ops on the base class) -------------------------

    def prepare(self, agent: Agent) -> bool:  # noqa: ARG002
        """
        Resolve any late-bound state before the task is first scheduled.

        Called while the task is NOT_STARTED (e.g. ``TaskOccupyContent``
        resolves currently available content here).

        Returns
        -------
        bool
            Whether the task is ready for scheduling.

        """
        return True

    def on_dispatch(self, agent: Agent) -> bool:  # noqa: ARG002
        """
        Claim resources immediately before the task begins moving.

        Returns
        -------
        bool
            Whether the task acquired any resources needed to start.

        """
        return True

    def on_start_moving(self, agent: Agent, engine: SpatialQuery) -> None:
        """Handle the transition to MOVING_TO_LOCATION."""

    def on_started(self, agent: Agent, current_time: int) -> None:
        """Handle the transition to IN_PROGRESS."""

    def on_completed(
        self, agent: Agent, current_time: int, engine: SpatialQuery
    ) -> None:
        """Handle the transition to COMPLETED."""

    # -- timing --------------------------------------------------------------

    def time_spent(self, current_time: int) -> int:
        """
        Calculate the time spent on the task so far.

        Raises
        ------
        TimeError
            If the task is COMPLETED or IN_PROGRESS but the corresponding
            timestamps are missing.

        """
        if self.progress == TaskProgress.COMPLETED:
            if self.time_started is None:
                msg = "Task marked as completed but start time is None."
                raise TimeError(msg)
            if self.time_completed is None:
                msg = "Task marked as completed but completion time is None."
                raise TimeError(msg)
            return self.time_completed - self.time_started

        if self.progress == TaskProgress.IN_PROGRESS:
            if self.time_started is None:
                msg = "Task marked as in progress but start time is None."
                raise TimeError(msg)
            return current_time - self.time_started

        return 0

    # -- state machine ---------------------------------------------------------

    def update_progress(
        self, current_time: int, agent: Agent, engine: SpatialQuery
    ) -> None:
        """
        Advance the task state machine by one tick.

        Dispatches to a handler per ``TaskProgress`` state. Completion is
        checked first since it can occur from any active state.
        """
        if self.progress == TaskProgress.COMPLETED:
            return

        if self.time_spent(current_time) >= self.time_needed:
            self._complete(current_time, agent, engine)
            return

        handler = {
            TaskProgress.IN_PROGRESS: self._tick_in_progress,
            TaskProgress.MOVING_TO_LOCATION: self._tick_moving,
            TaskProgress.NOT_STARTED: self._tick_not_started,
            TaskProgress.SUSPENDED: self._tick_not_started,
        }[self.progress]
        handler(current_time, agent, engine)

    def _complete(self, current_time: int, agent: Agent, engine: SpatialQuery) -> None:
        self.progress = TaskProgress.COMPLETED
        self.time_completed = current_time
        logger.info(
            "Task %s completed for Agent %s at t=%d.",
            self.task_type.name,
            agent.idx,
            current_time,
        )
        self.on_completed(agent, current_time, engine)

    def _tick_in_progress(
        self,
        current_time: int,  # noqa: ARG002
        agent: Agent,
        engine: SpatialQuery,  # noqa: ARG002
    ) -> None:
        logger.info(
            "Agent %s performing task %s at location %s.",
            agent.idx,
            self.task_type.name,
            self.location,
        )

    def _tick_moving(
        self, current_time: int, agent: Agent, engine: SpatialQuery
    ) -> None:
        if self.location is None:
            msg = f"Task {self.task_type.name} has no location to move to."
            raise SimulationModeError(msg)

        if engine.is_target_reached(
            agent.location, self.location, agent.interaction_radius
        ):
            self._start(current_time, agent)
            return

        engine.head_to_point(agent, (self.location.x, self.location.y))
        engine.move_one_step(agent)

    def _tick_not_started(
        self, current_time: int, agent: Agent, engine: SpatialQuery
    ) -> None:
        if self.location is None:
            msg = f"Task {self.task_type.name} has no location set."
            raise SimulationModeError(msg)

        if engine.is_target_reached(
            agent.location, self.location, agent.interaction_radius
        ):
            self._start(current_time, agent)
            return

        self.progress = TaskProgress.MOVING_TO_LOCATION
        self.on_start_moving(agent, engine)
        remove_agent_occupancy(agent, current_time=current_time, engine=engine)
        logger.info("Agent %s moving to task location %s.", agent.idx, self.location)
        engine.head_to_point(agent, (self.location.x, self.location.y))
        engine.move_one_step(agent)

    def _start(self, current_time: int, agent: Agent) -> None:
        self.progress = TaskProgress.IN_PROGRESS
        self.time_started = current_time
        self.on_started(agent, current_time)

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

__post_init__()

Validate timing attributes.

Source code in src/amr_hub_abm/task/task.py

def __post_init__(self) -> None:
    """Validate timing 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/task.py

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

on_completed(agent, current_time, engine)

Handle the transition to COMPLETED.

Source code in src/amr_hub_abm/task/task.py

def on_completed(
    self, agent: Agent, current_time: int, engine: SpatialQuery
) -> None:
    """Handle the transition to COMPLETED."""

on_dispatch(agent)

Claim resources immediately before the task begins moving.

Returns:

Type	Description
bool	Whether the task acquired any resources needed to start.

Source code in src/amr_hub_abm/task/task.py

def on_dispatch(self, agent: Agent) -> bool:  # noqa: ARG002
    """
    Claim resources immediately before the task begins moving.

    Returns
    -------
    bool
        Whether the task acquired any resources needed to start.

    """
    return True

on_start_moving(agent, engine)

Handle the transition to MOVING_TO_LOCATION.

Source code in src/amr_hub_abm/task/task.py

def on_start_moving(self, agent: Agent, engine: SpatialQuery) -> None:
    """Handle the transition to MOVING_TO_LOCATION."""

on_started(agent, current_time)

Handle the transition to IN_PROGRESS.

Source code in src/amr_hub_abm/task/task.py

def on_started(self, agent: Agent, current_time: int) -> None:
    """Handle the transition to IN_PROGRESS."""

prepare(agent)

Resolve any late-bound state before the task is first scheduled.

Called while the task is NOT_STARTED (e.g. TaskOccupyContent resolves currently available content here).

Returns:

Type	Description
bool	Whether the task is ready for scheduling.

Source code in src/amr_hub_abm/task/task.py

def prepare(self, agent: Agent) -> bool:  # noqa: ARG002
    """
    Resolve any late-bound state before the task is first scheduled.

    Called while the task is NOT_STARTED (e.g. ``TaskOccupyContent``
    resolves currently available content here).

    Returns
    -------
    bool
        Whether the task is ready for scheduling.

    """
    return True

time_spent(current_time)

Calculate the time spent on the task so far.

Raises:

Type	Description
TimeError	If the task is COMPLETED or IN_PROGRESS but the corresponding timestamps are missing.

Source code in src/amr_hub_abm/task/task.py

def time_spent(self, current_time: int) -> int:
    """
    Calculate the time spent on the task so far.

    Raises
    ------
    TimeError
        If the task is COMPLETED or IN_PROGRESS but the corresponding
        timestamps are missing.

    """
    if self.progress == TaskProgress.COMPLETED:
        if self.time_started is None:
            msg = "Task marked as completed but start time is None."
            raise TimeError(msg)
        if self.time_completed is None:
            msg = "Task marked as completed but completion time is None."
            raise TimeError(msg)
        return self.time_completed - self.time_started

    if self.progress == TaskProgress.IN_PROGRESS:
        if self.time_started is None:
            msg = "Task marked as in progress but start time is None."
            raise TimeError(msg)
        return current_time - self.time_started

    return 0

update_progress(current_time, agent, engine)

Advance the task state machine by one tick.

Dispatches to a handler per TaskProgress state. Completion is checked first since it can occur from any active state.

Source code in src/amr_hub_abm/task/task.py

def update_progress(
    self, current_time: int, agent: Agent, engine: SpatialQuery
) -> None:
    """
    Advance the task state machine by one tick.

    Dispatches to a handler per ``TaskProgress`` state. Completion is
    checked first since it can occur from any active state.
    """
    if self.progress == TaskProgress.COMPLETED:
        return

    if self.time_spent(current_time) >= self.time_needed:
        self._complete(current_time, agent, engine)
        return

    handler = {
        TaskProgress.IN_PROGRESS: self._tick_in_progress,
        TaskProgress.MOVING_TO_LOCATION: self._tick_moving,
        TaskProgress.NOT_STARTED: self._tick_not_started,
        TaskProgress.SUSPENDED: self._tick_not_started,
    }[self.progress]
    handler(current_time, agent, engine)

TaskAttendPatient dataclass

Bases: Task

An 'attend patient' task. Carries the patient so the UI/agent can refer to it.

Source code in src/amr_hub_abm/task/task.py

@dataclass
class TaskAttendPatient(Task):
    """An 'attend patient' task. Carries the patient so the UI/agent can refer to it."""

    task_type: TaskType = field(default=TaskType.ATTEND_PATIENT, kw_only=True)
    patient: Agent | None = None

    def __post_init__(self) -> None:
        """Validate patient and set location."""
        super().__post_init__()
        if self.patient is None:
            msg = "TaskAttendPatient requires a patient."
            raise SimulationModeError(msg)
        self.location = self.patient.location

__post_init__()

Validate patient and set location.

Source code in src/amr_hub_abm/task/task.py

def __post_init__(self) -> None:
    """Validate patient and set location."""
    super().__post_init__()
    if self.patient is None:
        msg = "TaskAttendPatient requires a patient."
        raise SimulationModeError(msg)
    self.location = self.patient.location

TaskDoorAccess dataclass

Bases: Task

A 'door access' task: position the agent just beyond the door midpoint.

Source code in src/amr_hub_abm/task/task.py

@dataclass
class TaskDoorAccess(Task):
    """A 'door access' task: position the agent just beyond the door midpoint."""

    task_type: TaskType = field(default=TaskType.DOOR_ACCESS, kw_only=True)
    door: Door | None = None
    building: str = ""
    floor: int = 0
    destination_room: int = -1
    buffer_distance: float = 0.05

    def __post_init__(self) -> None:
        """Validate door and compute midpoint location."""
        super().__post_init__()
        if self.door is None:
            msg = "TaskDoorAccess requires a door."
            raise SimulationModeError(msg)
        self.location = self._midpoint_location()

    def _midpoint(self) -> tuple[float, float]:
        assert self.door is not None  # noqa: S101 - validated in __post_init__
        if self.door.start is None or self.door.end is None:
            msg = "Door must have defined start and end points."
            raise SimulationModeError(msg)
        return (
            (self.door.start[0] + self.door.end[0]) / 2,
            (self.door.start[1] + self.door.end[1]) / 2,
        )

    def _midpoint_location(
        self, x_offset: float = 0.0, y_offset: float = 0.0
    ) -> Location:
        mid_x, mid_y = self._midpoint()
        return Location(
            building=self.building,
            floor=self.floor,
            x=mid_x + x_offset,
            y=mid_y + y_offset,
        )

    def on_start_moving(self, agent: Agent, engine: SpatialQuery) -> None:
        """Offset the target to the correct side of the door for the destination."""
        candidate = self._midpoint_location(
            x_offset=self.buffer_distance, y_offset=self.buffer_distance
        )
        candidate_room = engine.get_room(agent, coords=(candidate.x, candidate.y))
        if candidate_room is None:
            msg = "Proposed door-buffer location does not correspond to a valid room."
            raise SimulationModeError(msg)

        if candidate_room.room_id == self.destination_room:
            self.location = candidate
        else:
            self.location = self._midpoint_location(
                x_offset=-self.buffer_distance, y_offset=-self.buffer_distance
            )

__post_init__()

Validate door and compute midpoint location.

Source code in src/amr_hub_abm/task/task.py

def __post_init__(self) -> None:
    """Validate door and compute midpoint location."""
    super().__post_init__()
    if self.door is None:
        msg = "TaskDoorAccess requires a door."
        raise SimulationModeError(msg)
    self.location = self._midpoint_location()

on_start_moving(agent, engine)

Offset the target to the correct side of the door for the destination.

Source code in src/amr_hub_abm/task/task.py

def on_start_moving(self, agent: Agent, engine: SpatialQuery) -> None:
    """Offset the target to the correct side of the door for the destination."""
    candidate = self._midpoint_location(
        x_offset=self.buffer_distance, y_offset=self.buffer_distance
    )
    candidate_room = engine.get_room(agent, coords=(candidate.x, candidate.y))
    if candidate_room is None:
        msg = "Proposed door-buffer location does not correspond to a valid room."
        raise SimulationModeError(msg)

    if candidate_room.room_id == self.destination_room:
        self.location = candidate
    else:
        self.location = self._midpoint_location(
            x_offset=-self.buffer_distance, y_offset=-self.buffer_distance
        )

TaskOccupyContent dataclass

Bases: Task

An 'occupy content' task: find matching content in a room and sit on it.

Source code in src/amr_hub_abm/task/task.py

@dataclass
class TaskOccupyContent(Task):
    """An 'occupy content' task: find matching content in a room and sit on it."""

    task_type: TaskType = field(default=TaskType.OCCUPY_CONTENT, kw_only=True)
    content_type: int = -1
    room: Room | None = None
    preferred_content: Content | None = None
    content: Content | None = field(init=False, default=None)

    def __post_init__(self) -> None:
        """Validate that a room is provided."""
        super().__post_init__()
        if self.room is None:
            msg = "TaskOccupyContent requires a room."
            raise SimulationModeError(msg)

    def prepare(self, agent: Agent) -> bool:
        """Resolve an available content instance and set the task location."""
        assert self.room is not None  # noqa: S101 - validated in __post_init__
        matching_content = [
            content
            for content in self.room.contents
            if content.content_type == self.content_type
        ]
        if not matching_content:
            msg = (
                f"No content of type {self.content_type} found in {self.room.name} "
                f"for 'occupy_content' task."
            )
            raise SimulationModeError(msg)

        agent_id = (agent.idx, agent.agent_type)
        candidates = [
            content for content in matching_content if content.is_available_to(agent_id)
        ]
        if not candidates:
            self.content = None
            self.location = None
            return False

        content = self.preferred_content
        if content is None or not any(content is candidate for candidate in candidates):
            content = candidates[0]
        self.content = content
        self.location = content.location
        return True

    def on_dispatch(self, agent: Agent) -> bool:
        """Reserve the selected content before starting travel."""
        if self.content is None:
            return False
        if self.content.try_reserve((agent.idx, agent.agent_type)):
            return True
        self.content = None
        self.location = None
        return False

    def on_completed(
        self, agent: Agent, current_time: int, engine: SpatialQuery
    ) -> None:
        """Occupy the content once the task finishes."""
        agent_id = (agent.idx, agent.agent_type)
        if (
            self.content is None
            or self.content.reserver_id != agent_id
            or not add_agent_occupancy(
                agent, self.content, current_time=current_time, engine=engine
            )
        ):
            msg = f"Agent {agent.idx} could not occupy its reserved content."
            raise SimulationModeError(msg)

__post_init__()

Validate that a room is provided.

Source code in src/amr_hub_abm/task/task.py

def __post_init__(self) -> None:
    """Validate that a room is provided."""
    super().__post_init__()
    if self.room is None:
        msg = "TaskOccupyContent requires a room."
        raise SimulationModeError(msg)

on_completed(agent, current_time, engine)

Occupy the content once the task finishes.

Source code in src/amr_hub_abm/task/task.py

def on_completed(
    self, agent: Agent, current_time: int, engine: SpatialQuery
) -> None:
    """Occupy the content once the task finishes."""
    agent_id = (agent.idx, agent.agent_type)
    if (
        self.content is None
        or self.content.reserver_id != agent_id
        or not add_agent_occupancy(
            agent, self.content, current_time=current_time, engine=engine
        )
    ):
        msg = f"Agent {agent.idx} could not occupy its reserved content."
        raise SimulationModeError(msg)

on_dispatch(agent)

Reserve the selected content before starting travel.

Source code in src/amr_hub_abm/task/task.py

def on_dispatch(self, agent: Agent) -> bool:
    """Reserve the selected content before starting travel."""
    if self.content is None:
        return False
    if self.content.try_reserve((agent.idx, agent.agent_type)):
        return True
    self.content = None
    self.location = None
    return False

prepare(agent)

Resolve an available content instance and set the task location.

Source code in src/amr_hub_abm/task/task.py

def prepare(self, agent: Agent) -> bool:
    """Resolve an available content instance and set the task location."""
    assert self.room is not None  # noqa: S101 - validated in __post_init__
    matching_content = [
        content
        for content in self.room.contents
        if content.content_type == self.content_type
    ]
    if not matching_content:
        msg = (
            f"No content of type {self.content_type} found in {self.room.name} "
            f"for 'occupy_content' task."
        )
        raise SimulationModeError(msg)

    agent_id = (agent.idx, agent.agent_type)
    candidates = [
        content for content in matching_content if content.is_available_to(agent_id)
    ]
    if not candidates:
        self.content = None
        self.location = None
        return False

    content = self.preferred_content
    if content is None or not any(content is candidate for candidate in candidates):
        content = candidates[0]
    self.content = content
    self.location = content.location
    return True

TaskPriority

Bases: IntEnum

Enumeration of possible task priority levels.

Source code in src/amr_hub_abm/task/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/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/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

A 'workstation' task: occupy a workstation in a room.

Source code in src/amr_hub_abm/task/task.py

@dataclass
class TaskWorkstation(Task):
    """A 'workstation' task: occupy a workstation in a room."""

    task_type: TaskType = field(default=TaskType.WORKSTATION, kw_only=True)
    workstation_location: Location | None = None

    def __post_init__(self) -> None:
        """Validate that a room is provided."""
        super().__post_init__()

    def prepare(self, agent: Agent) -> bool:  # noqa: ARG002
        """Set the task location to the workstation location."""
        self.location = self.workstation_location
        return True

__post_init__()

Validate that a room is provided.

Source code in src/amr_hub_abm/task/task.py

def __post_init__(self) -> None:
    """Validate that a room is provided."""
    super().__post_init__()

prepare(agent)

Set the task location to the workstation location.

Source code in src/amr_hub_abm/task/task.py

def prepare(self, agent: Agent) -> bool:  # noqa: ARG002
    """Set the task location to the workstation location."""
    self.location = self.workstation_location
    return True

task_builders

Module for collection of TaskBuilder functions.

TaskBuilderContext dataclass

Context for building tasks.

Source code in src/amr_hub_abm/task/task_builders.py

@dataclass(slots=True, frozen=True)
class TaskBuilderContext:
    """Context for building tasks."""

    time: int
    time_needed: int
    location: Location | None
    patient: Agent | None
    door: Door | None
    destination_room_idx: int | None
    content_type: ContentType | None
    content_room: Room | None
    content: Content | None

build_attend_patient_task(context)

Build a TaskAttendPatient task.

Source code in src/amr_hub_abm/task/task_builders.py

def build_attend_patient_task(context: TaskBuilderContext) -> Task:
    """Build a TaskAttendPatient task."""
    if context.patient is None:
        msg = "Patient must be provided for attend_patient tasks."
        raise SimulationModeError(msg)

    return TaskAttendPatient(
        time_needed=context.time_needed,
        time_due=context.time,
        patient=context.patient,
    )

build_door_access_task(context)

Build a TaskDoorAccess task.

Source code in src/amr_hub_abm/task/task_builders.py

def build_door_access_task(context: TaskBuilderContext) -> Task:
    """Build a TaskDoorAccess task."""
    if context.location is None:
        msg = "Location must be provided for door access tasks."
        raise SimulationModeError(msg)

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

    if context.door is None:
        msg = "Door must be provided in additional_info for door access tasks."
        raise SimulationModeError(msg)

    if context.destination_room_idx is None:
        msg = "Destination room index must be provided for door access tasks."
        raise SimulationModeError(msg)

    return TaskDoorAccess(
        door=context.door,
        destination_room=context.destination_room_idx,
        time_needed=context.time_needed,
        time_due=context.time,
        building=context.location.building,
        floor=context.location.floor,
    )

build_occupy_content_task(context)

Build a TaskOccupyContent task.

Source code in src/amr_hub_abm/task/task_builders.py

def build_occupy_content_task(context: TaskBuilderContext) -> Task:
    """Build a TaskOccupyContent task."""
    if context.content_type is None:
        msg = "Content type must be provided for occupy content tasks."
        raise SimulationModeError(msg)

    if context.content_room is None:
        msg = "Content room must be provided for occupy content tasks."
        raise SimulationModeError(msg)

    return TaskOccupyContent(
        content_type=context.content_type,
        room=context.content_room,
        preferred_content=context.content,
        time_needed=context.time_needed,
        time_due=context.time,
    )

build_task_context(time, time_needed, location, additional_info=None)

Build a TaskBuilderContext from the legacy additional_info dict.

This adapter can be removed once callers pass TaskBuilderContext directly.

Source code in src/amr_hub_abm/task/task_builders.py

def build_task_context(
    time: int,
    time_needed: int,
    location: Location | None,
    additional_info: dict[str, object] | None = None,
) -> TaskBuilderContext:
    """
    Build a TaskBuilderContext from the legacy additional_info dict.

    This adapter can be removed once callers pass TaskBuilderContext directly.
    """
    additional_info = additional_info or {}

    return TaskBuilderContext(
        time=time,
        time_needed=time_needed,
        location=location,
        patient=cast("Agent | None", additional_info.get("patient")),
        door=cast("Door | None", additional_info.get("door")),
        destination_room_idx=cast("int | None", additional_info.get("destination")),
        content_type=cast("ContentType | None", additional_info.get("content_type")),
        content_room=cast("Room | None", additional_info.get("room")),
        content=cast("Content | None", additional_info.get("content")),
    )

build_workstation_task(context)

Build a TaskWorkstation task.

Source code in src/amr_hub_abm/task/task_builders.py

def build_workstation_task(context: TaskBuilderContext) -> Task:
    """Build a TaskWorkstation task."""
    if context.location is None:
        msg = "Location must be provided for workstation tasks."
        raise SimulationModeError(msg)

    return TaskWorkstation(
        workstation_location=context.location,
        time_needed=context.time_needed,
        time_due=context.time,
    )

task_duration

Module defining task duration configurations for the AMR Hub ABM simulation.

TaskDurationConfig dataclass

Task duration parameters read from the simulation configuration file.

Parameters:

Name	Type	Description	Default
time_needed_attend_patient	int	Timesteps needed for a healthcare worker to attend to a patient.	required
time_needed_door_access	int	Timesteps needed for a healthcare worker to access a door (e.g., open/close).	required
time_needed_workstation	int	Timesteps needed for a healthcare worker to use a workstation.	required
time_needed_occupy_content	int	Timesteps needed for a healthcare worker to occupy content (e.g. a bed, a chair, etc.).	required

Source code in src/amr_hub_abm/task/task_duration.py

@dataclass(frozen=True)
class TaskDurationConfig:
    """
    Task duration parameters read from the simulation configuration file.

    Parameters
    ----------
    time_needed_attend_patient : int
        Timesteps needed for a healthcare worker to attend to a patient.
    time_needed_door_access : int
        Timesteps needed for a healthcare worker to access a door (e.g., open/close).
    time_needed_workstation : int
        Timesteps needed for a healthcare worker to use a workstation.
    time_needed_occupy_content : int
        Timesteps needed for a healthcare worker to occupy content
        (e.g. a bed, a chair, etc.).

    """

    time_needed_attend_patient: int
    time_needed_door_access: int
    time_needed_workstation: int
    time_needed_occupy_content: int

    @classmethod
    def from_config(cls, config_data: dict[str, Any]) -> TaskDurationConfig:
        """
        Build a ``TaskDurationConfig`` from parsed YAML configuration data.

        There are deliberately no fallback defaults here: task durations must
        be defined explicitly in the simulation config file.

        Parameters
        ----------
        config_data : dict[str, Any]
            The parsed simulation configuration.

        Returns
        -------
        TaskDurationConfig
            The task duration parameters read from the config.

        """
        missing_keys = [
            key for key in REQUIRED_TASK_DURATION_KEYS if key not in config_data
        ]
        if missing_keys:
            msg = "Missing required task duration keys in simulation config: "
            msg += f"{missing_keys}"
            raise InvalidDefinitionError(msg)

        return cls(
            time_needed_attend_patient=config_data["time_needed_attend_patient"],
            time_needed_door_access=config_data["time_needed_door_access"],
            time_needed_workstation=config_data["time_needed_workstation"],
            time_needed_occupy_content=config_data["time_needed_occupy_content"],
        )

    @property
    def task_duration_mapping(self) -> dict[str, int]:
        """
        Get a mapping of event type strings to their corresponding task durations.

        Returns
        -------
        dict[str, int]
            A dictionary mapping event type strings to task durations in timesteps.

        """
        return {
            "attend_patient": self.time_needed_attend_patient,
            "door_access": self.time_needed_door_access,
            "workstation": self.time_needed_workstation,
            "occupy_content": self.time_needed_occupy_content,
        }

task_duration_mapping property

Get a mapping of event type strings to their corresponding task durations.

Returns:

Type	Description
dict[str, int]	A dictionary mapping event type strings to task durations in timesteps.

from_config(config_data) classmethod

Build a TaskDurationConfig from parsed YAML configuration data.

There are deliberately no fallback defaults here: task durations must be defined explicitly in the simulation config file.

Parameters:

Name	Type	Description	Default
config_data	dict[str, Any]	The parsed simulation configuration.	required

Returns:

Type	Description
TaskDurationConfig	The task duration parameters read from the config.

Source code in src/amr_hub_abm/task/task_duration.py

@classmethod
def from_config(cls, config_data: dict[str, Any]) -> TaskDurationConfig:
    """
    Build a ``TaskDurationConfig`` from parsed YAML configuration data.

    There are deliberately no fallback defaults here: task durations must
    be defined explicitly in the simulation config file.

    Parameters
    ----------
    config_data : dict[str, Any]
        The parsed simulation configuration.

    Returns
    -------
    TaskDurationConfig
        The task duration parameters read from the config.

    """
    missing_keys = [
        key for key in REQUIRED_TASK_DURATION_KEYS if key not in config_data
    ]
    if missing_keys:
        msg = "Missing required task duration keys in simulation config: "
        msg += f"{missing_keys}"
        raise InvalidDefinitionError(msg)

    return cls(
        time_needed_attend_patient=config_data["time_needed_attend_patient"],
        time_needed_door_access=config_data["time_needed_door_access"],
        time_needed_workstation=config_data["time_needed_workstation"],
        time_needed_occupy_content=config_data["time_needed_occupy_content"],
    )

tasklist

Module containing functions for managing tasks in the AMR Hub ABM framework.

perform_in_progress_task(agent, current_time, engine)

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

Source code in src/amr_hub_abm/task/tasklist.py

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

perform_moving_to_task_location(agent, current_time, engine)

Move the agent towards the location of its next task.

Source code in src/amr_hub_abm/task/tasklist.py

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

perform_suspended_task(agent, current_time, engine)

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

Source code in src/amr_hub_abm/task/tasklist.py

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

perform_to_be_started_task(agent, current_time, engine)

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

Source code in src/amr_hub_abm/task/tasklist.py

def perform_to_be_started_task(
    agent: Agent, current_time: int, engine: SpatialQuery
) -> bool:
    """Perform a to-be-started task and return True if a task was performed."""
    task = select_task_based_on_progress(
        agent.tasks, TaskProgress.NOT_STARTED, allow_multiple=True
    )
    if task is None:
        return False
    if not task.prepare(agent=agent):
        agent.tasks[:] = [
            candidate for candidate in agent.tasks if candidate is not task
        ]
        logger.info(
            "Agent %s removed task %s because it could not be prepared.",
            agent.idx,
            task.task_type.name,
        )
        return False
    assert task.location is not None  # noqa: S101

    # Engine handles estimating travel distances!
    task_move_time = (
        task.time_due
        - task.time_needed
        - engine.estimate_time_to_reach_location(agent, task.location)
    )
    logger.info(
        "Agent id %s next task move time: %s, current time: %s",
        agent.idx,
        task_move_time,
        current_time,
    )

    if current_time < task_move_time:
        task_durations = sim_config.task_durations

        agent.attempt_task_insertion(
            next_task=task,
            next_task_move_time=task_move_time,
            current_time=current_time,
            engine=engine,
            task_durations=task_durations,
        )
        return False

    if not task.on_dispatch(agent=agent):
        return True

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

select_task_based_on_progress(tasklist, progress, *, allow_multiple=False)

Select a task based on its progress.

Parameters:

Name	Type	Description	Default
tasklist	list[Task]	The list of tasks to filter.	required
progress	TaskProgress	The progress status to filter tasks by.	required
allow_multiple	bool	Whether to allow multiple tasks with the same progress status.	False

Returns:

Type	Description
Task | None	The selected task with the specified progress status, or None if no such task exists.

Source code in src/amr_hub_abm/task/tasklist.py

def select_task_based_on_progress(
    tasklist: list[Task], progress: TaskProgress, *, allow_multiple: bool = False
) -> Task | None:
    """
    Select a task based on its progress.

    Parameters
    ----------
    tasklist : list[Task]
        The list of tasks to filter.
    progress : TaskProgress
        The progress status to filter tasks by.
    allow_multiple : bool, optional
        Whether to allow multiple tasks with the same progress status.

    Returns
    -------
    Task | None
        The selected task with the specified progress status, or None if no such
        task exists.

    """
    tasks = [task for task in tasklist if task.progress == progress]
    if not tasks:
        return None
    if len(tasks) > 1 and not allow_multiple:
        msg = "Agent 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))