Index

Configuration tree search for valid atom move programs.

AStarTraversal dataclass

AStarTraversal(
    heuristic: HeuristicFunction,
    cost: CostFunction | None = None,
)

Bases: TraversalStrategyABC

A* traversal policy.

search

search(
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None
) -> SearchResult

Execute search over the tree using the supplied generator.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/astar.py

def search(
    self,
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None,
) -> SearchResult:
    _ = max_depth
    cost_fn: CostFunction = (
        self.cost if self.cost is not None else lambda node: float(node.depth)
    )
    frontier: list[PriorityEntry] = []
    root_f = cost_fn(tree.root) + self.heuristic(tree.root)
    heapq.heappush(frontier, PriorityEntry(root_f, tree.root))
    return run_frontier_search(
        tree=tree,
        generator=generator,
        goal=goal,
        pop_next=lambda: heapq.heappop(frontier).node if frontier else None,
        push_child=lambda child: heapq.heappush(
            frontier, PriorityEntry(cost_fn(child) + self.heuristic(child), child)
        ),
        frontier_has_items=lambda: bool(frontier),
        max_expansions=max_expansions,
        max_depth=None,
        goal_on_pop=True,
    )

BFSTraversal

Bases: TraversalStrategyABC

Breadth-first traversal policy.

search

search(
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None
) -> SearchResult

Execute search over the tree using the supplied generator.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/bfs.py

def search(
    self,
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None,
) -> SearchResult:
    frontier: deque[ConfigurationNode] = deque([tree.root])
    return run_frontier_search(
        tree=tree,
        generator=generator,
        goal=goal,
        pop_next=lambda: frontier.popleft() if frontier else None,
        push_child=frontier.append,
        frontier_has_items=lambda: bool(frontier),
        max_expansions=max_expansions,
        max_depth=max_depth,
        goal_on_pop=False,
    )

CandidateScorer dataclass

CandidateScorer(
    params: SearchParams, target: dict[int, LocationAddress]
)

Scores qubit-bus pairs and movesets for entropy-guided search.

score_all_qubit_bus_pairs

score_all_qubit_bus_pairs(
    node: ConfigurationNode,
    entropy: int,
    tree: ConfigurationTree,
) -> dict[tuple[int, MoveType, int, Direction], float]

Score all legal (qubit, move_type, bus_id, direction) tuples.

Returns mapping of (qubit_id, move_type, bus_id, direction) -> score. Only includes legal tuples where the qubit is on the bus source and the destination is unoccupied.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/scoring.py

def score_all_qubit_bus_pairs(
    self,
    node: ConfigurationNode,
    entropy: int,
    tree: ConfigurationTree,
) -> dict[tuple[int, MoveType, int, Direction], float]:
    """Score all legal (qubit, move_type, bus_id, direction) tuples.

    Returns mapping of (qubit_id, move_type, bus_id, direction) -> score.
    Only includes legal tuples where the qubit is on the bus source and
    the destination is unoccupied.
    """
    occupied = node.occupied_locations | tree.blocked_locations
    e_eff = min(entropy, self.params.e_max)

    # Identify unresolved qubits
    unresolved = {
        qid: loc
        for qid, loc in node.configuration.items()
        if qid in self.target and loc != self.target[qid]
    }
    if not unresolved:
        return {}

    # Cache d_now and m_now per qubit (same across all buses)
    d_now: dict[int, float] = {}
    m_now: dict[int, int] = {}
    for qid, loc in unresolved.items():
        d_now[qid] = self._distance_to_target(loc, self.target[qid], tree)
        m_now[qid] = self._mobility_at(loc, occupied, tree)

    # Compute raw deltas for all legal (qubit, move_type, bus_id, direction)
    d_after_cache: dict[tuple[int, LocationAddress], float] = {}
    m_after_cache: dict[LocationAddress, int] = {}
    raw_scores: dict[tuple[int, MoveType, int, Direction], tuple[float, float]] = {}
    for mt in (MoveType.SITE, MoveType.WORD):
        buses = (
            tree.arch_spec.site_buses
            if mt == MoveType.SITE
            else tree.arch_spec.word_buses
        )
        for bus_id in range(len(buses)):
            for direction in (Direction.FORWARD, Direction.BACKWARD):
                for qid, loc in unresolved.items():
                    lane = tree.lane_for_source(mt, bus_id, direction, loc)
                    if lane is None:
                        continue
                    _, dst = tree.arch_spec.get_endpoints(lane)
                    if dst in occupied:
                        continue
                    d_key = (qid, dst)
                    d_after = d_after_cache.get(d_key)
                    if d_after is None:
                        d_after = self._distance_to_target(
                            dst, self.target[qid], tree
                        )
                        d_after_cache[d_key] = d_after
                    m_after = m_after_cache.get(dst)
                    if m_after is None:
                        m_after = self._mobility_at(dst, occupied, tree)
                        m_after_cache[dst] = m_after
                    delta_d = d_now[qid] - d_after
                    delta_m = m_after - m_now[qid]
                    raw_scores[(qid, mt, bus_id, direction)] = (delta_d, delta_m)

    if not raw_scores:
        return {}

    # Normalize
    d_ref = max(1.0, max(abs(dd) for dd, _ in raw_scores.values()))
    m_ref = max(1.0, max(abs(dm) for _, dm in raw_scores.values()))

    # Apply entropy-weighted formula
    result: dict[tuple[int, MoveType, int, Direction], float] = {}
    for key, (delta_d, delta_m) in raw_scores.items():
        d_hat = delta_d / d_ref
        m_hat = delta_m / m_ref
        score = (self.params.w_d / e_eff) * d_hat + self.params.w_m * e_eff * m_hat
        result[key] = score

    return result

score_moveset

score_moveset(
    moveset: frozenset[LaneAddress],
    node: ConfigurationNode,
    tree: ConfigurationTree,
) -> float

Score a candidate moveset.

Returns alphadistance_moved + betaarrived_gain + gamma*mobility_gain.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/scoring.py

