API Reference

This page provides detailed information about the Locus Python API.

Core Interface

The primary entry point is the Detector class. It can be initialized with various configuration parameters passed as keyword arguments.

import locus

# Initialize with keyword arguments
detector = locus.Detector(
    families=[locus.TagFamily.AprilTag36h11, locus.TagFamily.ArUco4x4_50],
    decimation=2,
    threads=4,
)

batch = detector.detect(img)

locus.Detector

High-level detector.

Construction

Detector(profile="standard") — load a shipped JSON profile by name. Detector(config=my_cfg) — use a pre-built :class:DetectorConfig. Detector() — equivalent to profile="standard".

Per-call orchestration options (decimation, threads, families, max_concurrent_frames) stay outside the profile because they describe how the detector is invoked, not what it looks for.

threads sets the Rayon worker count. 0 or None (the default) uses the global Rayon pool (RAYON_NUM_THREADS / core count); n > 0 builds one scoped n-thread pool at construction and runs every detect / detect_concurrent call inside it, bounding this detector's CPU footprint. Detection output is identical for every value.

max_concurrent_frames sizes the pool of frame contexts that :meth:detect_concurrent leases (default 1); frames beyond it get a temporary context each.

Source code in crates/locus-py/locus/__init__.py

class Detector:
    """High-level detector.

    Construction:
        ``Detector(profile="standard")`` — load a shipped JSON profile by name.
        ``Detector(config=my_cfg)`` — use a pre-built :class:`DetectorConfig`.
        ``Detector()`` — equivalent to ``profile="standard"``.

    Per-call orchestration options (``decimation``, ``threads``, ``families``,
    ``max_concurrent_frames``) stay outside the profile because they describe
    *how* the detector is invoked, not *what* it looks for.

    ``threads`` sets the Rayon worker count. ``0`` or ``None`` (the default)
    uses the global Rayon pool (``RAYON_NUM_THREADS`` / core count); ``n > 0``
    builds one scoped ``n``-thread pool at construction and runs every
    ``detect`` / ``detect_concurrent`` call inside it, bounding this detector's
    CPU footprint. Detection output is identical for every value.

    ``max_concurrent_frames`` sizes the pool of frame contexts that
    :meth:`detect_concurrent` leases (default ``1``); frames beyond it get a
    temporary context each.
    """

    def __init__(
        self,
        profile: ProfileName | None = None,
        config: DetectorConfig | None = None,
        *,
        decimation: int | None = None,
        threads: int | None = None,
        families: list[TagFamily] | None = None,
        max_concurrent_frames: int | None = None,
    ) -> None:
        if profile is not None and config is not None:
            raise ValueError("Pass either `profile` or `config`, not both.")

        if config is None:
            config = DetectorConfig.from_profile(profile or "standard")

        if families is None:
            families = [TagFamily.AprilTag36h11]

        self._inner = _create_detector_from_config(
            config_json=config.model_dump_json(),
            decimation=decimation,
            threads=threads,
            families=[int(f) for f in families],
            max_concurrent_frames=max_concurrent_frames,
        )

    def config(self) -> DetectorConfig:
        """Returns the current detector configuration as a nested model.

        Rust serializes its live config into the profile-JSON format and Python
        re-parses it, so the readback is total over every field — no per-field
        transcription to drift out of sync.
        """
        return DetectorConfig.model_validate_json(self._inner.config())

    def set_families(self, families: list[TagFamily]):
        """Update the tag families to be detected."""
        family_values = [int(f) for f in families]
        self._inner.set_families(family_values)

    def detect(
        self,
        img: np.ndarray,
        intrinsics: CameraIntrinsics | None = None,
        tag_size: float | None = None,
        debug_telemetry: bool = False,
        **kwargs,
    ) -> DetectionBatch:
        """
        Detect tags in the image.

        Args:
            img: Input grayscale image (np.uint8).
            intrinsics: Optional CameraIntrinsics for 3D pose estimation.
            tag_size: Optional physical tag size (meters).

        Returns:
            A vectorized DetectionBatch object.
        """
        if img.dtype != np.uint8:
            raise ValueError(f"Input image must be uint8, got {img.dtype}")

        raw = self._inner.detect(
            img,
            intrinsics=intrinsics,
            tag_size=tag_size,
            debug_telemetry=debug_telemetry,
            **kwargs,
        )

        telemetry = None
        if raw.telemetry is not None:
            t = raw.telemetry
            telemetry = PipelineTelemetry(
                binarized=t.binarized,
                threshold_map=t.threshold_map,
                subpixel_jitter=t.subpixel_jitter,
                reprojection_errors=t.reprojection_errors,
            )

        return DetectionBatch(
            ids=raw.ids,
            corners=raw.corners,
            error_rates=raw.error_rates,
            poses=raw.poses,
            rejected_corners=raw.rejected_corners,
            rejected_error_rates=raw.rejected_error_rates,
            rejected_funnel_status=raw.rejected_funnel_status,
            telemetry=telemetry,
        )

    def detect_concurrent(
        self,
        frames: list[np.ndarray],
        intrinsics: CameraIntrinsics | None = None,
        tag_size: float | None = None,
    ) -> list[DetectionBatch]:
        """
        Detect tags in multiple frames concurrently.

        Releases the GIL for the entire parallel section. Telemetry and
        rejected-corner data are not available via this method.

        Args:
            frames: List of grayscale uint8 images.
            intrinsics: Optional CameraIntrinsics for 3D pose estimation.
            tag_size: Optional physical tag size (meters).

        Returns:
            A list of DetectionBatch, one per input frame, in the same order.
        """
        for i, img in enumerate(frames):
            if img.dtype != np.uint8:
                raise ValueError(f"Frame {i} must be uint8, got {img.dtype}")

        raw_results = self._inner.detect_concurrent(
            frames,
            intrinsics=intrinsics,
            tag_size=tag_size,
        )

        return [
            DetectionBatch(
                ids=r.ids,
                corners=r.corners,
                error_rates=r.error_rates,
                poses=r.poses,
                rejected_corners=r.rejected_corners,
                rejected_error_rates=r.rejected_error_rates,
                rejected_funnel_status=r.rejected_funnel_status,
            )
            for r in raw_results
        ]

