Skip to content

API reference

The public API is everything importable from cuplan. Algorithm modules are documented on their own pages; the shared vocabulary lives here.

Grid and problem types

Occupancy-grid world shared by every solver in cuplan.

The array is the data structure: a boolean (height, width) occupancy map, truthy where blocked, matching pymapf.core.grid.GridMap semantics — 4-connected moves, unit edge costs, one move per timestep — so a scenario ported between the two libraries means the same problem.

Grid

An immutable 4-connected occupancy grid.

Parameters:

Name Type Description Default
obstacles Iterable[Iterable]

2D array-like, truthy where a cell is blocked. Nested lists and NumPy arrays both work; the grid is copied and frozen.

 required
Source code in cuplan/grid.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class Grid:
    """An immutable 4-connected occupancy grid.

    Args:
        obstacles: 2D array-like, truthy where a cell is blocked. Nested
            lists and NumPy arrays both work; the grid is copied and
            frozen.
    """

    def __init__(self, obstacles: Iterable[Iterable]):
        array = np.asarray(obstacles)
        if array.ndim != 2 or array.size == 0:
            raise ValueError("obstacles must be a non-empty 2D array")
        self.obstacles: np.ndarray = array.astype(bool).copy()
        self.obstacles.setflags(write=False)
        self.height, self.width = self.obstacles.shape

    @classmethod
    def empty(cls, height: int, width: int) -> Grid:
        """Return an obstacle-free grid of the given shape."""
        return cls(np.zeros((height, width), dtype=bool))

    @property
    def free(self) -> np.ndarray:
        """Boolean map of traversable cells (the complement of obstacles)."""
        return ~self.obstacles

    @property
    def free_cells(self) -> int:
        """Number of traversable cells."""
        return int(self.free.sum())

    def in_bounds(self, cell: Cell) -> bool:
        r, c = cell
        return 0 <= r < self.height and 0 <= c < self.width

    def is_free(self, cell: Cell) -> bool:
        return self.in_bounds(cell) and not self.obstacles[cell]

    def neighbors(self, cell: Cell) -> list[Cell]:
        """Return the free, in-bounds 4-connected neighbours of ``cell``."""
        r, c = cell
        result = []
        for dr, dc in MOVES[:4]:
            n = (r + int(dr), c + int(dc))
            if self.is_free(n):
                result.append(n)
        return result

    def to_linear(self, cells: np.ndarray) -> np.ndarray:
        """Convert ``(..., 2)`` row/col coordinates to linear indices."""
        cells = np.asarray(cells)
        return cells[..., 0] * self.width + cells[..., 1]

    def from_linear(self, index: np.ndarray) -> np.ndarray:
        """Convert linear indices back to ``(..., 2)`` row/col pairs."""
        index = np.asarray(index)
        return np.stack([index // self.width, index % self.width], axis=-1)

    def __repr__(self) -> str:
        return (
            f"Grid(height={self.height}, width={self.width}, "
            f"obstacles={int(self.obstacles.sum())})"
        )

free property 

free

Boolean map of traversable cells (the complement of obstacles).

free_cells property 

free_cells

Number of traversable cells.

empty classmethod 

empty(height, width)

Return an obstacle-free grid of the given shape.

Source code in cuplan/grid.py 
41
42
43
44
@classmethod
def empty(cls, height: int, width: int) -> Grid:
    """Return an obstacle-free grid of the given shape."""
    return cls(np.zeros((height, width), dtype=bool))

neighbors 

neighbors(cell)

Return the free, in-bounds 4-connected neighbours of cell.

Source code in cuplan/grid.py 
63
64
65
66
67
68
69
70
71
def neighbors(self, cell: Cell) -> list[Cell]:
    """Return the free, in-bounds 4-connected neighbours of ``cell``."""
    r, c = cell
    result = []
    for dr, dc in MOVES[:4]:
        n = (r + int(dr), c + int(dc))
        if self.is_free(n):
            result.append(n)
    return result

to_linear 

to_linear(cells)

Convert (..., 2) row/col coordinates to linear indices.

Source code in cuplan/grid.py 
73
74
75
76
def to_linear(self, cells: np.ndarray) -> np.ndarray:
    """Convert ``(..., 2)`` row/col coordinates to linear indices."""
    cells = np.asarray(cells)
    return cells[..., 0] * self.width + cells[..., 1]

from_linear 

from_linear(index)

Convert linear indices back to (..., 2) row/col pairs.

Source code in cuplan/grid.py 
78
79
80
81
def from_linear(self, index: np.ndarray) -> np.ndarray:
    """Convert linear indices back to ``(..., 2)`` row/col pairs."""
    index = np.asarray(index)
    return np.stack([index // self.width, index % self.width], axis=-1)

Problem and solution types, mirroring pymapf's vocabulary.

Agent/Problem/Solution carry the same semantics as pymapf.core.solver: paths are lists of cells where index t is the position at timestep t, an agent parks on its goal after arrival, and validity means no vertex conflict (two agents on one cell) and no edge conflict (two agents swapping cells between t and t+1).

Agent dataclass

A planning agent: a unique name with start and goal cells.

Source code in cuplan/problem.py 
21
22
23
24
25
26
27
@dataclass(frozen=True)
class Agent:
    """A planning agent: a unique ``name`` with ``start`` and ``goal`` cells."""

    name: str
    start: Cell
    goal: Cell

Problem dataclass

A multi-agent path finding instance on a 4-connected grid.

Source code in cuplan/problem.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass
class Problem:
    """A multi-agent path finding instance on a 4-connected grid."""

    grid: Grid
    agents: list[Agent]

    def __post_init__(self) -> None:
        names = [a.name for a in self.agents]
        if len(names) != len(set(names)):
            raise ValueError("agent names must be unique")
        for agent in self.agents:
            if not self.grid.is_free(agent.start):
                raise ValueError(
                    f"agent {agent.name!r} start {agent.start} is blocked "
                    "or out of bounds"
                )
            if not self.grid.is_free(agent.goal):
                raise ValueError(
                    f"agent {agent.name!r} goal {agent.goal} is blocked "
                    "or out of bounds"
                )

    @property
    def starts(self) -> np.ndarray:
        """``(n_agents, 2)`` start cells, in agent order."""
        return np.array([a.start for a in self.agents], dtype=np.int32)

    @property
    def goals(self) -> np.ndarray:
        """``(n_agents, 2)`` goal cells, in agent order."""
        return np.array([a.goal for a in self.agents], dtype=np.int32)

starts property 

starts

(n_agents, 2) start cells, in agent order.

goals property 

goals

(n_agents, 2) goal cells, in agent order.

Conflict dataclass

A vertex or edge conflict between two agents in a joint plan.

Source code in cuplan/problem.py 
64
65
66
67
68
69
70
71
72
73
@dataclass(frozen=True)
class Conflict:
    """A vertex or edge conflict between two agents in a joint plan."""

    kind: str  # "vertex" or "edge"
    a: str
    b: str
    t: int
    cell_a: Cell
    cell_b: Cell  # equals cell_a for vertex conflicts

Solution dataclass

Result of a solve: one path per agent plus cost metrics.

paths[name][t] is the agent's cell at timestep t; index 0 is the start and the agent stays on its goal after the path ends.

Source code in cuplan/problem.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@dataclass
class Solution:
    """Result of a solve: one path per agent plus cost metrics.

    ``paths[name][t]`` is the agent's cell at timestep ``t``; index 0 is
    the start and the agent stays on its goal after the path ends.
    """

    paths: dict[str, Path]
    algorithm: str = ""
    backend: str = ""
    runtime: float = 0.0  # wall-clock seconds spent in ``solve``
    extra: dict[str, float] = field(default_factory=dict)

    @property
    def makespan(self) -> int:
        """Timestep at which the last agent settles on its goal."""
        return max((len(p) - 1 for p in self.paths.values()), default=0)

    @property
    def sum_of_costs(self) -> int:
        """Sum over agents of the time spent before settling on the goal."""
        return sum(len(p) - 1 for p in self.paths.values())

    def first_conflict(self) -> Conflict | None:
        """Return the earliest conflict, or None for a valid plan."""
        return find_first_conflict(self.paths)

    def is_valid(self) -> bool:
        """True when the joint plan has no vertex or edge conflict."""
        return self.first_conflict() is None

makespan property 

makespan

Timestep at which the last agent settles on its goal.

sum_of_costs property 

sum_of_costs

Sum over agents of the time spent before settling on the goal.

first_conflict 

first_conflict()

Return the earliest conflict, or None for a valid plan.

Source code in cuplan/problem.py 
100
101
102
def first_conflict(self) -> Conflict | None:
    """Return the earliest conflict, or None for a valid plan."""
    return find_first_conflict(self.paths)

is_valid 

is_valid()

True when the joint plan has no vertex or edge conflict.

Source code in cuplan/problem.py 
104
105
106
def is_valid(self) -> bool:
    """True when the joint plan has no vertex or edge conflict."""
    return self.first_conflict() is None

find_first_conflict 

find_first_conflict(paths)

Return the earliest vertex or edge conflict between any agent pair.

Paths are implicitly padded: an agent that has arrived occupies its goal at every later timestep, exactly as in pymapf.

Vectorized over agent pairs per timestep, so validating a 500-agent plan costs milliseconds rather than the O(n^2 T) Python loop it replaces.

Source code in cuplan/problem.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def find_first_conflict(paths: dict[str, Path]) -> Conflict | None:
    """Return the earliest vertex or edge conflict between any agent pair.

    Paths are implicitly padded: an agent that has arrived occupies its
    goal at every later timestep, exactly as in pymapf.

    Vectorized over agent pairs per timestep, so validating a
    500-agent plan costs milliseconds rather than the O(n^2 T) Python
    loop it replaces.
    """
    names = list(paths)
    n = len(names)
    if n < 2:
        return None
    horizon = max(len(p) for p in paths.values())
    # positions[t, i] = linear-ish tuple array; use structured comparison.
    pos = np.empty((horizon + 1, n, 2), dtype=np.int64)
    for i, name in enumerate(names):
        p = np.asarray(paths[name], dtype=np.int64)
        pos[: len(p), i] = p
        pos[len(p) :, i] = p[-1]
    # Encode cells as single integers for fast pairwise comparison.
    width = int(pos[..., 1].max()) + 2
    code = pos[..., 0] * width + pos[..., 1]  # (horizon+1, n)
    for t in range(horizon):
        now, nxt = code[t], code[t + 1]
        # Vertex conflicts at time t: duplicate codes.
        order = np.argsort(now, kind="stable")
        dup = now[order][:-1] == now[order][1:]
        if dup.any():
            k = int(np.argmax(dup))
            i, j = sorted((int(order[k]), int(order[k + 1])))
            cell = tuple(int(x) for x in pos[t, i])
            return Conflict("vertex", names[i], names[j], t, cell, cell)
        # Edge conflicts between t and t+1: i and j swap cells.
        swap = (now[:, None] == nxt[None, :]) & (nxt[:, None] == now[None, :])
        np.fill_diagonal(swap, False)
        if swap.any():
            i, j = np.argwhere(swap)[0]
            i, j = int(min(i, j)), int(max(i, j))
            return Conflict(
                "edge",
                names[i],
                names[j],
                t + 1,
                tuple(int(x) for x in pos[t + 1, i]),
                tuple(int(x) for x in pos[t + 1, j]),
            )
    # Final-timestep vertex conflicts (parked agents sharing a goal).
    now = code[horizon]
    order = np.argsort(now, kind="stable")
    dup = now[order][:-1] == now[order][1:]
    if dup.any():
        k = int(np.argmax(dup))
        i, j = sorted((int(order[k]), int(order[k + 1])))
        cell = tuple(int(x) for x in pos[horizon, i])
        return Conflict("vertex", names[i], names[j], horizon, cell, cell)
    return None

paths_from_array 

paths_from_array(steps, names, goals)

Convert a (T+1, n, 2) position array to per-agent paths.

The parked tail an agent spends on its goal is trimmed, matching pymapf's sum-of-costs convention (waiting on the goal at the end of a plan costs nothing).

Source code in cuplan/problem.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def paths_from_array(
    steps: np.ndarray, names: list[str], goals: np.ndarray
) -> dict[str, Path]:
    """Convert a ``(T+1, n, 2)`` position array to per-agent paths.

    The parked tail an agent spends on its goal is trimmed, matching
    pymapf's sum-of-costs convention (waiting on the goal at the end of
    a plan costs nothing).
    """
    paths: dict[str, Path] = {}
    for i, name in enumerate(names):
        cells: Path = [tuple(int(x) for x in cell) for cell in steps[:, i]]
        goal = tuple(int(x) for x in goals[i])
        end = len(cells) - 1
        while end > 0 and cells[end] == goal and cells[end - 1] == goal:
            end -= 1
        paths[name] = cells[: end + 1]
    return paths

Backend selection

Backend selection: NumPy reference or CUDA via CuPy.

pip install cuplan alone gives the CPU reference backend; installing the [cuda12] or [cuda11] extra adds the CUDA one. CuPy needs only the NVIDIA driver at runtime — kernels are CUDA C compiled on first use through NVRTC, so no CUDA toolkit install is required on the host.

CudaUnavailableError

Bases: RuntimeError

Raised when backend="cuda" is requested but no device works.

Source code in cuplan/backend.py 
21
22
class CudaUnavailableError(RuntimeError):
    """Raised when ``backend="cuda"`` is requested but no device works."""

cuda_available cached 

cuda_available()

Return True if CuPy is importable and a CUDA device executes.

The probe runs one tiny kernel rather than trusting the import: a machine with CuPy installed but no usable driver fails at launch time, and that is the failure this function must report. Set CUPLAN_FORCE_CPU=1 to make it return False, which is how CI tests the fallback path on GPU machines.

Source code in cuplan/backend.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
@functools.lru_cache(maxsize=1)
def cuda_available() -> bool:
    """Return True if CuPy is importable and a CUDA device executes.

    The probe runs one tiny kernel rather than trusting the import: a
    machine with CuPy installed but no usable driver fails at launch
    time, and that is the failure this function must report.
    Set ``CUPLAN_FORCE_CPU=1`` to make it return False, which is how CI
    tests the fallback path on GPU machines.
    """
    if os.environ.get("CUPLAN_FORCE_CPU"):
        return False
    try:
        import cupy

        cupy.cuda.runtime.getDeviceCount()
        # Exercise a real launch: allocation + elementwise kernel + copy.
        result = int((cupy.arange(4, dtype=cupy.int32) ** 2).sum())
        return result == 14
    except Exception:
        return False

resolve_backend 

resolve_backend(backend='auto')

Map "auto" to the best available backend, validating the name.

"auto" prefers CUDA when :func:cuda_available holds, otherwise falls back to the NumPy reference. "cuda" raises :class:CudaUnavailableError instead of silently degrading — a benchmark that quietly ran on the CPU is worse than one that failed.

Source code in cuplan/backend.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def resolve_backend(backend: Backend = "auto") -> Literal["cpu", "cuda"]:
    """Map ``"auto"`` to the best available backend, validating the name.

    ``"auto"`` prefers CUDA when :func:`cuda_available` holds, otherwise
    falls back to the NumPy reference. ``"cuda"`` raises
    :class:`CudaUnavailableError` instead of silently degrading — a
    benchmark that quietly ran on the CPU is worse than one that failed.
    """
    if backend not in _VALID:
        raise ValueError(f"backend must be one of {_VALID}, got {backend!r}")
    if backend == "cpu":
        return "cpu"
    if backend == "cuda":
        if not cuda_available():
            raise CudaUnavailableError(
                "backend='cuda' requested but no working CUDA device was "
                "found. Install cuplan[cuda12] (or [cuda11]) and "
                "check `nvidia-smi`; use backend='auto' to fall back."
            )
        return "cuda"
    return "cuda" if cuda_available() else "cpu"

get_cupy 

get_cupy()

Import and return CuPy, raising a helpful error if absent.

Source code in cuplan/backend.py 
71
72
73
74
75
76
77
78
79
80
def get_cupy():
    """Import and return CuPy, raising a helpful error if absent."""
    try:
        import cupy
    except ImportError as error:  # pragma: no cover - exercised without cupy
        raise CudaUnavailableError(
            "CuPy is not installed. Install cuplan[cuda12] for "
            "CUDA 12.x drivers or cuplan[cuda11] for CUDA 11.x."
        ) from error
    return cupy

Reservations

Space-time reservation table shared by the constrained searches.

The table is two dense arrays over (timestep, cell):

  • vertex[t, v] — cell v is occupied at time t (vertex constraint, Silver 2005).
  • arrived_from[t, v] — linear index of the cell the occupying agent came from, or -1. Because vertex reservations guarantee at most one agent arrives at v per timestep, this single integer encodes every edge (swap) constraint: a move u -> v arriving at t is illegal exactly when arrived_from[t, u] == v.

Dense arrays instead of pymapf's constraint sets is the whole trick: membership tests become array lookups the wavefront can do for every cell at once, on either backend (the xp module is NumPy or CuPy).

ReservationTable

Dense vertex + edge reservations over a bounded time horizon.

Parameters:

Name Type Description Default
grid Grid

the occupancy grid the reservations refer to.

 required
horizon int

last timestep (inclusive) the table covers. Searches against the table cannot return paths longer than this.

 required
xp module

array module — numpy (default) or cupy. Solvers running on the CUDA backend keep the table on the device so reserving a path never round-trips through host memory.

 numpy
Source code in cuplan/reservations.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class ReservationTable:
    """Dense vertex + edge reservations over a bounded time horizon.

    Args:
        grid: the occupancy grid the reservations refer to.
        horizon: last timestep (inclusive) the table covers. Searches
            against the table cannot return paths longer than this.
        xp (module): array module — ``numpy`` (default) or ``cupy``. Solvers
            running on the CUDA backend keep the table on the device so
            reserving a path never round-trips through host memory.
    """

    def __init__(self, grid: Grid, horizon: int, xp=np):
        if horizon < 1:
            raise ValueError("horizon must be >= 1")
        self.grid = grid
        self.horizon = int(horizon)
        self.xp = xp
        cells = grid.height * grid.width
        self.vertex = xp.zeros((self.horizon + 1, cells), dtype=xp.uint8)
        self.arrived_from = xp.full(
            (self.horizon + 1, cells), -1, dtype=xp.int32
        )

    def reserve_path(self, path: Sequence[tuple]) -> None:
        """Reserve a full agent path, parking it on its last cell forever.

        ``path[t]`` is the agent's cell at time ``t``. After the path
        ends the agent is assumed to stay on its final cell, so that
        cell is blocked through the end of the horizon — the same
        convention as pymapf's prioritized planner.
        """
        xp = self.xp
        cells = np.asarray(path, dtype=np.int64)
        linear = cells[:, 0] * self.grid.width + cells[:, 1]
        if len(linear) > self.horizon + 1:
            raise ValueError("path is longer than the table horizon")
        steps = xp.asarray(linear)
        t = xp.arange(len(linear))
        self.vertex[t, steps] = 1
        # Park on the final cell for the rest of the horizon.
        self.vertex[len(linear) :, int(linear[-1])] = 1
        # Record arrivals for edge (swap) constraints.
        if len(linear) > 1:
            self.arrived_from[t[1:], steps[1:]] = steps[:-1].astype(xp.int32)

    def block_vertex(self, cell: tuple, t: int) -> None:
        """Add a single vertex constraint: ``cell`` is occupied at ``t``."""
        r, c = cell
        self.vertex[t, r * self.grid.width + c] = 1

    def block_edge(self, u: tuple, v: tuple, t: int) -> None:
        """Forbid traversing ``u -> v`` arriving at time ``t``."""
        w = self.grid.width
        self.arrived_from[t, u[0] * w + u[1]] = np.int32(v[0] * w + v[1])

    def last_vertex_time(self, cell: tuple) -> int:
        """Latest ``t`` with a vertex reservation on ``cell`` (-1 if none).

        A search must not settle on its goal before this time — the
        reservation could push it off again.
        """
        r, c = cell
        column = self.vertex[:, r * self.grid.width + c]
        hits = self.xp.flatnonzero(column)
        return int(hits[-1]) if len(hits) else -1

reserve_path 

reserve_path(path)

Reserve a full agent path, parking it on its last cell forever.

path[t] is the agent's cell at time t. After the path ends the agent is assumed to stay on its final cell, so that cell is blocked through the end of the horizon — the same convention as pymapf's prioritized planner.

Source code in cuplan/reservations.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def reserve_path(self, path: Sequence[tuple]) -> None:
    """Reserve a full agent path, parking it on its last cell forever.

    ``path[t]`` is the agent's cell at time ``t``. After the path
    ends the agent is assumed to stay on its final cell, so that
    cell is blocked through the end of the horizon — the same
    convention as pymapf's prioritized planner.
    """
    xp = self.xp
    cells = np.asarray(path, dtype=np.int64)
    linear = cells[:, 0] * self.grid.width + cells[:, 1]
    if len(linear) > self.horizon + 1:
        raise ValueError("path is longer than the table horizon")
    steps = xp.asarray(linear)
    t = xp.arange(len(linear))
    self.vertex[t, steps] = 1
    # Park on the final cell for the rest of the horizon.
    self.vertex[len(linear) :, int(linear[-1])] = 1
    # Record arrivals for edge (swap) constraints.
    if len(linear) > 1:
        self.arrived_from[t[1:], steps[1:]] = steps[:-1].astype(xp.int32)

block_vertex 

block_vertex(cell, t)

Add a single vertex constraint: cell is occupied at t.

Source code in cuplan/reservations.py 
75
76
77
78
def block_vertex(self, cell: tuple, t: int) -> None:
    """Add a single vertex constraint: ``cell`` is occupied at ``t``."""
    r, c = cell
    self.vertex[t, r * self.grid.width + c] = 1

block_edge 

block_edge(u, v, t)

Forbid traversing u -> v arriving at time t.

Source code in cuplan/reservations.py 
80
81
82
83
def block_edge(self, u: tuple, v: tuple, t: int) -> None:
    """Forbid traversing ``u -> v`` arriving at time ``t``."""
    w = self.grid.width
    self.arrived_from[t, u[0] * w + u[1]] = np.int32(v[0] * w + v[1])

last_vertex_time 

last_vertex_time(cell)

Latest t with a vertex reservation on cell (-1 if none).

A search must not settle on its goal before this time — the reservation could push it off again.

Source code in cuplan/reservations.py 
85
86
87
88
89
90
91
92
93
94
def last_vertex_time(self, cell: tuple) -> int:
    """Latest ``t`` with a vertex reservation on ``cell`` (-1 if none).

    A search must not settle on its goal before this time — the
    reservation could push it off again.
    """
    r, c = cell
    column = self.vertex[:, r * self.grid.width + c]
    hits = self.xp.flatnonzero(column)
    return int(hits[-1]) if len(hits) else -1

Kernel loading

CUDA C kernel sources and their NVRTC loader.

Kernels ship as .cu source files inside the wheel and are compiled on first use through CuPy's :class:cupy.RawModule (NVRTC). CuPy caches compiled cubins on disk, so the cost is paid once per machine, not once per process.

kernel_source 

kernel_source(name)

Return the CUDA C source of kernels/<name>.cu.

Source code in cuplan/kernels/__init__.py 
19
20
21
def kernel_source(name: str) -> str:
    """Return the CUDA C source of ``kernels/<name>.cu``."""
    return (resources.files(__package__) / f"{name}.cu").read_text()

kernel_names 

kernel_names()

List the kernel source files bundled with the package.

Source code in cuplan/kernels/__init__.py 
24
25
26
27
28
29
30
def kernel_names() -> list[str]:
    """List the kernel source files bundled with the package."""
    return sorted(
        path.name[:-3]
        for path in resources.files(__package__).iterdir()
        if path.name.endswith(".cu")
    )

load_module cached 

load_module(name)

Compile kernels/<name>.cu and return the cupy.RawModule.

Source code in cuplan/kernels/__init__.py 
33
34
35
36
37
@functools.cache
def load_module(name: str):
    """Compile ``kernels/<name>.cu`` and return the ``cupy.RawModule``."""
    cupy = get_cupy()
    return cupy.RawModule(code=kernel_source(name), options=_OPTIONS)

get_kernel 

get_kernel(module_name, kernel)

Return a launchable cupy.RawKernel from a bundled module.

Source code in cuplan/kernels/__init__.py 
40
41
42
def get_kernel(module_name: str, kernel: str):
    """Return a launchable ``cupy.RawKernel`` from a bundled module."""
    return load_module(module_name).get_function(kernel)

Roadmap stubs

Planned solvers: documented stubs, not implementations.

Each class below names the algorithm, the paper, and the intended parallelization strategy, and raises :class:NotImplementedError from its constructor so nothing can mistake a stub for a solver. pymapf has working CPU implementations of all of them.

CBS

Bases: _Planned

Conflict-Based Search (Sharon et al. 2015, AIJ 219:40-66).

Optimal two-level search. GPU plan: the high-level constraint tree is sequential, but sibling nodes' low-level searches are independent — batch them as parallel space-time wavefronts, one stream each.

Source code in cuplan/roadmap.py 
25
26
27
28
29
30
31
class CBS(_Planned):
    """Conflict-Based Search (Sharon et al. 2015, AIJ 219:40-66).

    Optimal two-level search. GPU plan: the high-level constraint tree
    is sequential, but sibling nodes' low-level searches are independent
    — batch them as parallel space-time wavefronts, one stream each.
    """

LaCAM

Bases: _Planned

LaCAM (Okumura 2023, AAAI): complete search wrapping PIBT.

GPU plan: reuse cuplan's PIBT step (batched candidate evaluation); the lazy high-level DFS stays on the host.

Source code in cuplan/roadmap.py 
34
35
36
37
38
39
class LaCAM(_Planned):
    """LaCAM (Okumura 2023, AAAI): complete search wrapping PIBT.

    GPU plan: reuse cuplan's PIBT step (batched candidate evaluation);
    the lazy high-level DFS stays on the host.
    """

LNS

Bases: _Planned

MAPF-LNS (Li et al. 2021, IJCAI): large neighbourhood search.

GPU plan: destroy/repair proposals are independent — evaluate many neighbourhoods concurrently and keep the best repair.

Source code in cuplan/roadmap.py 
42
43
44
45
46
47
class LNS(_Planned):
    """MAPF-LNS (Li et al. 2021, IJCAI): large neighbourhood search.

    GPU plan: destroy/repair proposals are independent — evaluate many
    neighbourhoods concurrently and keep the best repair.
    """

SIPP

Bases: _Planned

Safe Interval Path Planning (Phillips and Likhachev 2011, ICRA).

GPU plan: safe-interval construction from a reservation table is a per-cell scan (one thread per cell); the interval graph search itself is small enough to stay on the host.

Source code in cuplan/roadmap.py 
50
51
52
53
54
55
56
class SIPP(_Planned):
    """Safe Interval Path Planning (Phillips and Likhachev 2011, ICRA).

    GPU plan: safe-interval construction from a reservation table is a
    per-cell scan (one thread per cell); the interval graph search
    itself is small enough to stay on the host.
    """

NMPC

Bases: _Planned

Decentralized nonlinear MPC (mirroring pymapf's NMPC agent).

GPU plan: sampling-based MPC (MPPI) — thousands of rollouts per agent per step, each one thread.

Source code in cuplan/roadmap.py 
59
60
61
62
63
64
class NMPC(_Planned):
    """Decentralized nonlinear MPC (mirroring pymapf's NMPC agent).

    GPU plan: sampling-based MPC (MPPI) — thousands of rollouts per
    agent per step, each one thread.
    """

Benchmark harness

Reproducible random MAPF scenarios shared across libraries.

A scenario is a seeded random obstacle grid plus distinct start and goal cells, all mutually reachable (verified with a flood fill from the first start). The same object converts to a cuplan :class:~cuplan.problem.Problem and a pymapf MAPFProblem, which is what makes the benchmark apples-to-apples.

Scenario dataclass

A reproducible MAPF instance description.

Source code in cuplan/benchmark/scenarios.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@dataclass(frozen=True)
class Scenario:
    """A reproducible MAPF instance description."""

    grid: Grid
    starts: tuple[tuple[int, int], ...]
    goals: tuple[tuple[int, int], ...]
    seed: int

    @property
    def n_agents(self) -> int:
        return len(self.starts)

    def to_cuplan(self) -> Problem:
        """Return the instance as a cuplan :class:`~cuplan.problem.Problem`."""
        agents = [
            Agent(name=f"a{i}", start=s, goal=g)
            for i, (s, g) in enumerate(zip(self.starts, self.goals, strict=True))
        ]
        return Problem(grid=self.grid, agents=agents)

    def to_pymapf(self):
        """Return the instance as a ``pymapf`` problem (imported lazily)."""
        from pymapf.core.grid import GridMap
        from pymapf.core.solver import Agent as PAgent
        from pymapf.core.solver import MAPFProblem

        grid = GridMap(self.grid.obstacles.astype(int).tolist())
        agents = [
            PAgent(name=f"a{i}", start=s, goal=g)
            for i, (s, g) in enumerate(zip(self.starts, self.goals, strict=True))
        ]
        return MAPFProblem(grid=grid, agents=agents)

to_cuplan 

to_cuplan()

Return the instance as a cuplan :class:~cuplan.problem.Problem.

Source code in cuplan/benchmark/scenarios.py 
36
37
38
39
40
41
42
def to_cuplan(self) -> Problem:
    """Return the instance as a cuplan :class:`~cuplan.problem.Problem`."""
    agents = [
        Agent(name=f"a{i}", start=s, goal=g)
        for i, (s, g) in enumerate(zip(self.starts, self.goals, strict=True))
    ]
    return Problem(grid=self.grid, agents=agents)

to_pymapf 

to_pymapf()

Return the instance as a pymapf problem (imported lazily).

Source code in cuplan/benchmark/scenarios.py 
44
45
46
47
48
49
50
51
52
53
54
55
def to_pymapf(self):
    """Return the instance as a ``pymapf`` problem (imported lazily)."""
    from pymapf.core.grid import GridMap
    from pymapf.core.solver import Agent as PAgent
    from pymapf.core.solver import MAPFProblem

    grid = GridMap(self.grid.obstacles.astype(int).tolist())
    agents = [
        PAgent(name=f"a{i}", start=s, goal=g)
        for i, (s, g) in enumerate(zip(self.starts, self.goals, strict=True))
    ]
    return MAPFProblem(grid=grid, agents=agents)

random_scenario 

random_scenario(size, n_agents, obstacle_density=0.15, seed=0)

Generate a connected random instance.

Obstacles are sampled i.i.d. at obstacle_density; starts and goals are distinct free cells drawn from the largest connected component, so every agent's goal is reachable.

Parameters:

Name Type Description Default
size int

grid is size x size.

 required
n_agents int

number of agents (must fit in the free space).

 required
obstacle_density float

fraction of blocked cells.

 0.15
seed int

RNG seed; same seed, same instance.

 0
Source code in cuplan/benchmark/scenarios.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def random_scenario(
    size: int,
    n_agents: int,
    obstacle_density: float = 0.15,
    seed: int = 0,
) -> Scenario:
    """Generate a connected random instance.

    Obstacles are sampled i.i.d. at ``obstacle_density``; starts and
    goals are distinct free cells drawn from the largest connected
    component, so every agent's goal is reachable.

    Args:
        size: grid is ``size x size``.
        n_agents: number of agents (must fit in the free space).
        obstacle_density: fraction of blocked cells.
        seed: RNG seed; same seed, same instance.
    """
    rng = np.random.default_rng(seed)
    for _attempt in range(64):
        obstacles = rng.random((size, size)) < obstacle_density
        grid = Grid(obstacles)
        free = np.argwhere(grid.free)
        if len(free) < 2 * n_agents:
            continue
        # Largest connected component via one flood fill per candidate seed.
        seed_cell = free[rng.integers(len(free))]
        dist = distance_maps(grid, seed_cell[None, :], backend="cpu")[0]
        component = np.argwhere(dist >= 0)
        if len(component) < 2 * n_agents:
            continue
        picks = rng.choice(len(component), size=2 * n_agents, replace=False)
        cells: list[tuple[int, int]] = [
            (int(r), int(c)) for r, c in component[picks]
        ]
        return Scenario(
            grid=grid,
            starts=tuple(cells[:n_agents]),
            goals=tuple(cells[n_agents:]),
            seed=seed,
        )
    raise RuntimeError(
        f"could not build a scenario with {n_agents} agents on a "
        f"{size}x{size} grid at density {obstacle_density}"
    )

Benchmark runners: same scenarios, every solver, honest numbers.

Each record carries the machine-independent facts (solver, backend, grid size, agent count, seed) and the measured ones (wall time, sum of costs, success). Wall time covers the full solve call including host/device transfers — the number a user would actually see.

BenchmarkResult dataclass

One (scenario, solver) measurement.

Source code in cuplan/benchmark/harness.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@dataclass
class BenchmarkResult:
    """One (scenario, solver) measurement."""

    family: str  # "prioritized" | "pibt" | "bfs" | "velocity_obstacles"
    solver: str  # e.g. "cuplan-cuda", "pymapf"
    size: int
    n_agents: int
    seed: int
    success: bool
    runtime: float  # seconds, full solve() including transfers
    cost: int | None = None  # sum of costs, when solved
    extra: dict[str, float] = field(default_factory=dict)

    def as_dict(self) -> dict:
        return asdict(self)

machine_description 

machine_description()

One-line description of the benchmark machine.

Source code in cuplan/benchmark/harness.py 
44
45
46
47
48
49
50
51
52
def machine_description() -> str:
    """One-line description of the benchmark machine."""
    gpu = ""
    if cuda_available():
        import cupy

        props = cupy.cuda.runtime.getDeviceProperties(0)
        gpu = f", GPU {props['name'].decode()}"
    return f"{platform.processor() or platform.machine()}{gpu}"

run_mapf_benchmark 

run_mapf_benchmark(sizes, agent_counts, seeds, obstacle_density=0.15, include_pymapf=True, include_cuda=None, progress=None)

Run prioritized planning and PIBT across scenario axes.

Every (size, agents, seed) triple builds one scenario handed to all solvers. Agent counts that do not fit a grid size are skipped.

Returns the flat list of records; aggregation is the reporter's job.

Source code in cuplan/benchmark/harness.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def run_mapf_benchmark(
    sizes: list[int],
    agent_counts: list[int],
    seeds: list[int],
    obstacle_density: float = 0.15,
    include_pymapf: bool = True,
    include_cuda: bool | None = None,
    progress: Callable[[str], None] | None = None,
) -> list[BenchmarkResult]:
    """Run prioritized planning and PIBT across scenario axes.

    Every (size, agents, seed) triple builds one scenario handed to all
    solvers. Agent counts that do not fit a grid size are skipped.

    Returns the flat list of records; aggregation is the reporter's job.
    """
    include_cuda = cuda_available() if include_cuda is None else include_cuda
    say = progress or (lambda s: None)
    results: list[BenchmarkResult] = []

    for size in sizes:
        for n_agents in agent_counts:
            if 2 * n_agents > size * size * (1 - obstacle_density) * 0.5:
                continue
            for seed in seeds:
                scenario = random_scenario(
                    size, n_agents, obstacle_density, seed
                )
                say(f"{size}x{size}, {n_agents} agents, seed {seed}")
                results.extend(
                    _run_mapf_instance(
                        scenario, include_pymapf, include_cuda
                    )
                )
    return results

run_bfs_benchmark 

run_bfs_benchmark(sizes, batch_sizes, seeds, obstacle_density=0.15, include_cuda=None, progress=None)

Benchmark the batched distance-map primitive: CPU vs CUDA.

This is the primitive every solver consumes (heuristic tables, the PIBT oracle), measured directly: one flood fill per source, batched. pymapf has no batched equivalent — its per-goal Dijkstra cost is included in the solver families' timings.

Source code in cuplan/benchmark/harness.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def run_bfs_benchmark(
    sizes: list[int],
    batch_sizes: list[int],
    seeds: list[int],
    obstacle_density: float = 0.15,
    include_cuda: bool | None = None,
    progress: Callable[[str], None] | None = None,
) -> list[BenchmarkResult]:
    """Benchmark the batched distance-map primitive: CPU vs CUDA.

    This is the primitive every solver consumes (heuristic tables, the
    PIBT oracle), measured directly: one flood fill per source, batched.
    pymapf has no batched equivalent — its per-goal Dijkstra cost is
    included in the solver families' timings.
    """
    from ..bfs import distance_maps

    include_cuda = cuda_available() if include_cuda is None else include_cuda
    say = progress or (lambda s: None)
    results: list[BenchmarkResult] = []
    backends = ["cpu"] + (["cuda"] if include_cuda else [])

    for size in sizes:
        for batch in batch_sizes:
            for seed in seeds:
                scenario = random_scenario(
                    size,
                    min(batch, int(size * size * 0.2)),
                    obstacle_density,
                    seed,
                )
                goals = np.asarray(scenario.goals)
                for backend in backends:
                    say(f"bfs {size}x{size} batch {len(goals)} {backend}")
                    if backend == "cuda":  # warm the NVRTC cache
                        distance_maps(scenario.grid, goals[:1], backend="cuda")
                    started = time.perf_counter()
                    distance_maps(scenario.grid, goals, backend=backend)
                    runtime = time.perf_counter() - started
                    results.append(
                        BenchmarkResult(
                            family="bfs",
                            solver=f"cuplan-{backend}",
                            size=size,
                            n_agents=len(goals),
                            seed=seed,
                            success=True,
                            runtime=runtime,
                        )
                    )
    return results

run_vo_benchmark 

run_vo_benchmark(agent_counts, seeds, n_steps=80, include_cuda=None, progress=None)

Benchmark velocity-obstacle steps: cuplan CPU vs CUDA.

Agents start on a circle with antipodal goals — the classic all-cross stress case. pymapf's simulator is not timed here: its sequential in-step update solves a different problem per agent (see :mod:cuplan.velocity_obstacles), so wall-clock comparison would be misleading; the MAPF families carry the cross-library numbers.

Source code in cuplan/benchmark/harness.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def run_vo_benchmark(
    agent_counts: list[int],
    seeds: list[int],
    n_steps: int = 80,
    include_cuda: bool | None = None,
    progress: Callable[[str], None] | None = None,
) -> list[BenchmarkResult]:
    """Benchmark velocity-obstacle steps: cuplan CPU vs CUDA.

    Agents start on a circle with antipodal goals — the classic
    all-cross stress case. pymapf's simulator is not timed here: its
    sequential in-step update solves a different problem per agent (see
    :mod:`cuplan.velocity_obstacles`), so wall-clock comparison would
    be misleading; the MAPF families carry the cross-library numbers.
    """
    from ..velocity_obstacles import VelocityObstacleSim

    include_cuda = cuda_available() if include_cuda is None else include_cuda
    say = progress or (lambda s: None)
    results: list[BenchmarkResult] = []
    backends = ["cpu"] + (["cuda"] if include_cuda else [])

    for n_agents in agent_counts:
        for seed in seeds:
            rng = np.random.default_rng(seed)
            angles = np.sort(rng.uniform(0, 2 * np.pi, n_agents))
            r = 2.0 + 0.35 * n_agents
            starts = r * np.stack([np.cos(angles), np.sin(angles)], axis=1)
            goals = -starts
            for backend in backends:
                say(f"vo {n_agents} agents seed {seed} {backend}")
                sim = VelocityObstacleSim(backend=backend)
                for s, g in zip(starts, goals, strict=True):
                    sim.add_agent(s, g)
                run = sim.run(n_steps)
                reached = run.goals_reached(goals, tolerance=0.5)
                results.append(
                    BenchmarkResult(
                        family="velocity_obstacles",
                        solver=f"cuplan-{backend}",
                        size=0,
                        n_agents=n_agents,
                        seed=seed,
                        success=True,
                        runtime=run.runtime,
                        cost=None,
                        extra={
                            "goals_reached": float(reached),
                            "min_separation": run.min_separation(),
                        },
                    )
                )
    return results

Turn benchmark records into CSV, Markdown, and Frontier-styled charts.

write_csv 

write_csv(results, path)

Write the flat records as CSV (one row per measurement).

Source code in cuplan/benchmark/report.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def write_csv(results: list[BenchmarkResult], path: Path) -> None:
    """Write the flat records as CSV (one row per measurement)."""
    fields = [
        "family",
        "solver",
        "size",
        "n_agents",
        "seed",
        "success",
        "runtime",
        "cost",
    ]
    with path.open("w", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fields, extrasaction="ignore")
        writer.writeheader()
        for record in results:
            writer.writerow(record.as_dict())

read_csv 

read_csv(path)

Load records written by :func:write_csv, e.g. to re-render reports.

Source code in cuplan/benchmark/report.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def read_csv(path: Path) -> list[BenchmarkResult]:
    """Load records written by :func:`write_csv`, e.g. to re-render reports."""
    records = []
    with path.open() as handle:
        for row in csv.DictReader(handle):
            records.append(
                BenchmarkResult(
                    family=row["family"],
                    solver=row["solver"],
                    size=int(row["size"]),
                    n_agents=int(row["n_agents"]),
                    seed=int(row["seed"]),
                    success=row["success"] == "True",
                    runtime=float(row["runtime"]),
                    cost=int(float(row["cost"])) if row["cost"] else None,
                )
            )
    return records

write_markdown 

write_markdown(results, path)

Write an aggregated Markdown table with the measurement conditions.

Source code in cuplan/benchmark/report.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def write_markdown(results: list[BenchmarkResult], path: Path) -> None:
    """Write an aggregated Markdown table with the measurement conditions."""
    agg = _aggregate(results)
    lines = [
        "# Benchmark results",
        "",
        f"Machine: {machine_description()}. Median over seeds; wall time",
        "covers the full solve including host/device transfers. Random",
        "grids at 15% obstacle density; identical instances for every",
        "solver. Reproduce with `python -m cuplan.benchmark`.",
        "",
        "| family | solver | grid | agents | success | median time (s) | median cost |",
        "| :-- | :-- | --: | --: | --: | --: | --: |",
    ]
    for key in sorted(agg):
        family, solver, size, n_agents = key
        row = agg[key]
        time_s = (
            f"{row['median_runtime']:.4f}"
            if row["median_runtime"] is not None
            else "—"
        )
        cost = (
            f"{row['median_cost']:.0f}" if row["median_cost"] is not None else "—"
        )
        grid = f"{size}x{size}" if size else "—"
        lines.append(
            f"| {family} | {solver} | {grid} | {n_agents} | "
            f"{row['success_rate']:.0%} | {time_s} | {cost} |"
        )
    path.write_text("\n".join(lines) + "\n")

write_charts 

write_charts(results, out_dir)

Render runtime-scaling charts per family. Returns written paths.

Source code in cuplan/benchmark/report.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
def write_charts(results: list[BenchmarkResult], out_dir: Path) -> list[Path]:
    """Render runtime-scaling charts per family. Returns written paths."""
    import matplotlib

    matplotlib.use("Agg")
    import matplotlib.pyplot as plt

    _style()
    agg = _aggregate(results)
    written: list[Path] = []

    families = sorted({k[0] for k in agg})
    for family in families:
        sizes = sorted({k[2] for k in agg if k[0] == family})
        largest = sizes[-1] if sizes else 0
        keys = [k for k in agg if k[0] == family and k[2] == largest]
        solvers = sorted({k[1] for k in keys})
        fig, ax = plt.subplots(figsize=(6.4, 4.0))
        for solver in solvers:
            points = sorted(
                (k[3], agg[k]["median_runtime"])
                for k in keys
                if k[1] == solver and agg[k]["median_runtime"] is not None
            )
            if not points:
                continue
            xs, ys = zip(*points, strict=True)
            ax.plot(
                xs,
                ys,
                marker="o",
                label=solver,
                color=_SERIES_COLORS.get(solver),
            )
        ax.set_yscale("log")
        ax.set_xlabel(
            "sources (one BFS per agent)" if family == "bfs" else "agents"
        )
        ax.set_ylabel("median wall time (s, log scale)")
        grid_note = f" — {largest}x{largest} grid" if largest else ""
        ax.set_title(f"{family}{grid_note}")
        ax.legend()
        path = out_dir / f"{family}-scaling.png"
        fig.savefig(path)
        plt.close(fig)
        written.append(path)
    return written