def score_moveset(
    self,
    moveset: frozenset[LaneAddress],
    node: ConfigurationNode,
    tree: ConfigurationTree,
) -> float:
    """Score a candidate moveset.

    Returns alpha*distance_moved + beta*arrived_gain + gamma*mobility_gain.
    """
    occupied = node.occupied_locations | tree.blocked_locations
    distance_moved = 0.0
    arrived_gain = 0
    mobility_before = 0
    mobility_after = 0

    # Identify moved qubits and build post-move config
    new_config: dict[int, LocationAddress] = dict(node.configuration)
    moved_qubits: list[tuple[int, LocationAddress, LocationAddress]] = []

    for lane in moveset:
        src, dst = tree.arch_spec.get_endpoints(lane)
        qid = node.get_qubit_at(src)
        if qid is None:
            continue
        moved_qubits.append((qid, src, dst))
        new_config[qid] = dst

    new_occupied = frozenset(new_config.values()) | tree.blocked_locations

    for qid, src, dst in moved_qubits:
        if qid not in self.target:
            continue
        d_before = self._distance_to_target(src, self.target[qid], tree)
        d_after = self._distance_to_target(dst, self.target[qid], tree)
        distance_moved += max(0.0, d_before - d_after)
        if dst == self.target[qid]:
            arrived_gain += 1
        mobility_before += self._mobility_at(src, occupied, tree)
        mobility_after += self._mobility_at(dst, new_occupied, tree)

    mobility_gain = mobility_after - mobility_before

    return (
        self.params.alpha * distance_moved
        + self.params.beta * arrived_gain
        + self.params.gamma * mobility_gain
    )

ConfigurationNode dataclass

ConfigurationNode(
    configuration: dict[int, LocationAddress],
    parent: ConfigurationNode | None = None,
    parent_moves: frozenset[LaneAddress] | None = None,
    children: dict[
        frozenset[LaneAddress], ConfigurationNode
    ] = dict(),
    depth: int = 0,
)

A node in the configuration tree representing a valid atom placement.

Each node tracks which qubits are at which physical locations, how this configuration was reached (parent + moves), and what configurations are reachable from here (children).

children class-attribute instance-attribute

children: dict[
    frozenset[LaneAddress], ConfigurationNode
] = field(default_factory=dict)

Children keyed by the move set that produced them.

config_key cached property

config_key: Configuration

Canonical hashable key for this configuration.

Two nodes with the same atom placement (regardless of history) produce the same config_key.

configuration instance-attribute

configuration: dict[int, LocationAddress]

Mapping of qubit ID to current physical location.

depth class-attribute instance-attribute

depth: int = 0

Distance from the root node.

occupied_locations cached property

occupied_locations: frozenset[LocationAddress]

The set of physical locations currently occupied by atoms.

parent class-attribute instance-attribute

parent: ConfigurationNode | None = None

The parent node that produced this configuration, or None for the root.

parent_moves class-attribute instance-attribute

parent_moves: frozenset[LaneAddress] | None = None

The move set applied to the parent to produce this configuration.

get_qubit_at

get_qubit_at(location: LocationAddress) -> int | None

Return the qubit ID at a location, or None if empty.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/configuration.py

def get_qubit_at(self, location: LocationAddress) -> int | None:
    """Return the qubit ID at a location, or None if empty."""
    return self._qubit_at_location.get(location)

is_occupied

is_occupied(location: LocationAddress) -> bool

Check whether a physical location has an atom.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/configuration.py

def is_occupied(self, location: LocationAddress) -> bool:
    """Check whether a physical location has an atom."""
    return location in self._occupied_locations

path_to_root

path_to_root() -> list[frozenset[LaneAddress]]

Walk from this node to the root, returning move sets in root-to-leaf order.

The returned list has length equal to this node's depth. Each element is the frozenset of lane addresses applied at that step.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/configuration.py

def path_to_root(self) -> list[frozenset[LaneAddress]]:
    """Walk from this node to the root, returning move sets in root-to-leaf order.

    The returned list has length equal to this node's depth. Each element
    is the frozenset of lane addresses applied at that step.
    """
    moves: list[frozenset[LaneAddress]] = []
    node = self
    while node.parent is not None:
        assert node.parent_moves is not None
        moves.append(node.parent_moves)
        node = node.parent
    moves.reverse()
    return moves

to_move_program

to_move_program() -> tuple[tuple[LaneAddress, ...], ...]

Convert the path from root to this node into a move program.

Returns a tuple of tuples, where each inner tuple is a sorted sequence of lane addresses for one move step. Sorting is by encoded lane value for deterministic ordering.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/configuration.py

def to_move_program(self) -> tuple[tuple[LaneAddress, ...], ...]:
    """Convert the path from root to this node into a move program.

    Returns a tuple of tuples, where each inner tuple is a sorted
    sequence of lane addresses for one move step. Sorting is by
    encoded lane value for deterministic ordering.
    """
    return tuple(
        tuple(sorted(ms, key=lambda lane: lane.encode()))
        for ms in self.path_to_root()
    )

ConfigurationTree dataclass

ConfigurationTree(
    arch_spec: ArchSpec,
    root: ConfigurationNode,
    blocked_locations: frozenset[
        LocationAddress
    ] = frozenset(),
)

Tree that explores the space of valid atom configurations.

Starting from an initial placement, the tree manages the transposition table and validates move sets. Move generation and node expansion are delegated to MoveGenerator implementations.

NOTE: If deadlock density is high, consider refactoring to a DAG (directed acyclic graph) where nodes can have multiple parents. This enables backward propagation of deadlock information — when a subtree is exhausted, all parents are notified and can prune early. Currently the transposition table prevents re-expanding seen configurations, but does not propagate deadlock status upstream.

apply_move_set

apply_move_set(
    node: ConfigurationNode,
    move_set: frozenset[LaneAddress],
    strict: bool = True,
) -> ConfigurationNode | None

Apply a move set to a node, returning a new child or None.

Resolves lane endpoints, checks for collisions, and creates a child node if valid.

The move set is first validated against the arch spec (AOD geometry, consistency, bus membership). This check runs as an assert and is independent of strict — it signals an internal bug, not a recoverable runtime condition. It can be disabled with python -O.

Parameters:

Name	Type	Description	Default
node	ConfigurationNode	The node to apply moves to.	required
move_set	frozenset[LaneAddress]	The set of lane addresses to apply.	required
strict	bool	If True (default), raises InvalidMoveError when a move set causes a collision. If False, silently returns None for invalid moves.	True

Returns:

Type	Description
ConfigurationNode | None	A new ConfigurationNode, or None if:
ConfigurationNode | None	The move is invalid and strict=False
ConfigurationNode | None	The configuration was already reached via a different branch at equal-or-lesser depth (transposition table)