Functions

__init__

__init__(profile: ProfileName | None = None, config: DetectorConfig | None = None, *, decimation: int | None = None, threads: int | None = None, families: list[TagFamily] | None = None, max_concurrent_frames: int | None = None) -> None

Source code in crates/locus-py/locus/__init__.py

def __init__(
    self,
    profile: ProfileName | None = None,
    config: DetectorConfig | None = None,
    *,
    decimation: int | None = None,
    threads: int | None = None,
    families: list[TagFamily] | None = None,
    max_concurrent_frames: int | None = None,
) -> None:
    if profile is not None and config is not None:
        raise ValueError("Pass either `profile` or `config`, not both.")

    if config is None:
        config = DetectorConfig.from_profile(profile or "standard")

    if families is None:
        families = [TagFamily.AprilTag36h11]

    self._inner = _create_detector_from_config(
        config_json=config.model_dump_json(),
        decimation=decimation,
        threads=threads,
        families=[int(f) for f in families],
        max_concurrent_frames=max_concurrent_frames,
    )

detect

detect(img: ndarray, intrinsics: CameraIntrinsics | None = None, tag_size: float | None = None, debug_telemetry: bool = False, **kwargs) -> DetectionBatch

Detect tags in the image.

Parameters:

Name	Type	Description	Default
img	ndarray	Input grayscale image (np.uint8).	required
intrinsics	CameraIntrinsics | None	Optional CameraIntrinsics for 3D pose estimation.	None
tag_size	float | None	Optional physical tag size (meters).	None

Returns:

Type	Description
DetectionBatch	A vectorized DetectionBatch object.

Source code in crates/locus-py/locus/__init__.py