Raises:

Type	Description
AssertionError	If the move set fails lane-group validation (regardless of strict; disabled with python -O).
InvalidMoveError	If strict=True and the move set causes a collision or contains an invalid lane address.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def apply_move_set(
    self,
    node: ConfigurationNode,
    move_set: frozenset[LaneAddress],
    strict: bool = True,
) -> ConfigurationNode | None:
    """Apply a move set to a node, returning a new child or None.

    Resolves lane endpoints, checks for collisions, and creates
    a child node if valid.

    The move set is first validated against the arch spec (AOD
    geometry, consistency, bus membership).  This check runs as an
    ``assert`` and is independent of *strict* — it signals an
    internal bug, not a recoverable runtime condition.  It can be
    disabled with ``python -O``.

    Args:
        node: The node to apply moves to.
        move_set: The set of lane addresses to apply.
        strict: If True (default), raises InvalidMoveError when a
            move set causes a collision. If False, silently returns
            None for invalid moves.

    Returns:
        A new ConfigurationNode, or None if:
        - The move is invalid and strict=False
        - The configuration was already reached via a different
          branch at equal-or-lesser depth (transposition table)

    Raises:
        AssertionError: If the move set fails lane-group validation
            (regardless of *strict*; disabled with ``python -O``).
        InvalidMoveError: If strict=True and the move set causes a
            collision or contains an invalid lane address.
    """
    outcome = self.try_move_set(node, move_set, strict=strict)
    return (
        outcome.child if outcome.status == ExpansionStatus.CREATED_CHILD else None
    )

expand_node

expand_node(
    node: ConfigurationNode,
    generator: MoveGenerator,
    strict: bool = True,
) -> list[ConfigurationNode]

Expand a node using the given generator.

Generates candidate move sets, validates each (collision checks, transposition table), and creates child nodes. Nodes already seen at equal-or-lesser depth are skipped.

Parameters:

Name	Type	Description	Default
node	ConfigurationNode	The node to expand.	required
generator	MoveGenerator	Produces candidate move sets.	required
strict	bool	If True, raises on invalid moves. If False, skips them.	True

Returns:

Type	Description
list[ConfigurationNode]	List of newly created child nodes (may be empty).

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def expand_node(
    self,
    node: ConfigurationNode,
    generator: MoveGenerator,
    strict: bool = True,
) -> list[ConfigurationNode]:
    """Expand a node using the given generator.

    Generates candidate move sets, validates each (collision checks,
    transposition table), and creates child nodes. Nodes already
    seen at equal-or-lesser depth are skipped.

    Args:
        node: The node to expand.
        generator: Produces candidate move sets.
        strict: If True, raises on invalid moves. If False, skips them.

    Returns:
        List of newly created child nodes (may be empty).
    """
    children: list[ConfigurationNode] = []
    for outcome in self.expand_node_detailed(node, generator, strict=strict):
        if outcome.status == ExpansionStatus.CREATED_CHILD:
            assert outcome.child is not None
            children.append(outcome.child)
    return children

expand_node_detailed

expand_node_detailed(
    node: ConfigurationNode,
    generator: MoveGenerator,
    strict: bool = True,
) -> Iterator[ExpansionOutcome]

Expand a node and yield one outcome per generated candidate.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def expand_node_detailed(
    self,
    node: ConfigurationNode,
    generator: MoveGenerator,
    strict: bool = True,
) -> Iterator[ExpansionOutcome]:
    """Expand a node and yield one outcome per generated candidate."""
    for move_set in generator.generate(node, self):
        yield self.try_move_set(node, move_set, strict=strict)

from_initial_placement classmethod

from_initial_placement(
    arch_spec: ArchSpec,
    placement: dict[int, LocationAddress],
    blocked_locations: frozenset[
        LocationAddress
    ] = frozenset(),
) -> ConfigurationTree

Create a tree from an initial qubit placement.

Parameters:

Name	Type	Description	Default
arch_spec	ArchSpec	Architecture specification for lane validation.	required
placement	dict[int, LocationAddress]	Mapping of qubit IDs to their initial locations.	required

Returns:

Type	Description
ConfigurationTree	A new ConfigurationTree rooted at the given placement.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

@classmethod
def from_initial_placement(
    cls,
    arch_spec: ArchSpec,
    placement: dict[int, LocationAddress],
    blocked_locations: frozenset[LocationAddress] = frozenset(),
) -> ConfigurationTree:
    """Create a tree from an initial qubit placement.

    Args:
        arch_spec: Architecture specification for lane validation.
        placement: Mapping of qubit IDs to their initial locations.

    Returns:
        A new ConfigurationTree rooted at the given placement.
    """
    root = ConfigurationNode(configuration=dict(placement))
    return cls(
        arch_spec=arch_spec,
        root=root,
        blocked_locations=frozenset(blocked_locations),
    )

lane_for_source

lane_for_source(
    move_type: MoveType,
    bus_id: int,
    direction: Direction,
    source: LocationAddress,
) -> LaneAddress | None

Resolve one lane by source for a specific triplet.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def lane_for_source(
    self,
    move_type: MoveType,
    bus_id: int,
    direction: Direction,
    source: LocationAddress,
) -> LaneAddress | None:
    """Resolve one lane by source for a specific triplet."""
    return self._lane_by_src[(move_type, bus_id, direction)].get(source)

lanes_for

lanes_for(
    move_type: MoveType, bus_id: int, direction: Direction
) -> Iterator[LaneAddress]

Yield all lane addresses for a specific (move_type, bus_id, direction).

Parameters:

Name	Type	Description	Default
move_type	MoveType	The move type (SITE or WORD).	required
bus_id	int	The bus index.	required
direction	Direction	The direction (FORWARD or BACKWARD).	required

Yields:

Type	Description
LaneAddress	LaneAddress values.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def lanes_for(
    self,
    move_type: MoveType,
    bus_id: int,
    direction: Direction,
) -> Iterator[LaneAddress]:
    """Yield all lane addresses for a specific (move_type, bus_id, direction).

    Args:
        move_type: The move type (SITE or WORD).
        bus_id: The bus index.
        direction: The direction (FORWARD or BACKWARD).

    Yields:
        LaneAddress values.
    """
    yield from self._lanes_by_triplet[(move_type, bus_id, direction)]

outgoing_lanes

outgoing_lanes(
    source: LocationAddress,
) -> tuple[LaneAddress, ...]

Return all precomputed outgoing lanes from source.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def outgoing_lanes(self, source: LocationAddress) -> tuple[LaneAddress, ...]:
    """Return all precomputed outgoing lanes from source."""
    return self._outgoing_lanes_by_src.get(source, ())

try_move_set

try_move_set(
    node: ConfigurationNode,
    move_set: frozenset[LaneAddress],
    strict: bool = True,
) -> ExpansionOutcome

Attempt one move set and return a detailed outcome.

In strict mode, invalid lanes and collisions raise InvalidMoveError (matching apply_move_set behavior). In non-strict mode, they are returned as INVALID_LANE/COLLISION outcomes.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def try_move_set(
    self,
    node: ConfigurationNode,
    move_set: frozenset[LaneAddress],
    strict: bool = True,
) -> ExpansionOutcome:
    """Attempt one move set and return a detailed outcome.

    In strict mode, invalid lanes and collisions raise InvalidMoveError
    (matching apply_move_set behavior). In non-strict mode, they are
    returned as INVALID_LANE/COLLISION outcomes.
    """
    lane_errors = self.arch_spec.check_lane_group(list(move_set))
    if lane_errors:
        msg = f"Move set failed lane-group validation: {'; '.join(str(e) for e in lane_errors)}"
        if strict:
            raise InvalidMoveError(msg)
        return ExpansionOutcome(
            move_set=move_set,
            status=ExpansionStatus.INVALID_LANE,
            error_message=msg,
        )

    existing_child = node.children.get(move_set)
    if existing_child is not None:
        return ExpansionOutcome(
            move_set=move_set,
            status=ExpansionStatus.ALREADY_CHILD,
            child=existing_child,
        )

    new_config = dict(node.configuration)
    occupied = node.occupied_locations
    blocked = self.blocked_locations

    for lane in move_set:
        try:
            src, dst = self.arch_spec.get_endpoints(lane)
        except Exception as e:
            msg = f"Invalid lane address {lane!r}: {e}"
            if strict:
                raise InvalidMoveError(msg) from e
            return ExpansionOutcome(
                move_set=move_set,
                status=ExpansionStatus.INVALID_LANE,
                error_message=msg,
            )

        qid = node.get_qubit_at(src)
        if qid is None:
            continue

        # Bus src and dst are disjoint sets, so within a single-bus
        # move set, two sources cannot map to the same destination.
        # We only need to check against stationary atoms.
        if dst in occupied or dst in blocked:
            blocker = node.get_qubit_at(dst)
            blocker_text = (
                f"qubit {blocker}" if blocker is not None else "a blocked location"
            )
            msg = (
                f"Collision: qubit {qid} moving to {dst!r} "
                f"which is occupied by {blocker_text}"
            )
            if strict:
                raise InvalidMoveError(msg)
            return ExpansionOutcome(
                move_set=move_set,
                status=ExpansionStatus.COLLISION,
                error_message=msg,
            )

        new_config[qid] = dst

    # Check transposition table
    new_node = ConfigurationNode(
        configuration=new_config,
        parent=node,
        parent_moves=move_set,
        depth=node.depth + 1,
    )
    key = new_node.config_key

    if key in self.seen:
        existing = self.seen[key]
        if existing.depth <= new_node.depth:
            return ExpansionOutcome(
                move_set=move_set,
                status=ExpansionStatus.TRANSPOSITION_SEEN,
                existing_node=existing,
            )

    # Register in transposition table and add as child
    self.seen[key] = new_node
    node.children[move_set] = new_node
    return ExpansionOutcome(
        move_set=move_set,
        status=ExpansionStatus.CREATED_CHILD,
        child=new_node,
    )

valid_lanes

valid_lanes(
    node: ConfigurationNode,
    move_type: MoveType | None = None,
    bus_id: int | None = None,
    direction: Direction | None = None,
) -> Iterator[LaneAddress]

Yield valid individual lane addresses from a configuration.

A lane is valid if its source is occupied and its destination is not occupied. Optionally filter by move_type, bus_id, and direction — None means include all.

Parameters:

Name	Type	Description	Default
node	ConfigurationNode	The configuration to query.	required
move_type	MoveType | None	Filter to this move type, or None for all.	None
bus_id	int | None	Filter to this bus ID, or None for all.	None
direction	Direction | None	Filter to this direction, or None for all.	None

Yields:

Type	Description
LaneAddress	Valid LaneAddress values.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/tree.py

def valid_lanes(
    self,
    node: ConfigurationNode,
    move_type: MoveType | None = None,
    bus_id: int | None = None,
    direction: Direction | None = None,
) -> Iterator[LaneAddress]:
    """Yield valid individual lane addresses from a configuration.

    A lane is valid if its source is occupied and its destination
    is not occupied. Optionally filter by move_type, bus_id, and
    direction — None means include all.

    Args:
        node: The configuration to query.
        move_type: Filter to this move type, or None for all.
        bus_id: Filter to this bus ID, or None for all.
        direction: Filter to this direction, or None for all.

    Yields:
        Valid LaneAddress values.
    """
    occupied = node.occupied_locations
    blocked = self.blocked_locations

    move_types = (
        [move_type] if move_type is not None else [MoveType.SITE, MoveType.WORD]
    )
    directions = (
        [direction]
        if direction is not None
        else [Direction.FORWARD, Direction.BACKWARD]
    )

    for mt in move_types:
        buses = (
            self.arch_spec.site_buses
            if mt == MoveType.SITE
            else self.arch_spec.word_buses
        )
        bus_ids = [bus_id] if bus_id is not None else list(range(len(buses)))

        for bid in bus_ids:
            for d in directions:
                for lane in self.lanes_for(mt, bid, d):
                    src, dst = self.arch_spec.get_endpoints(lane)
                    if (
                        src in occupied
                        and dst not in occupied
                        and dst not in blocked
                    ):
                        yield lane

EntropyGuidedSearch

EntropyGuidedSearch(
    tree: ConfigurationTree,
    target: dict[int, LocationAddress],
    goal: GoalPredicate,
    params: SearchParams = SearchParams(),
    max_depth: int | None = None,
    max_expansions: int | None = None,
    on_step: (
        Callable[[str, ConfigurationNode, StepInfo], None]
        | None
    ) = None,
)