def detect(
    self,
    img: np.ndarray,
    intrinsics: CameraIntrinsics | None = None,
    tag_size: float | None = None,
    debug_telemetry: bool = False,
    **kwargs,
) -> DetectionBatch:
    """
    Detect tags in the image.

    Args:
        img: Input grayscale image (np.uint8).
        intrinsics: Optional CameraIntrinsics for 3D pose estimation.
        tag_size: Optional physical tag size (meters).

    Returns:
        A vectorized DetectionBatch object.
    """
    if img.dtype != np.uint8:
        raise ValueError(f"Input image must be uint8, got {img.dtype}")

    raw = self._inner.detect(
        img,
        intrinsics=intrinsics,
        tag_size=tag_size,
        debug_telemetry=debug_telemetry,
        **kwargs,
    )

    telemetry = None
    if raw.telemetry is not None:
        t = raw.telemetry
        telemetry = PipelineTelemetry(
            binarized=t.binarized,
            threshold_map=t.threshold_map,
            subpixel_jitter=t.subpixel_jitter,
            reprojection_errors=t.reprojection_errors,
        )

    return DetectionBatch(
        ids=raw.ids,
        corners=raw.corners,
        error_rates=raw.error_rates,
        poses=raw.poses,
        rejected_corners=raw.rejected_corners,
        rejected_error_rates=raw.rejected_error_rates,
        rejected_funnel_status=raw.rejected_funnel_status,
        telemetry=telemetry,
    )

Concurrent Detection

Detector supports concurrent multi-frame processing via detect_concurrent. The internal pool size is set with max_concurrent_frames, either as a Detector keyword (locus.Detector(max_concurrent_frames=8)) or through the DetectorBuilder.

See the Concurrent Detection how-to for full usage examples.

DetectorBuilder

A fluent builder for Detector. It carries orchestration only — the configuration, tag families, decimation, threads and pool size. Detection settings live in DetectorConfig: edit one and pass it to with_config. build() re-validates the configuration through the Pydantic model and returns a locus.Detector.

import locus

cfg = locus.DetectorConfig.from_profile("standard")
cfg.decoder.min_contrast = 12.0

detector = (
    locus.DetectorBuilder()
    .with_config(cfg)
    .with_family(locus.TagFamily.AprilTag36h11)
    .with_threads(4)
    .with_max_concurrent_frames(8)  # Enable parallel batch processing
    .build()
)

Method	Description
with_profile(name)	Use a shipped profile ("standard", "grid", "high_accuracy").
with_config(config)	Use a DetectorConfig (default: the standard profile).
with_family(family)	Add a tag family to detect (default: AprilTag36h11).
with_decimation(n)	Spatial decimation factor (default 1).
with_threads(n)	Rayon worker count for the detector (see Thread control). 0 (default) = the global Rayon pool.
with_max_concurrent_frames(n)	Pool size for detect_concurrent (default 1 = sequential).
build()	Validate the configuration and build the Detector.

detect_concurrent(frames, *, intrinsics=None, tag_size=None) -> list[DetectionResult]

Detect tags in multiple frames concurrently using Rayon. Releases the GIL for the entire parallel section. Pool contexts are managed internally. Rejected-corner data and telemetry are not available via this method.

Parameter	Type	Description
frames	list[np.ndarray]	List of (H, W) uint8 grayscale frames.
intrinsics	CameraIntrinsics \| None	Camera intrinsics for 3D pose estimation.
tag_size	float \| None	Physical tag side length in metres.

Thread control

threads (Detector(threads=n) / DetectorBuilder.with_threads(n)) sets the number of Rayon workers the detector uses.

• 0 — the default — leaves the pipeline on Rayon's global pool, sized by RAYON_NUM_THREADS or the machine's core count. No extra threads are spawned.
• n > 0 builds one scoped Rayon pool of exactly n threads at detector construction, and runs detect and detect_concurrent inside it. The detector's CPU footprint is then bounded by n regardless of what the rest of the process does with the global pool, which is what you want when several detectors (or a detector and an unrelated Rayon workload) share a machine. For detect_concurrent the bound covers the frame-level fan-out and the intra-frame parallelism, so the two cannot over-subscribe each other.

The pool is built once per detector, never per frame. Detection output is identical for every value of threads: each parallel stage writes to disjoint, index-addressed chunks, so only latency and CPU usage change.

Fixed in 0.7.x

Before this release threads was accepted, stored and reported by Detector.config() but never read — the pipeline always ran on the global pool. Code that relied on threads=1 to keep the detector single-threaded was in fact running on all cores; it now gets one thread and will be correspondingly slower. Pass threads=0 (or omit it) for the previous behaviour.

Configuration

Locus uses Pydantic for robust configuration validation.

locus.DetectorConfig

Bases: BaseModel

Nested detector configuration.