Entropy-guided depth-first search with reversion and sequential fallback.

Encapsulates the internal state of the search so that helper methods can access shared context (tree, target, params, etc.) without requiring long parameter lists.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/entropy_guided.py

def __init__(
    self,
    tree: ConfigurationTree,
    target: dict[int, LocationAddress],
    goal: GoalPredicate,
    params: SearchParams = SearchParams(),
    max_depth: int | None = None,
    max_expansions: int | None = None,
    on_step: Callable[[str, ConfigurationNode, StepInfo], None] | None = None,
) -> None:
    self.tree = tree
    self.target = target
    self.goal = goal
    self.params = params
    self.max_depth = max_depth
    self.max_expansions = max_expansions

    self.scorer = CandidateScorer(params=params, target=target)
    self.search_nodes: dict[int, _SearchNode] = {}
    self.debug = _DebugEmitter(scorer=self.scorer, tree=tree, on_step=on_step)
    self.nodes_expanded = 0
    self.max_depth_reached = 0

run

run(generator: MoveGenerator) -> SearchResult

Execute the entropy-guided search and return the result.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/entropy_guided.py

def run(self, generator: MoveGenerator) -> SearchResult:
    """Execute the entropy-guided search and return the result."""
    found_goals: list[ConfigurationNode] = []
    if self.goal(self.tree.root):
        self.debug.goal(self.tree.root, entropy=0)
        return SearchResult(
            nodes_expanded=0,
            max_depth_reached=0,
            goal_nodes=(self.tree.root,),
        )

    current_node = self.tree.root

    while self.max_expansions is None or self.nodes_expanded < self.max_expansions:
        sn = self._get_or_create_search_node(current_node)
        if self.max_depth is not None and current_node.depth >= self.max_depth:
            sn.force_entropy(self.params.e_max)

        if sn.entropy >= self.params.e_max:
            ancestor = current_node
            for _ in range(self.params.reversion_steps):
                if ancestor.parent is None:
                    break
                ancestor = ancestor.parent

            ancestor_sn = self._get_or_create_search_node(ancestor)
            if ancestor.parent is None and ancestor_sn.entropy >= self.params.e_max:
                fallback_result = self._sequential_fallback(self.tree.root)
                combined_goals = found_goals + list(fallback_result.goal_nodes)
                return self._make_result(goal_nodes=tuple(combined_goals))

            ancestor_sn.bump_entropy(self.params.delta_e)
            self.debug.revert(ancestor, ancestor_sn, sn)
            current_node = ancestor
            continue

        candidate = self._get_next_candidate(sn, current_node, generator)

        if candidate is None:
            no_valid_qid = self._first_unresolved_qubit_without_valid_move(
                current_node
            )
            sn.bump_entropy(
                self.params.delta_e,
                "no-valid-moves",
                no_valid_moves_qubit=no_valid_qid,
            )
            self.debug.entropy_bump(current_node, sn)
            continue

        outcome = self.tree.try_move_set(current_node, candidate, strict=False)
        sn.candidates_tried += 1

        if outcome.status != ExpansionStatus.CREATED_CHILD:
            if outcome.status == ExpansionStatus.TRANSPOSITION_SEEN:
                reason = "state-seen"
                seen_node_id = (
                    id(outcome.existing_node)
                    if outcome.existing_node is not None
                    else None
                )
                no_valid_qid = None
            else:
                reason = "no-valid-moves"
                seen_node_id = None
                no_valid_qid = self._first_unresolved_qubit_without_valid_move(
                    current_node
                )

            sn.bump_entropy(
                self.params.delta_e,
                reason,
                no_valid_moves_qubit=no_valid_qid,
                state_seen_node_id=seen_node_id,
            )
            self.debug.entropy_bump(current_node, sn)
            continue
        assert outcome.child is not None
        child = outcome.child

        self.nodes_expanded += 1
        self.max_depth_reached = max(self.max_depth_reached, child.depth)
        self.debug.descend(child, sn, candidate, current_node)

        if self.goal(child):
            self.debug.goal(child, entropy=sn.entropy)
            found_goals.append(child)
            if len(found_goals) >= self.params.max_goal_candidates:
                return self._make_result(goal_nodes=tuple(found_goals))
            current_node = self._cutoff_ancestor(child)
            continue

        current_node = child

    return self._make_result(goal_nodes=tuple(found_goals))

EntropyGuidedTraversal dataclass

EntropyGuidedTraversal(
    target: dict[int, LocationAddress],
    params: SearchParams = SearchParams(),
    on_step: (
        Callable[[str, ConfigurationNode, StepInfo], None]
        | None
    ) = None,
)

Bases: TraversalStrategyABC

Traversal strategy adapter for entropy-guided search.

search

search(
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None
) -> SearchResult

Execute search over the tree using the supplied generator.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/entropy_guided.py

def search(
    self,
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None,
) -> SearchResult:
    search = EntropyGuidedSearch(
        tree=tree,
        target=self.target,
        goal=goal,
        params=self.params,
        max_depth=max_depth,
        max_expansions=max_expansions,
        on_step=self.on_step,
    )
    if isinstance(generator, HeuristicMoveGenerator):
        entropy_view: dict[int, EntropyNode] = search.search_nodes  # type: ignore[assignment]
        generator.search_nodes = entropy_view
    return search.run(generator=generator)

ExhaustiveMoveGenerator dataclass

ExhaustiveMoveGenerator(
    max_x_capacity: int | None = None,
    max_y_capacity: int | None = None,
)

Enumerates all valid AOD grids from the configuration.

For each (move_type, bus_id, direction) group, finds all source positions, enumerates grid subsets within AOD capacity, and yields the full grid of lane addresses.

Pre-filters grids where an occupied source has an occupied destination (collision) as an optimization.

NOTE for Rust port: replace itertools.combinations with Gosper's hack for bitmask enumeration of exactly-k-set-bits subsets. This avoids iterating all 2^n masks when capacity is small. See #298.

max_x_capacity class-attribute instance-attribute

max_x_capacity: int | None = None

Maximum number of unique X positions the AOD can address.

max_y_capacity class-attribute instance-attribute

max_y_capacity: int | None = None

Maximum number of unique Y positions the AOD can address.

GreedyBestFirstTraversal dataclass

GreedyBestFirstTraversal(heuristic: HeuristicFunction)

Bases: TraversalStrategyABC

Greedy best-first traversal policy.

search

search(
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None
) -> SearchResult

Execute search over the tree using the supplied generator.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/greedy.py

def search(
    self,
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None,
) -> SearchResult:
    _ = max_depth
    frontier: list[PriorityEntry] = []
    heapq.heappush(frontier, PriorityEntry(self.heuristic(tree.root), tree.root))
    return run_frontier_search(
        tree=tree,
        generator=generator,
        goal=goal,
        pop_next=lambda: heapq.heappop(frontier).node if frontier else None,
        push_child=lambda child: heapq.heappush(
            frontier, PriorityEntry(self.heuristic(child), child)
        ),
        frontier_has_items=lambda: bool(frontier),
        max_expansions=max_expansions,
        max_depth=None,
        goal_on_pop=False,
    )

GreedyMoveGenerator dataclass

GreedyMoveGenerator(target: dict[int, LocationAddress])

Greedy shortest-path move generator.

For each unresolved qubit, computes the shortest path to its target while routing around all other atoms (but not itself) and takes the first lane. Those first-step lanes are grouped by (move_type, bus_id, direction) and partitioned into AOD-compatible rectangular grids via BusContext.

target instance-attribute

target: dict[int, LocationAddress]

Mapping of qubit ID to target location.

generate

generate(
    node: ConfigurationNode, tree: ConfigurationTree
) -> Iterator[frozenset[LaneAddress]]

Yield candidate move sets.

Steps: 1. For each qubit, find shortest path to target routing around all other atoms (but not itself). 2. Take the first lane from each path. 3. Group by (move_type, bus_id, direction). 4. Build AOD-compatible rectangular grids per group.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/generators/greedy.py

def generate(
    self,
    node: ConfigurationNode,
    tree: ConfigurationTree,
) -> Iterator[frozenset[LaneAddress]]:
    """Yield candidate move sets.

    Steps:
    1. For each qubit, find shortest path to target routing around
       all other atoms (but not itself).
    2. Take the first lane from each path.
    3. Group by (move_type, bus_id, direction).
    4. Build AOD-compatible rectangular grids per group.
    """
    occupied = node.occupied_locations | tree.blocked_locations

    first_lanes: dict[int, LaneAddress] = {}
    for qid, current in node.configuration.items():
        target_loc = self.target.get(qid)
        if target_loc is None or current == target_loc:
            continue

        result = tree.path_finder.find_path(
            current, target_loc, occupied=occupied - {current}
        )
        if result is None:
            continue

        lanes, _ = result
        if lanes:
            first_lanes[qid] = lanes[0]

    if not first_lanes:
        return

    groups: dict[
        tuple[MoveType, int, Direction],
        dict[int, LaneAddress],
    ] = {}
    for qid, lane in first_lanes.items():
        key = (lane.move_type, lane.bus_id, lane.direction)
        groups.setdefault(key, {})[qid] = lane

    for (mt, bid, d), entries in groups.items():
        ctx = BusContext.from_tree(tree, occupied, mt, bid, d)
        yield from ctx.build_aod_grids(entries)

HeuristicMoveGenerator dataclass

HeuristicMoveGenerator(
    scorer: CandidateScorer,
    params: SearchParams,
    search_nodes: dict[int, EntropyNode],
)

Generates ranked candidate movesets using entropy-weighted scoring.

Implements the MoveGenerator protocol. The traversal sets search_nodes before the search loop begins so the generator can read per-node entropy.

search_nodes instance-attribute

search_nodes: dict[int, EntropyNode]

Mapping of id(ConfigurationNode) -> entropy metadata node.

generate

generate(
    node: ConfigurationNode, tree: ConfigurationTree
) -> Iterator[frozenset[LaneAddress]]

Yield candidate movesets ranked by moveset score (descending).

Steps: 1. Get entropy from traversal metadata node. 2. Score all (qubit, move_type, bus_id, direction) pairs. 3. Per-qubit: keep top C triples. 4. Group by (move_type, bus_id, direction), collect positive-score qubits. 5. Resolve conflicts within each group (greedy by score). 6. Build moveset per group, score with score_moveset, sort descending.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/generators/heuristic.py

def generate(
    self,
    node: ConfigurationNode,
    tree: ConfigurationTree,
) -> Iterator[frozenset[LaneAddress]]:
    """Yield candidate movesets ranked by moveset score (descending).

    Steps:
    1. Get entropy from traversal metadata node.
    2. Score all (qubit, move_type, bus_id, direction) pairs.
    3. Per-qubit: keep top C triples.
    4. Group by (move_type, bus_id, direction), collect positive-score qubits.
    5. Resolve conflicts within each group (greedy by score).
    6. Build moveset per group, score with score_moveset, sort descending.
    """
    search_node = self.search_nodes.get(id(node))
    entropy = search_node.entropy if search_node is not None else 1

    scores = self.scorer.score_all_qubit_bus_pairs(node, entropy, tree)
    if not scores:
        return

    # Per-qubit: keep top C (move_type, bus_id, direction) triples
    qubit_top: dict[int, list[tuple[MoveType, int, Direction, float]]] = {}
    for (qid, mt, bid, d), score in scores.items():
        qubit_top.setdefault(qid, []).append((mt, bid, d, score))

    for qid in qubit_top:
        qubit_top[qid].sort(key=lambda x: x[3], reverse=True)
        qubit_top[qid] = qubit_top[qid][: self.params.top_c]

    # Group by (move_type, bus_id, direction): collect positive-scoring qubits
    groups: dict[
        tuple[MoveType, int, Direction],
        list[tuple[int, float]],
    ] = {}
    for qid, top_triples in qubit_top.items():
        for mt, bid, d, score in top_triples:
            if score > 0:
                groups.setdefault((mt, bid, d), []).append((qid, score))

    # Fallback: if no group has positive-scoring qubits, use best single entry
    if not groups:
        best_key = max(scores, key=lambda key: scores[key])
        qid, mt, bid, d = best_key
        groups[(mt, bid, d)] = [(qid, scores[best_key])]

    # Build one moveset per group with conflict resolution
    occupied = node.occupied_locations | tree.blocked_locations
    candidates: list[tuple[float, frozenset[LaneAddress]]] = []

    for (mt, bid, d), qubit_scores in groups.items():
        # Sort qubits by score descending for greedy conflict resolution
        qubit_scores.sort(key=lambda x: x[1], reverse=True)

        lanes: list[LaneAddress] = []
        used_dsts: set[LocationAddress] = set()

        for qid, _ in qubit_scores:
            loc = node.configuration[qid]
            lane = tree.lane_for_source(mt, bid, d, loc)
            if lane is None:
                continue
            _, dst = tree.arch_spec.get_endpoints(lane)
            if dst in occupied:
                continue
            # Destination conflict: another qubit in this group targets same dst
            if dst in used_dsts:
                continue
            # Lane compatibility check with existing lanes in this group
            if lanes and not all(
                tree.arch_spec.compatible_lanes(lane, existing)
                for existing in lanes
            ):
                continue
            lanes.append(lane)
            used_dsts.add(dst)

        if not lanes:
            continue

        moveset = frozenset(lanes)
        ms_score = self.scorer.score_moveset(moveset, node, tree)
        candidates.append((ms_score, moveset))

    # Sort by moveset score descending
    candidates.sort(key=lambda x: x[0], reverse=True)

    for _, moveset in candidates:
        yield moveset