The three shipped profiles live in crates/locus-core/profiles/*.json and are embedded into the Rust crate at compile time; the wheel reads the exact same bytes through the FFI. Those JSON profiles are authoritative. Load via :meth:from_profile for a shipped profile or :meth:from_profile_json for a user-supplied JSON string.

Source code in crates/locus-py/locus/_config.py

class DetectorConfig(BaseModel):
    """Nested detector configuration.

    The three shipped profiles live in ``crates/locus-core/profiles/*.json``
    and are embedded into the Rust crate at compile time; the wheel reads
    the exact same bytes through the FFI. Those JSON profiles are
    authoritative. Load via :meth:`from_profile` for a shipped profile or
    :meth:`from_profile_json` for a user-supplied JSON string.
    """

    model_config = ConfigDict(extra="forbid", frozen=False, arbitrary_types_allowed=True)

    name: str | None = None
    threshold: ThresholdConfig = Field(default_factory=ThresholdConfig)
    quad: QuadConfig = Field(default_factory=QuadConfig)
    decoder: DecoderConfig = Field(default_factory=DecoderConfig)
    pose: PoseConfig = Field(default_factory=PoseConfig)
    segmentation: SegmentationConfig = Field(default_factory=SegmentationConfig)

    @model_validator(mode="after")
    def _check_cross_group_compat(self) -> DetectorConfig:
        if (
            self.quad.extraction_mode == QuadExtractionMode.EdLines
            and self.decoder.refinement_mode == CornerRefinementMode.Erf
        ):
            raise ValueError(
                "quad.extraction_mode=EdLines is incompatible with decoder.refinement_mode=Erf"
            )
        # Mirrors `DetectorConfig::validate` in `crates/locus-core/src/config.rs`.
        if isinstance(self.quad.extraction_policy, _AdaptivePpbPolicy):
            if self.decoder.refinement_mode != getattr(CornerRefinementMode, "None"):
                raise ValueError(
                    "quad.extraction_policy=AdaptivePpb requires decoder.refinement_mode=None; "
                    "the AdaptivePpb routes carry their own refinement modes"
                )
            p = self.quad.extraction_policy.AdaptivePpb
            if p.low_extraction == p.high_extraction:
                raise ValueError(
                    "quad.extraction_policy.AdaptivePpb has identical low/high extraction "
                    'modes; use "Static" for single-mode operation'
                )
            # `threshold` bounds are already enforced by the Pydantic Field
            # constraints (gt=1.0, lt=5.0) on AdaptivePpbConfig; re-checking
            # here would duplicate the message surface.
            for ext, refine in (
                (p.low_extraction, p.low_refinement),
                (p.high_extraction, p.high_refinement),
            ):
                if ext == QuadExtractionMode.EdLines and refine == CornerRefinementMode.Erf:
                    raise ValueError(
                        "quad.extraction_policy.AdaptivePpb route uses EdLines with "
                        "refinement_mode=Erf, which is incompatible"
                    )
        return self

    @classmethod
    def from_profile(cls, name: ProfileName) -> DetectorConfig:
        """Load a shipped profile by name.

        General-purpose profiles must stay on ``QuadExtractionPolicy::Static``.
        Adaptive routing lives in explicit opt-in profiles
        (currently only ``high_accuracy``).
        """
        if name not in SHIPPED_PROFILES:
            raise ValueError(
                f"Unknown shipped profile {name!r}; expected one of {sorted(SHIPPED_PROFILES)}"
            )
        cfg = cls.model_validate_json(_shipped_profile_json(name))
        if name in _STATIC_ONLY_PROFILES and cfg.quad.extraction_policy != "Static":
            raise ValueError(
                f'Shipped profile {name!r} must carry quad.extraction_policy="Static"; '
                "adaptive routing lives in opt-in profiles only"
            )
        return cfg

    @classmethod
    def from_profile_json(cls, json_str: str) -> DetectorConfig:
        """Load a user-supplied profile from a JSON string."""
        return cls.model_validate_json(json_str)

Functions

from_profile classmethod

from_profile(name: ProfileName) -> DetectorConfig

Load a shipped profile by name.

General-purpose profiles must stay on QuadExtractionPolicy::Static. Adaptive routing lives in explicit opt-in profiles (currently only high_accuracy).

Source code in crates/locus-py/locus/_config.py

@classmethod
def from_profile(cls, name: ProfileName) -> DetectorConfig:
    """Load a shipped profile by name.

    General-purpose profiles must stay on ``QuadExtractionPolicy::Static``.
    Adaptive routing lives in explicit opt-in profiles
    (currently only ``high_accuracy``).
    """
    if name not in SHIPPED_PROFILES:
        raise ValueError(
            f"Unknown shipped profile {name!r}; expected one of {sorted(SHIPPED_PROFILES)}"
        )
    cfg = cls.model_validate_json(_shipped_profile_json(name))
    if name in _STATIC_ONLY_PROFILES and cfg.quad.extraction_policy != "Static":
        raise ValueError(
            f'Shipped profile {name!r} must carry quad.extraction_policy="Static"; '
            "adaptive routing lives in opt-in profiles only"
        )
    return cfg

from_profile_json classmethod

from_profile_json(json_str: str) -> DetectorConfig

Load a user-supplied profile from a JSON string.

Source code in crates/locus-py/locus/_config.py

@classmethod
def from_profile_json(cls, json_str: str) -> DetectorConfig:
    """Load a user-supplied profile from a JSON string."""
    return cls.model_validate_json(json_str)

Data Models

These classes represent the output and internal state of the detection pipeline.

locus.DetectionBatch dataclass

Vectorized detection results.

This dataclass contains parallel NumPy arrays representing a batch of detections.

Source code in crates/locus-py/locus/__init__.py

@dataclass(frozen=True)
class DetectionBatch:
    """
    Vectorized detection results.

    This dataclass contains parallel NumPy arrays representing a batch of detections.
    """

    ids: np.ndarray  # Shape: (N,), Dtype: int32
    corners: np.ndarray  # Shape: (N, 4, 2), Dtype: float32
    error_rates: np.ndarray  # Shape: (N,), Dtype: float32
    poses: np.ndarray | None = None  # Shape: (N, 7), Dtype: float32. [tx, ty, tz, qw, qx, qy, qz]
    telemetry: "PipelineTelemetry | None" = None
    rejected_corners: np.ndarray | None = None  # Shape: (M, 4, 2), Dtype: float32
    rejected_error_rates: np.ndarray | None = None  # Shape: (M,), Dtype: float32
    # Shape: (M,), Dtype: uint8. Codes from `FunnelStatus`.
    rejected_funnel_status: np.ndarray | None = None

    @property
    def centers(self) -> np.ndarray:
        """Compute centers from corners: (N, 2)"""
        return np.mean(self.corners, axis=1)

    def __len__(self) -> int:
        return len(self.ids)

Attributes

centers property

centers: ndarray

Compute centers from corners: (N, 2)

locus.Pose

Geometry

locus.CameraIntrinsics

Camera intrinsic parameters with optional lens distortion.

Construct with just (fx, fy, cx, cy) for a rectified camera, or pass distortion_model and dist_coeffs for a distorted camera:

# Pinhole (no distortion)
k = CameraIntrinsics(fx=800.0, fy=800.0, cx=400.0, cy=300.0)

# Brown-Conrady
k = CameraIntrinsics(
    fx=800.0, fy=800.0, cx=400.0, cy=300.0,
    distortion_model=DistortionModel.BrownConrady,
    dist_coeffs=[-0.3, 0.1, 0.001, -0.002, 0.0],
)

# Kannala-Brandt fisheye
k = CameraIntrinsics(
    fx=380.0, fy=380.0, cx=320.0, cy=240.0,
    distortion_model=DistortionModel.KannalaBrandt,
    dist_coeffs=[0.1, -0.01, 0.001, 0.0],
)

Attributes

dist_coeffs property writable

dist_coeffs: list[float]

Distortion coefficients corresponding to the chosen model.

• BrownConrady: [k1, k2, p1, p2, k3] (5 values)
• KannalaBrandt: [k1, k2, k3, k4] (4 values)
• Pinhole: ignored (may be empty)

distortion_model property writable

distortion_model: DistortionModel

The distortion model to use. Defaults to [DistortionModel::Pinhole].

Board Pose Estimation

Locus supports board-level 6-DOF pose estimation for AprilGrid and ChAruco boards. All board types enforce dictionary bounds at construction time — a ValueError is raised if the board requires more marker IDs than the target TagFamily provides.

For the underlying algorithm, see Board-Level Pose Estimation.

AprilGrid

locus.AprilGrid(
    rows: int,
    cols: int,
    spacing: float,        # gap between markers (metres)
    marker_length: float,  # physical marker side length (metres)
    family: TagFamily,
)

Immutable topology for an AprilTag grid board. spacing = square_length - marker_length. Raises ValueError if rows * cols > family.max_id_count().

Attribute	Type	Description
rows	int	Number of tag rows.
cols	int	Number of tag columns.

CharucoBoard

locus.CharucoBoard(
    rows: int,
    cols: int,
    square_length: float,  # physical checkerboard square side (metres)
    marker_length: float,  # physical ArUco marker side (metres)
    family: TagFamily,
)

Immutable topology for a ChAruco board. Markers occupy squares where (row + col) is even. Saddle points are at the outer corners of each square ((rows-1)*(cols-1) total). Raises ValueError if the required marker count exceeds family.max_id_count().

Attribute	Type	Description
rows	int	Number of square rows.
cols	int	Number of square columns.

BoardEstimator

Unified stateful estimator for both AprilGrid and ChAruco boards. All scratch buffers are pre-allocated at construction; estimate() performs zero heap allocations.

Construction:

# AprilGrid board
estimator = locus.BoardEstimator(board: AprilGrid)

# ChAruco board
estimator = locus.BoardEstimator.from_charuco(board: CharucoBoard)

Estimation:

result: BoardEstimateResult = estimator.estimate(
    detector: locus.Detector,
    img: np.ndarray,           # (H, W) uint8 grayscale
    intrinsics: CameraIntrinsics,
)

BoardEstimateResult attributes:

Attribute	Shape / Type	Description
ids	(N,) int32	Decoded tag IDs visible in this frame.
corners	(N, 4, 2) float32	Refined tag corner image coordinates.
board_pose	(7,) float64 or None	[tx, ty, tz, qw, qx, qy, qz] (Hamilton, scalar first) in camera frame — None if insufficient observations. Pose origin is the board's geometric centre.
board_cov	(6, 6) float64 or None	Body-frame pose covariance in \(\mathfrak{se}(3)\) tangent space, rotation first: [ω, v].

Enumerations

locus.TagFamily

Bases: IntEnum

Fiducial marker dictionary to decode.

Attributes

AprilTag16h5 class-attribute instance-attribute

AprilTag16h5 = ...

AprilTag 16h5 (30 codes, 4x4 bits, minimum Hamming distance 5).

AprilTag36h11 class-attribute instance-attribute

AprilTag36h11 = ...

AprilTag 36h11 (587 codes, 6x6 bits, minimum Hamming distance 11).

ArUco4x4_100 class-attribute instance-attribute

ArUco4x4_100 = ...

ArUco DICT_4X4_100.

ArUco4x4_50 class-attribute instance-attribute

ArUco4x4_50 = ...

ArUco DICT_4X4_50.

ArUco6x6_250 class-attribute instance-attribute

ArUco6x6_250 = ...

ArUco DICT_6X6_250.

ArUcoMip36h12 class-attribute instance-attribute

ArUcoMip36h12 = ...

ArUco MIP 36h12 (250 codes, 6x6 bits; DICT_ARUCO_MIP_36h12).

locus.CornerRefinementMode

Bases: IntEnum

Sub-pixel corner refinement applied during quad extraction.

Attributes

Erf class-attribute instance-attribute

Erf = ...

Fit each edge with a PSF-blurred step (ERF) model and intersect the edge lines. Incompatible with EdLines extraction.

None_ class-attribute instance-attribute

None_ = ...

Keep the extractor's corners (EdLines' Gauss-Newton sub-pixel corners, or ContourRdp's contour vertices). Required with AdaptivePpb routing.

locus.SegmentationConnectivity

Bases: IntEnum

Pixel connectivity of the dark regions that segmentation groups into candidates.

Attributes

Eight class-attribute instance-attribute

Eight = ...

8-connectivity: pixels also connect diagonally. Joins borders thinner than a pixel on the diagonal, but fuses diagonally touching squares.

Four class-attribute instance-attribute

Four = ...

4-connectivity: pixels connect horizontally and vertically only. Keeps dark squares that touch at a corner (checkerboards, AprilGrid connectors) apart.