InvalidMoveError

Bases: Exception

Raised when a generator produces an invalid move set in strict mode.

MoveGenerator

Bases: Protocol

Interface for generating candidate move sets from a configuration.

Implementations yield candidate move sets. Validation and node creation are handled by ConfigurationTree.expand_node.

generate

generate(
    node: ConfigurationNode, tree: ConfigurationTree
) -> Iterator[frozenset[LaneAddress]]

Yield candidate move sets from the given configuration.

Parameters:

Name	Type	Description	Default
node	ConfigurationNode	The configuration to generate moves from.	required
tree	ConfigurationTree	The configuration tree (provides arch_spec, path_finder).	required

Yields:

Type	Description
frozenset[LaneAddress]	frozenset[LaneAddress] — each candidate parallel move set.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/generators/base.py

def generate(
    self,
    node: ConfigurationNode,
    tree: ConfigurationTree,
) -> Iterator[frozenset[LaneAddress]]:
    """Yield candidate move sets from the given configuration.

    Args:
        node: The configuration to generate moves from.
        tree: The configuration tree (provides arch_spec, path_finder).

    Yields:
        frozenset[LaneAddress] — each candidate parallel move set.
    """
    ...

SearchParams dataclass

SearchParams(
    w_d: float = 1.0,
    w_m: float = 0.1,
    alpha: float = 1.0,
    beta: float = 2.0,
    gamma: float = 0.5,
    top_c: int = 3,
    max_candidates: int = 2,
    reversion_steps: int = 1,
    delta_e: int = 1,
    e_max: int = 4,
    max_goal_candidates: int = 1,
)

Tunable parameters for entropy-guided search.

Attributes:

Name	Type	Description
w_d	float	Distance weight in per-qubit-bus scoring.
w_m	float	Mobility weight in per-qubit-bus scoring.
alpha	float	Distance weight in moveset scoring.
beta	float	Arrived-gain weight in moveset scoring.
gamma	float	Mobility weight in moveset scoring.
top_c	int	Top candidate (move_type, bus_id, direction) triples per qubit.
max_candidates	int	Candidates to try per entropy level before regenerating.
reversion_steps	int	Steps to revert up the tree on deadlock.
delta_e	int	Entropy increment per revisit or failed generation.
e_max	int	Entropy threshold that triggers reversion.
max_goal_candidates	int	Number of goal nodes to collect before stopping.

SearchResult dataclass

SearchResult(
    *,
    nodes_expanded: int,
    max_depth_reached: int,
    goal_nodes: tuple[ConfigurationNode, ...] = (),
    goal_node: ConfigurationNode | None = None
)

Result of a search strategy.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/goal.py

def __init__(
    self,
    *,
    nodes_expanded: int,
    max_depth_reached: int,
    goal_nodes: tuple[ConfigurationNode, ...] = (),
    goal_node: ConfigurationNode | None = None,
) -> None:
    # Canonicalize all results into goal_nodes so single/multi-solution
    # handling uses one representation everywhere.
    if goal_node is not None:
        if goal_nodes and goal_nodes[0] is not goal_node:
            raise ValueError(
                "goal_node must match first item in goal_nodes when both are set"
            )
        goal_nodes = (goal_node, *goal_nodes[1:]) if goal_nodes else (goal_node,)

    self.nodes_expanded = nodes_expanded
    self.max_depth_reached = max_depth_reached
    self.goal_nodes = goal_nodes

goal_node property

goal_node: ConfigurationNode | None

Best goal node, or None if no goal was found.

goal_nodes instance-attribute

goal_nodes: tuple[ConfigurationNode, ...] = goal_nodes

Goal nodes found during search, ordered best to worst.

max_depth_reached instance-attribute

max_depth_reached: int = max_depth_reached

Maximum depth reached during search.

nodes_expanded instance-attribute

nodes_expanded: int = nodes_expanded

Total number of nodes expanded during search.

TraversalStrategyABC

Bases: ABC

Strategy interface for traversing a configuration tree.

Traversal strategies choose which frontier node to examine next. Candidate move generation is delegated to the provided MoveGenerator.

search abstractmethod

search(
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None
) -> SearchResult

Execute search over the tree using the supplied generator.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/interface.py

@abc.abstractmethod
def search(
    self,
    *,
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    max_expansions: int | None = None,
    max_depth: int | None = None,
) -> SearchResult:
    """Execute search over the tree using the supplied generator."""
    ...

entropy_guided_search

entropy_guided_search(
    tree: ConfigurationTree,
    target: dict[int, LocationAddress],
    goal: GoalPredicate,
    params: SearchParams = SearchParams(),
    max_depth: int | None = None,
    max_expansions: int | None = None,
    on_step: (
        Callable[[str, ConfigurationNode, StepInfo], None]
        | None
    ) = None,
) -> SearchResult

Entropy-guided depth-first search with reversion and sequential fallback.

Thin wrapper around EntropyGuidedSearch for backward compatibility. Also takes target directly (in addition to goal) because scoring needs the raw target mapping.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/entropy_guided.py

def entropy_guided_search(
    tree: ConfigurationTree,
    target: dict[int, LocationAddress],
    goal: GoalPredicate,
    params: SearchParams = SearchParams(),
    max_depth: int | None = None,
    max_expansions: int | None = None,
    on_step: Callable[[str, ConfigurationNode, StepInfo], None] | None = None,
) -> SearchResult:
    """Entropy-guided depth-first search with reversion and sequential fallback.

    Thin wrapper around ``EntropyGuidedSearch`` for backward compatibility.
    Also takes *target* directly (in addition to *goal*) because scoring needs
    the raw target mapping.
    """
    scorer = CandidateScorer(params=params, target=target)
    search = EntropyGuidedSearch(
        tree=tree,
        target=target,
        goal=goal,
        params=params,
        max_depth=max_depth,
        max_expansions=max_expansions,
        on_step=on_step,
    )
    entropy_view: dict[int, EntropyNode] = search.search_nodes  # type: ignore[assignment]
    generator = HeuristicMoveGenerator(
        scorer=scorer,
        params=params,
        search_nodes=entropy_view,
    )
    return search.run(generator=generator)

greedy_best_first

greedy_best_first(
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    heuristic: HeuristicFunction,
    max_expansions: int | None = None,
) -> SearchResult

Greedy best-first search using heuristic only.

Expands the node with the lowest heuristic value first. Does not consider accumulated path cost — purely greedy. Fast but not guaranteed to find the optimal (shortest) solution.

Parameters:

Name	Type	Description	Default
tree	ConfigurationTree	The configuration tree to search.	required
generator	MoveGenerator	Move generator for producing candidates.	required
goal	GoalPredicate	Predicate that returns True for goal configurations.	required
heuristic	HeuristicFunction	Estimates cost to goal. Lower is better.	required
max_expansions	int | None	Maximum number of nodes to expand before giving up. None means no limit.	None

Returns:

Type	Description
SearchResult	SearchResult with the goal node (or None if not found).

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/greedy.py

def greedy_best_first(
    tree: ConfigurationTree,
    generator: MoveGenerator,
    goal: GoalPredicate,
    heuristic: HeuristicFunction,
    max_expansions: int | None = None,
) -> SearchResult:
    """Greedy best-first search using heuristic only.

    Expands the node with the lowest heuristic value first. Does not
    consider accumulated path cost — purely greedy. Fast but not
    guaranteed to find the optimal (shortest) solution.

    Args:
        tree: The configuration tree to search.
        generator: Move generator for producing candidates.
        goal: Predicate that returns True for goal configurations.
        heuristic: Estimates cost to goal. Lower is better.
        max_expansions: Maximum number of nodes to expand before
            giving up. None means no limit.

    Returns:
        SearchResult with the goal node (or None if not found).
    """
    return GreedyBestFirstTraversal(heuristic=heuristic).search(
        tree=tree,
        generator=generator,
        goal=goal,
        max_expansions=max_expansions,
    )

partial_placement_goal

partial_placement_goal(
    target: dict[int, LocationAddress],
    min_placed: int | None = None,
) -> GoalPredicate

Goal: at least some qubits are at their target locations.

If min_placed is None, all qubits in target must be placed (same as placement_goal). Otherwise, at least min_placed qubits must be at their target.

Parameters:

Name	Type	Description	Default
target	dict[int, LocationAddress]	Mapping of qubit ID to desired location.	required
min_placed	int | None	Minimum number of qubits that must be at their target. None means all.	None

Returns:

Type	Description
GoalPredicate	A GoalPredicate.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/goal.py

def partial_placement_goal(
    target: dict[int, LocationAddress],
    min_placed: int | None = None,
) -> GoalPredicate:
    """Goal: at least some qubits are at their target locations.

    If `min_placed` is None, all qubits in `target` must be placed
    (same as `placement_goal`). Otherwise, at least `min_placed`
    qubits must be at their target.

    Args:
        target: Mapping of qubit ID to desired location.
        min_placed: Minimum number of qubits that must be at their
            target. None means all.

    Returns:
        A GoalPredicate.
    """
    required = min_placed if min_placed is not None else len(target)

    def goal(node: ConfigurationNode) -> bool:
        placed = sum(
            1 for qid, loc in target.items() if node.configuration.get(qid) == loc
        )
        return placed >= required

    return goal

placement_goal

placement_goal(
    target: dict[int, LocationAddress],
) -> GoalPredicate

Goal: all specified qubits are at their target locations.

Every qubit in target must be at the exact location. Qubits not in target are ignored.

Parameters:

Name	Type	Description	Default
target	dict[int, LocationAddress]	Mapping of qubit ID to desired location.	required

Returns:

Type	Description
GoalPredicate	A GoalPredicate that returns True when all targets are met.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/goal.py

def placement_goal(target: dict[int, LocationAddress]) -> GoalPredicate:
    """Goal: all specified qubits are at their target locations.

    Every qubit in `target` must be at the exact location. Qubits not
    in `target` are ignored.

    Args:
        target: Mapping of qubit ID to desired location.

    Returns:
        A GoalPredicate that returns True when all targets are met.
    """

    def goal(node: ConfigurationNode) -> bool:
        return all(node.configuration.get(qid) == loc for qid, loc in target.items())

    return goal

zone_goal

zone_goal(
    zone_id: int, arch_spec: ArchSpec
) -> GoalPredicate

Goal: all qubits are located in the specified zone.

A qubit is "in the zone" if its location's word_id is in the zone's word list.

Parameters:

Name	Type	Description	Default
zone_id	int	The zone to target.	required
arch_spec	ArchSpec	Architecture specification (for zone → word mapping).	required

Returns:

Type	Description
GoalPredicate	A GoalPredicate that returns True when all qubits are in the zone.

Source code in .venv/lib/python3.12/site-packages/bloqade/lanes/search/traversal/goal.py

def zone_goal(zone_id: int, arch_spec: ArchSpec) -> GoalPredicate:
    """Goal: all qubits are located in the specified zone.

    A qubit is "in the zone" if its location's word_id is in the
    zone's word list.

    Args:
        zone_id: The zone to target.
        arch_spec: Architecture specification (for zone → word mapping).

    Returns:
        A GoalPredicate that returns True when all qubits are in the zone.
    """
    zone_words = set(arch_spec.zones[zone_id])

    def goal(node: ConfigurationNode) -> bool:
        return all(loc.word_id in zone_words for loc in node.configuration.values())

    return goal