API Reference

Synchronous datasets

ProfiledDataset

apairo.core.profiled_dataset.ProfiledDataset

Bases: SynchronousDataset, ConfigurableDataset

Synchronous dataset driven by a YAML structural profile.

Subclasses declare a _profile class attribute pointing to a YAML file (relative to apairo/dataset/profiles/ or an absolute path). The profile describes the directory layout, file extensions, dtypes, and any type transformations. All file discovery, loading, split filtering, and derived key resolution are handled automatically.

Example

Minimal subclass::

class MyDataset(ProfiledDataset):
    _profile = "my_dataset.yaml"

Usage::

ds = MyDataset("/data/my_dataset", keys=["lidar", "labels"], split="train")
sample = ds[0]
# sample.data["lidar"]  -> np.ndarray
# sample.data["labels"] -> np.ndarray

Attributes:

Name	Type	Description
available_keys	frozenset[str]	Frozenset of key names declared in the profile. Populated at class definition time from the YAML file.

See Also

YAML Profiles <https://apairo-robotics.github.io/apairo/datasets/yaml-profiles/>_ for the full profile specification.

Source code in apairo/core/profiled_dataset.py

class ProfiledDataset(SynchronousDataset, ConfigurableDataset):
    """Synchronous dataset driven by a YAML structural profile.

    Subclasses declare a `_profile` class attribute pointing to a YAML file
    (relative to `apairo/dataset/profiles/` or an absolute path).  The profile
    describes the directory layout, file extensions, dtypes, and any type
    transformations.  All file discovery, loading, split filtering, and derived
    key resolution are handled automatically.

    Example:
        Minimal subclass::

            class MyDataset(ProfiledDataset):
                _profile = "my_dataset.yaml"

        Usage::

            ds = MyDataset("/data/my_dataset", keys=["lidar", "labels"], split="train")
            sample = ds[0]
            # sample.data["lidar"]  -> np.ndarray
            # sample.data["labels"] -> np.ndarray

    Attributes:
        available_keys: Frozenset of key names declared in the profile.
            Populated at class definition time from the YAML file.

    See Also:
        `YAML Profiles <https://apairo-robotics.github.io/apairo/datasets/yaml-profiles/>`_
        for the full profile specification.
    """

    _profile: str

    def __init_subclass__(cls, **kwargs: object) -> None:
        super().__init_subclass__(**kwargs)
        profile_attr = cls.__dict__.get("_profile")
        if profile_attr:
            p = Path(profile_attr)
            profile_path = p if p.is_absolute() else _PROFILES_DIR / p
            if profile_path.exists():
                with open(profile_path) as f:
                    raw = yaml.safe_load(f)
                cls.available_keys = frozenset(raw.get("modalities", {}).keys())

    def _load_profile(self, root_dir: str | Path) -> dict:
        """Parse the ``_profile`` YAML into modality/layer geometry and set
        ``_root``.  Shared by :meth:`__init__` and :meth:`init` -- the latter
        needs the profile geometry without discovering files."""
        profile_path = (
            Path(self._profile)
            if Path(self._profile).is_absolute()
            else _PROFILES_DIR / self._profile
        )
        with open(profile_path) as f:
            raw = yaml.safe_load(f)

        self._modalities: dict[str, ModalitySpec] = {
            k: ModalitySpec.from_dict(k, v) for k, v in raw["modalities"].items()
        }
        self._layers: list[LayerSpec] = _parse_layers(raw["layers"])

        layer_types = [layer.type for layer in self._layers]
        self._modality_layer_idx: int = layer_types.index("modality")
        seq_idx = (
            layer_types.index("sequence")
            if "sequence" in layer_types
            else len(self._layers) - 1
        )
        self._seq_depth: int = len(self._layers) - seq_idx
        self._seq_layer_idx: int = seq_idx
        self._has_sequence_layer: bool = "sequence" in layer_types
        # Split spec is structural (lives in the profile) -- resolve it here so
        # the profile geometry is fully described without file discovery (used by
        # init/inventory, not only __init__).
        self._splits_spec: SplitSpec | None = _parse_splits_spec(raw.get("splits", {}))
        # A dataset-level clock declaration (the clock's origin is dataset-
        # specific -- see _collect_frame_clock). Structural, so set here.
        self._clock_spec: dict | None = raw.get("clock")

        self._root = Path(root_dir)
        return raw

    @classmethod
    def init(
        cls,
        directory: str | Path,
        *,
        merge: bool = False,
        overwrite: bool = False,
        name: str | None = None,
    ) -> Path:
        """Write ``.apairo/channels.yaml`` from the dataset profile.

        Maps the profile's canonical channel names onto the raw directories
        present on disk (e.g. ``os1_cloud_node_kitti_bin`` -> ``lidar``).  One
        config at *directory* covers every sequence the profile spans.

        Preprocessed channels are *not* auto-registered: after init, inspect
        :meth:`unregistered_channels` (the CLI prints them) and declare the ones
        you want with :meth:`register_channel`.

        Args:
            directory: Dataset root directory.
            merge: Add profile channels to an existing config, leaving channels
                already declared untouched.
            overwrite: Discard any existing ``.apairo`` and rebuild from scratch.
            name: Dataset name recorded in the root manifest
                (``.apairo/dataset.yaml``); defaults to the directory name.

        Returns:
            Path of the written ``channels.yaml`` (the manifest
            ``dataset.yaml``, recording the dataset class, is written alongside).
        """
        if overwrite and merge:
            raise ValueError("overwrite and merge are mutually exclusive.")

        self = cls.__new__(cls)
        self._load_profile(directory)
        root = Path(directory)

        config = self._bootstrap_config(root)
        if not config["channels"]:
            raise FileNotFoundError(
                f"No {cls.__name__} profile channels found under '{root}'."
            )

        if config_exists(root) and not overwrite:
            if not merge:
                raise FileExistsError(
                    f"{root / CONFIG_DIR} already exists -- pass overwrite to "
                    f"rebuild or merge to add profile channels."
                )
            existing = read_config(root)
            channels = dict(existing.get("channels", {}))
            for key, meta in config["channels"].items():
                channels.setdefault(key, meta)
            config = {
                **existing,
                "version": existing.get("version", 1),
                "channels": channels,
            }

        write_config(root, config)
        # Record the dataset identity in the root manifest so tooling (e.g.
        # `apairo status`) can dispatch through this profile instead of falling
        # back to the profile-unaware generic reading of channels.yaml.
        manifest = read_manifest(root)
        manifest["class"] = cls.__name__
        if name is not None:
            manifest["name"] = name
        manifest.setdefault("name", root.name)
        write_manifest(root, manifest)
        return root / CONFIG_DIR / CHANNELS_FILE

    @classmethod
    def unregistered_channels(cls, directory: str | Path) -> dict[str, str]:
        """Directories that look like channels but are in neither the profile nor
        the current config -- candidate preprocessed channels.

        Best-effort and report-only: returns ``{name: loader}`` for sub-directories
        at the profile's modality depth that hold loadable files.  Nothing is
        registered; declare the ones you want with :meth:`register_channel`.
        """
        self = cls.__new__(cls)
        self._load_profile(directory)
        root = Path(directory)
        config = read_config(root) if config_exists(root) else {"channels": {}}
        return self._preprocessed_candidates(root, config)

    @classmethod
    def remove_channel(
        cls, directory: str | Path, key: str, *, data: bool = False
    ) -> dict:
        """Remove a channel, profile-aware (overrides
        :meth:`ConfigurableDataset.remove_channel`).

        A profiled dataset keeps a single ``channels.yaml`` at the root but stores
        each channel's data **per sequence** (e.g. ``Rellis-3D/<seq>/<modality>``),
        so the generic ``root/key`` deletion would miss the sequence directories.
        This drops the (root) declaration and, with ``data=True``, deletes the
        channel's storage in *every* sequence the profile spans.

        Args:
            directory: Dataset root directory.
            key: Channel name to remove (the profile's canonical name).
            data: Also delete the channel's per-sequence data (destructive).
        """
        entry = _remove_channel(directory, key, data=False)
        if data:
            import shutil

            self = cls.__new__(cls)
            self._load_profile(directory)
            for target in self._channel_storage(key):
                if target.is_dir():
                    shutil.rmtree(target, ignore_errors=True)
                elif target.exists():
                    target.unlink()
        return entry

    def _channel_storage(self, key: str) -> list[Path]:
        """The channel's on-disk storage across all sequences: the per-sequence
        modality directories (per-frame channels) or the per-sequence stacked
        files (sequence-file channels). Profile geometry only -- safe on a
        partially-built instance."""
        spec = self._modalities[key]
        mapped = self._mapped_name(key)
        fixed_parts = _fixed_layer_parts(self._layers)
        base = self._root / Path(*fixed_parts) if fixed_parts else self._root
        if not base.is_dir():
            return []
        if spec.is_sequence_file:
            return sorted(base.glob(f"**/{mapped}{spec.ext}"), key=_natural_key)
        return sorted(
            (d for d in base.glob(f"**/{mapped}") if d.is_dir()), key=_natural_key
        )

    @classmethod
    def inventory(cls, directory: str | Path) -> dict:
        """Structural self-description from a root path, without loading data.

        The path-based form of :meth:`describe`: builds the profile geometry
        (no file discovery, no loaders) and reports identity, sequences, channel
        layout, splits and calibration.  Tolerant -- it describes a partial
        dataset without raising, unlike the constructor.  See :meth:`describe`
        for the returned schema.
        """
        self = cls.__new__(cls)
        self._load_profile(directory)
        return self._structure()

    def __init__(
        self,
        root_dir: str | Path,
        keys: list[str] | None = None,
        split: str | None = None,
        sequences: list[str] | None = None,
        *,
        sequence_ids: list[str] | None = None,
    ) -> None:
        self._load_profile(root_dir)

        # ``sequences`` selects which sequences to load (None = all), the
        # symmetric counterpart of ``split``.  ``sequence_ids`` is the deprecated
        # spelling.
        if sequence_ids is not None:
            warnings.warn(
                "ProfiledDataset(..., sequence_ids=...) is deprecated; use "
                "'sequences=' (same meaning: the sequence ids to keep).",
                DeprecationWarning,
                stacklevel=2,
            )
            if sequences is None:
                sequences = sequence_ids

        self._split_filter = split
        self._sequence_ids_filter: frozenset[str] | None = (
            frozenset(sequences) if sequences is not None else None
        )

        # _splits_spec is set by _load_profile (structural, profile-derived).
        self._frame_filter: set[tuple[str, str]] | None = None
        if (
            split is not None
            and self._splits_spec is not None
            and self._splits_spec.type == "lst"
        ):
            lst_rel = self._splits_spec.files.get(split)
            if lst_rel is None:
                available = list(self._splits_spec.files.keys())
                raise ValueError(
                    f"Split '{split}' not declared in profile. Available: {available}"
                )
            self._frame_filter = _read_lst_frame_set(self._root / lst_rel)

        # .apairo is the source of truth: raw channels present + preprocessed channels created.
        config = self._load_or_create_config(self._root)
        channels: dict = config.get("channels", {})

        # Aliases: a channel may expose a public ``alias``; requests and
        # ``sample.data`` speak the alias, while file discovery and the profile
        # keep the real channel name. ``to_real`` inverts the table so a
        # requested alias resolves to the directory that backs it. This unifies
        # channel naming across heterogeneous datasets in a single pipeline.
        self._alias_of: dict[str, str] = {
            k: v["alias"] for k, v in channels.items() if v.get("alias")
        }
        for real, alias in self._alias_of.items():
            if alias in channels and alias != real:
                raise ValueError(
                    f"Channel '{real}' is aliased as '{alias}', which is also a "
                    f"channel name in '{self._root}' -- requests for '{alias}' "
                    f"would be ambiguous. Clear it with "
                    f"`apairo alias {real} --remove` or pick another alias."
                )
        to_real = {alias: real for real, alias in self._alias_of.items()}

        if keys is None:
            keys = [
                self._alias_of.get(k, k)
                for k, v in channels.items()
                if v.get("kind", "raw") == "raw" and not self._modalities[k].optional
            ]

        # Normalize each requested key (alias *or* real name) to its real
        # channel name, and remember the public name it is exposed under so the
        # loaders and ``sample.data`` speak the request's language: ask for the
        # alias and everything downstream says the alias; ask for the real name
        # and it says the real name. Default keys speak the alias.
        real_keys = [to_real.get(k, k) for k in keys]
        self._public_of: dict[str, str] = dict(zip(real_keys, keys, strict=True))
        keys = real_keys

        # Classify each requested key.
        raw_keys: list[str] = []
        derived_keys: list[str] = []
        for k in keys:
            ch = channels.get(k)
            if ch is None:
                # Not in .apairo — allow if it is a profile key (raw, not yet scanned).
                if k not in self._modalities:
                    raise KeyError(
                        f"Key '{k}' is not available in '{self._root}'. "
                        f"Available: {sorted(channels)}. "
                        f"Register preprocessed channels with "
                        f"{type(self).__name__}.register_channel()."
                    )
                raw_keys.append(k)
            elif ch.get("kind", "raw") == "raw":
                raw_keys.append(k)
            else:
                derived_keys.append(k)

        self._set_keys(list(keys))
        self._files: dict[str, list[Path]] = {}
        self._loaders: dict = {}
        self._ref_key: str | None = None
        stacked_native: dict[str, ModalitySpec] = {}

        for key in raw_keys:
            spec = self._modalities[key]
            if spec.is_sequence_file:
                paths = self._discover_sequence_files(key)
                if not paths and not spec.optional:
                    raise FileNotFoundError(
                        f"Key '{key}': no '{self._mapped_name(key)}{spec.ext}' "
                        f"files found under {self._root}."
                    )
                if paths:
                    # Deferred: built once the global frame order is known, so the
                    # stacked rows can be aligned to the selected frames.
                    stacked_native[key] = spec
            else:
                paths = self._discover_native(key)
                if not paths and not spec.optional:
                    raise FileNotFoundError(
                        f"Key '{key}' declared in profile but no files found under {self._root}."
                    )
                if paths:
                    self._files[key] = paths
                    self._loaders[key] = _PerFrameLoader(paths, spec)
                    if self._ref_key is None:
                        self._ref_key = key

        frame_counts = {k: len(v) for k, v in self._loaders.items()}
        if len(set(frame_counts.values())) > 1:
            raise ValueError(f"Mismatched frame counts per key: {frame_counts}")

        self._modality_idx: int = self._modality_layer_idx
        if self._ref_key and self._files.get(self._ref_key):
            first = self._files[self._ref_key][0]
            rel_parts = first.relative_to(self._root).parts
            mapped = self._mapped_name(self._ref_key)
            if mapped in rel_parts:
                self._modality_idx = rel_parts.index(mapped)

        stacked_derived: list[str] = []
        for key in derived_keys:
            loader = channels[key]["loader"]
            # "npy" == one stacked file per sequence (NPYLoader-style, row per
            # frame); "npys"/"bin"/"img" == one file per frame. Stacked channels
            # are deferred until the frame order is known.
            if loader == "npy":
                stacked_derived.append(key)
                continue
            ext = "npy" if loader == "npys" else loader
            paths = self._discover_derived(key, ext)
            spec = ModalitySpec(ext=f".{ext}", loader=ext)
            self._loaders[key] = _PerFrameLoader(paths, spec)

        # If no native key was loaded (e.g. preprocessing a derived channel),
        # fall back to the first derived key as the path reference so that
        # derived_path() can resolve output locations.
        if self._ref_key is None:
            for key in derived_keys:
                loader = self._loaders.get(key)
                if loader is not None and loader.paths:
                    self._files[key] = loader.paths
                    self._ref_key = key
                    first = self._files[self._ref_key][0]
                    rel_parts = first.relative_to(self._root).parts
                    mapped = self._mapped_name(self._ref_key)
                    if mapped in rel_parts:
                        self._modality_idx = rel_parts.index(mapped)
                    break

        # Global frame order (anchor-driven) -- built before the stacked
        # sequence-file loaders, which align their rows to it.
        self._seq_groups: dict[str, list[int]] = {}
        anchor = (
            self._files.get(self._ref_key)
            if self._ref_key
            else next(
                (
                    v.paths
                    for v in self._loaders.values()
                    if isinstance(v, _PerFrameLoader)
                ),
                None,
            )
        )
        if anchor:
            for i, path in enumerate(anchor):
                seq_name = self._seq_root(path).name
                self._seq_groups.setdefault(seq_name, []).append(i)

        # Stacked channels (one file per sequence, one row per frame), mapped into
        # the selected frame order so a split or filter keeps them aligned with
        # the per-frame channels.  Native sequence files live directly in the
        # sequence dir (id = parent name); derived stacked files sit in a channel
        # subdir at modality depth (id = _seq_root).
        for key, spec in stacked_native.items():
            seq_paths = {p.parent.name: p for p in self._discover_sequence_files(key)}
            self._loaders[key] = self._build_stacked_loader(
                seq_paths, reshape=spec.reshape, reader=_loadtxt_2d
            )
        for key in stacked_derived:
            paths = self._discover_derived(key, "npy", apply_frame_filter=False)
            seq_paths = {self._seq_root(p).name: p for p in paths}
            self._loaders[key] = self._build_stacked_loader(
                seq_paths, reshape=None, reader=np.load
            )

        # Expose loaders and keys under their public (alias) name; file
        # discovery, the profile and ``_ref_key`` stay on the real channel name
        # internally (so ``self._files``/``derived_path`` keep locating files).
        loaded_real = [k for k in keys if k in self._loaders]
        self._loaders = {self._public_of[k]: v for k, v in self._loaders.items()}

        # Every channel must now expose the same number of selected frames.
        frame_counts = {k: len(v) for k, v in self._loaders.items()}
        if len(set(frame_counts.values())) > 1:
            raise ValueError(f"Mismatched frame counts per key: {frame_counts}")

        self._set_keys([self._public_of[k] for k in loaded_real])

        # Shared frame clock (synchronous): frame `i` is the same co-captured
        # sample across every channel, so a timestamp carried by ANY loaded
        # channel is the frame's clock -- shared by the whole sample. ``None``
        # when no channel carries one (clockless, e.g. positional SemanticKITTI).
        self.timestamps = self._collect_frame_clock(channels)

    def _collect_frame_clock(self, channels: dict) -> np.ndarray | None:
        """The dataset's shared per-frame clock, or ``None`` when clockless.

        The clock's *origin* is a concrete-dataset concern, so it is resolved from
        the most specific declaration down -- mirroring the async family's
        ``_key_providers`` layering:

        1. ``self._clock_provider`` -- a subclass callable ``(dataset) -> ndarray``
           (the escape hatch, for a computed or otherwise-shaped clock);
        2. the profile's ``clock:`` block -- a dataset-level declaration
           (``{dir, name/units/scale, ext?}`` self-contained, or ``{channel: X}``
           reusing channel *X*'s ``key``), so the clock is available even when its
           source channel is not among the loaded keys;
        3. a loaded channel that declares a ``key`` in ``channels.yaml`` (in-band);
        4. otherwise ``None`` -- clockless.

        The array is always in the selected frame order (splits and sequence
        selection applied) and validated per sequence.
        """
        provider = getattr(self, "_clock_provider", None)
        if provider is not None:
            return self._validate_clock("_clock_provider", provider(self))
        clock_spec = getattr(self, "_clock_spec", None)
        if clock_spec is not None:
            clock = self._clock_from_spec(clock_spec, channels)
            if clock is not None:
                return clock
            # declared but its source is absent on disk -> fall through to an
            # in-band channel key (a subset without the clock source stays usable).

        from apairo.core.keys import parse_filename_key

        for real_key, files in self._files.items():
            spec = channels.get(real_key, {}).get("key")
            if spec is None or not files:
                continue
            arr = parse_filename_key(
                [p.name for p in files],
                spec,
                directory=files[0].parent,
                label=f"Channel '{real_key}'",
            )
            return self._validate_clock(f"Channel '{real_key}'", arr)
        return None

    def _clock_from_spec(self, spec: dict, channels: dict) -> np.ndarray | None:
        """Resolve a profile ``clock:`` declaration to the per-frame array, aligned
        to the selected frames -- so the clock channel need not be loaded, and
        splits still line up (alignment is by ``(sequence, row)``, not filename).
        Returns ``None`` when the declared clock source is simply absent on disk
        (a camera-less subset stays clockless rather than failing)."""
        from apairo.core.keys import parse_filename_key

        if "file" in spec and "dir" not in spec and "channel" not in spec:
            return self._clock_from_sidecar(spec["file"])
        if "channel" in spec:
            ch = spec["channel"]
            if ch not in self._modalities:
                raise ValueError(
                    f"clock: channel '{ch}' is not a modality of {type(self).__name__}."
                )
            key_spec = channels.get(ch, {}).get("key")
            if key_spec is None:
                raise ValueError(
                    f"clock: channel '{ch}' declares no 'key' to read the clock from."
                )
            files_full = self._discover_native(ch, apply_frame_filter=False)
            label = f"clock (channel '{ch}')"
        else:
            directory = spec.get("dir")
            if directory is None:
                raise ValueError(
                    "clock: needs a 'channel', or a 'dir' with a 'name'/'file' key."
                )
            ext = spec.get("ext", "")
            pattern = f"**/{directory}/**/*{ext}" if ext else f"**/{directory}/**/*"
            files_full = sorted(
                (p for p in self._root.glob(pattern) if p.is_file()), key=_natural_key
            )
            key_spec = {
                k: spec[k] for k in ("name", "file", "scale", "units") if k in spec
            }
            label = "clock"

        if not files_full:  # declared but absent -> clockless (not an error)
            return None
        aligned = self._align_clock_files(files_full, label)
        if aligned is None:  # no per-frame anchor or partial coverage -> clockless
            return None
        return self._validate_clock(
            label,
            parse_filename_key(
                [p.name for p in aligned],
                key_spec,
                directory=aligned[0].parent,
                label=label,
            ),
        )

    def _clock_from_sidecar(self, name: str, label: str = "clock") -> np.ndarray | None:
        """A per-sequence sidecar clock: each sequence root holds *name* (one
        timestamp per frame, in the sequence's full order) -- e.g. SemanticKITTI's
        ``times.txt``. Aligned to the selected frames by ``(sequence, row)``.
        Returns ``None`` when the sidecar is simply absent (clockless subset)."""
        from apairo.loader import load_timestamps

        safe_config_name(name, label=f"{label} sidecar")
        seq_stamps: dict[str, np.ndarray] = {}
        for seq_dir in self._sequence_dirs():
            path = seq_dir / name
            if path.exists():
                seq_stamps[seq_dir.name] = load_timestamps(path)
        if not seq_stamps:
            return None
        if self._ref_key is None or not self._seq_groups:
            return None  # no per-frame anchor -> clockless
        rows = self._full_anchor_rows()
        seqs, stems = self.frame_sequence_ids, self.frame_stems
        out: list[float] = []
        for i in range(len(self)):
            seq = seqs[i]
            row = rows[seq][stems[i]]
            arr = seq_stamps.get(seq)
            if arr is None:  # a loaded sequence has no sidecar -> clockless
                return None
            if row >= len(arr):
                raise ValueError(
                    f"{label}: sequence '{seq}' sidecar '{name}' is shorter than its "
                    f"frames (need row {row}, have {len(arr)})."
                )
            out.append(float(arr[row]))
        return self._validate_clock(label, np.asarray(out, dtype=float))

    def _align_clock_files(
        self, files_full: list[Path], label: str
    ) -> list[Path] | None:
        """Map a clock channel's *full* (unfiltered) per-sequence files onto the
        selected frames by ``(sequence, row)`` -- the same alignment the stacked
        sequence-file loaders use, so the clock lines up under any split/filter
        even when its filenames differ from the anchor's. Returns ``None``
        (clockless) when there is no per-frame anchor, or when a loaded sequence
        carries no clock source at all -- partial coverage stays clockless rather
        than crashing an otherwise-valid load."""
        if self._ref_key is None or not self._seq_groups:
            return None
        by_seq: dict[str, list[Path]] = {}
        for f in files_full:
            by_seq.setdefault(self._seq_root(f).name, []).append(f)
        rows = self._full_anchor_rows()
        seqs, stems = self.frame_sequence_ids, self.frame_stems
        out: list[Path] = []
        for i in range(len(self)):
            seq = seqs[i]
            row = rows[seq][stems[i]]
            seq_files = by_seq.get(seq, [])
            if not seq_files:  # a loaded sequence has no clock source -> clockless
                return None
            if row >= len(seq_files):
                raise ValueError(
                    f"{label}: sequence '{seq}' has {len(seq_files)} clock file(s) "
                    f"but the frames need row {row} -- the clock source must be 1:1 "
                    f"co-indexed with the frames."
                )
            out.append(seq_files[row])
        return out

    def _validate_clock(self, label: str, arr) -> np.ndarray:
        arr = np.asarray(arr, dtype=float).ravel()
        if len(arr) != len(self):
            raise ValueError(
                f"{label}: clock has {len(arr)} value(s) for {len(self)} frame(s)."
            )
        self._check_clock_monotonic(label, arr)
        return arr

    def _check_clock_monotonic(self, label: str, arr: np.ndarray) -> None:
        """A frame clock must be non-decreasing *within each sequence*; it resets
        across sequences (each recording carries its own timeline), so the flat
        array is validated per sequence, never globally."""
        groups = list(self._seq_groups.values()) or [list(range(len(arr)))]
        for idxs in groups:
            seq = arr[np.asarray(idxs, dtype=int)]
            if seq.size > 1 and np.any(np.diff(seq) < 0):
                raise ValueError(
                    f"{label}: frame clock is not non-decreasing within a sequence "
                    f"-- check the key captures the time field."
                )

    def _seq_root(self, path: Path) -> Path:
        d = path
        for _ in range(self._seq_depth):
            d = d.parent
        return d

    def derived_path(self, idx: int, key: str, ext: str) -> Path:
        ref_key = self._ref_key
        if ref_key is None:
            raise RuntimeError("derived_path() needs a loaded reference channel.")
        ref = self._files[ref_key][idx]
        rel = ref.relative_to(self._root)
        parts = list(rel.parts)
        src_spec = self._modalities.get(ref_key)
        n = len(src_spec.effective_subpath(ref_key)) if src_spec else 1
        parts[self._modality_idx : self._modality_idx + n] = [key]
        parts[-1] = f"{ref.stem}.{ext}"
        return self._root / Path(*parts)

    def _is_present(self, root_dir: Path, key: str) -> bool:
        spec = self._modalities[key]
        mapped = self._mapped_name(key)
        fixed_parts = _fixed_layer_parts(self._layers)
        if spec.is_sequence_file:
            return any(root_dir.glob(f"**/{mapped}{spec.ext}"))
        if fixed_parts:
            prefix = Path(*fixed_parts)
            return any(root_dir.glob(str(prefix / "**" / mapped / f"*{spec.ext}")))
        return any(root_dir.glob(f"**/{mapped}/**/*{spec.ext}"))

    def _bootstrap_config(self, root_dir: Path) -> dict:
        channels = {}
        for key in sorted(self.available_keys):
            if self._is_present(root_dir, key):
                spec = self._modalities[key]
                loader = spec.loader or _EXT_TO_LOADER.get(spec.ext, "bin")
                channels[key] = {"loader": loader}
        return {"version": 1, "channels": channels}

    def _preprocessed_candidates(self, root_dir: Path, config: dict) -> dict[str, str]:
        """Modality-depth directories on disk that are neither a mapped raw
        channel nor already declared -- i.e. likely preprocessed channels apairo
        has not been told about.  Report-only helper for :meth:`unregistered_channels`."""
        from apairo.dataset.async_layout.dataset import (
            _detect_loader,  # local: avoid import cycle
        )

        declared = set(config.get("channels", {}))
        raw_dirs = {self._mapped_name(k) for k in self.available_keys}
        prefix = [
            layer.value if layer.type == "fixed" and layer.value is not None else "*"
            for layer in self._layers[: self._modality_layer_idx]
        ]
        pattern = str(Path(*prefix) / "*") if prefix else "*"

        found: dict[str, str] = {}
        for d in sorted(root_dir.glob(pattern)):
            if not d.is_dir() or d.name.startswith(".") or d.name in found:
                continue
            if d.name in raw_dirs or d.name in declared:
                continue
            loader = _detect_loader(d)
            if loader is not None:
                found[d.name] = loader
        return found

    def _mapped_name(self, key: str) -> str:
        layer = self._layers[self._modality_layer_idx]
        if isinstance(layer.value, dict):
            return layer.value.get(key, key)
        return key

    def _discover_sequence_files(self, key: str) -> list[Path]:
        """Find sequence-level files (one per sequence, not per frame)."""
        spec = self._modalities[key]
        fixed_parts = _fixed_layer_parts(self._layers)
        mapped = self._mapped_name(key)

        if fixed_parts:
            prefix = Path(*fixed_parts)
            pattern = str(prefix / f"**/{mapped}{spec.ext}")
        else:
            pattern = f"**/{mapped}{spec.ext}"

        paths = sorted(self._root.glob(pattern), key=_natural_key)
        if self._sequence_ids_filter is not None:
            paths = [p for p in paths if p.parent.name in self._sequence_ids_filter]
        return paths

    def _discover_derived(
        self, key: str, ext: str, apply_frame_filter: bool = True
    ) -> list[Path]:
        fixed_parts = _fixed_layer_parts(self._layers)
        if fixed_parts:
            prefix = Path(*fixed_parts)
            pattern = str(prefix / "**" / key / f"*.{ext}")
        else:
            pattern = f"**/{key}/**/*.{ext}"

        files = sorted(self._root.glob(pattern), key=_natural_key)
        # Directory-based splits live in the path (filter by it); lst-based splits
        # have no split layer and are resolved by _frame_filter below -- mirror
        # _discover_native so a derived channel splits the same way a native one does.
        if self._split_filter:
            split_layer = next(
                (layer for layer in self._layers if layer.type == "split"), None
            )
            if split_layer is not None:
                files = [
                    f
                    for f in files
                    if self._split_filter in f.relative_to(self._root).parts
                ]
        if self._sequence_ids_filter is not None:
            files = [
                f for f in files if self._seq_root(f).name in self._sequence_ids_filter
            ]
        if self._frame_filter is not None and apply_frame_filter:
            files = [
                f
                for f in files
                if (self._seq_root(f).name, f.stem) in self._frame_filter
            ]
        if not files:
            raise FileNotFoundError(
                f"Derived key '{key}': no .{ext} files found under '{self._root}'. "
                f"Run run_preprocess(...) to generate them."
            )
        return files

    def _discover_native(self, key: str, apply_frame_filter: bool = True) -> list[Path]:
        spec = self._modalities[key]
        fixed_parts = _fixed_layer_parts(self._layers)
        mapped = self._mapped_name(key)

        if fixed_parts:
            prefix = Path(*fixed_parts)
            pattern = str(prefix / "**" / mapped / f"*{spec.ext}")
        else:
            pattern = f"**/{mapped}/**/*{spec.ext}"

        files = sorted(self._root.glob(pattern), key=_natural_key)

        if self._split_filter:
            split_layer = next(
                (layer for layer in self._layers if layer.type == "split"), None
            )
            if split_layer is not None:
                files = [
                    f
                    for f in files
                    if self._split_filter in f.relative_to(self._root).parts
                ]
        if self._sequence_ids_filter is not None:
            files = [
                f for f in files if self._seq_root(f).name in self._sequence_ids_filter
            ]
        if self._frame_filter is not None and apply_frame_filter:
            files = [
                f
                for f in files
                if (self._seq_root(f).name, f.stem) in self._frame_filter
            ]
        return files

    def _full_anchor_rows(self) -> dict[str, dict[str, int]]:
        """Row of each frame stem within its sequence's full (unfiltered) order.

        Stacked sequence files store a sequence's frames as rows in this order,
        so this maps a selected frame's stem to its row -- letting a stacked
        channel honour the same split/predicate selection as the per-frame
        channels.  Keyed off the anchor (``_ref_key``) glob without the frame
        filter."""
        if self._ref_key is None:
            raise RuntimeError("_full_anchor_rows() needs a per-frame anchor channel.")
        cached = getattr(self, "_anchor_rows_cache", None)
        if cached is not None:
            return cached
        per_seq: dict[str, list[str]] = {}
        for f in self._discover_native(self._ref_key, apply_frame_filter=False):
            per_seq.setdefault(self._seq_root(f).name, []).append(f.stem)
        # Memoized: the anchor glob is construction-fixed, but this is called once
        # per stacked channel and once per clock alignment -- re-globbing the whole
        # anchor tree each time is wasteful.
        self._anchor_rows_cache: dict[str, dict[str, int]] = {
            seq: {stem: i for i, stem in enumerate(stems)}
            for seq, stems in per_seq.items()
        }
        return self._anchor_rows_cache

    def _build_stacked_loader(
        self, seq_paths, reshape, reader
    ) -> _StackedSequenceLoader:
        """Wrap one stacked file per sequence (``{seq_id: path}``) into a loader
        aligned to the selected global frame order, so it stays in step with the
        per-frame channels under splits and filters."""
        if self._ref_key is not None and self._seq_groups:
            rows = self._full_anchor_rows()
            seqs, stems = self.frame_sequence_ids, self.frame_stems
            index = [(seqs[i], rows[seqs[i]][stems[i]]) for i in range(len(self))]
        else:
            # No per-frame anchor to define a selection: expose every row in order.
            index = [
                (seq, r)
                for seq in sorted(seq_paths)
                for r in range(len(reader(seq_paths[seq])))
            ]
        return _StackedSequenceLoader(seq_paths, reader, index, reshape)

    @property
    def loaders(self) -> dict:  # type: ignore[override]  # property over the base attribute
        """Per-channel loaders, indexed by global frame index."""
        return self._loaders

    def __len__(self) -> int:
        if not self._loaders:
            return 0
        return len(next(iter(self._loaders.values())))

    def _sequence_dirs(self) -> list[Path]:
        """Sequence directories on disk, from the profile geometry alone.

        Lists directories at the profile's sequence depth (children of the
        ``fixed`` prefix).  Structural only -- no file discovery -- so it works
        on a partially-built instance (``init``/``inventory``)."""
        prefix = _fixed_layer_parts(self._layers[: self._seq_layer_idx])
        base = self._root / Path(*prefix) if prefix else self._root
        if not base.is_dir():
            return []
        if not self._has_sequence_layer:
            return [base]
        return sorted(
            d for d in base.iterdir() if d.is_dir() and not d.name.startswith(".")
        )

    def _structure(self) -> dict:
        """The structured self-description returned by :meth:`describe` /
        :meth:`inventory`.  Profile geometry + cheap filesystem probes only --
        no loaders, no per-frame counting (that is recoverable from a loaded
        dataset: ``len(ds)``, ``ds[i].data[key].shape``, ``ds.sequence_ids``)."""
        root = self._root
        fixed = [layer.value for layer in self._layers if layer.type == "fixed"]

        raw_channels: dict[str, dict] = {}
        present: list[str] = []
        missing: list[str] = []
        for key in sorted(self.available_keys):
            spec = self._modalities[key]
            is_present = self._is_present(root, key)
            (present if is_present else missing).append(key)
            raw_channels[key] = {
                "loader": spec.loader or _EXT_TO_LOADER.get(spec.ext, "bin"),
                "dir": self._mapped_name(key),  # canonical -> on-disk subdir name
                "present": is_present,
                "optional": spec.optional,
                "sequence_file": spec.is_sequence_file,
            }

        preprocess: dict[str, dict] = {}
        if config_exists(root):
            preprocess = {
                k: v
                for k, v in read_config(root).get("channels", {}).items()
                if v.get("kind") == "preprocess"
            }

        manifest = read_manifest(root)
        return {
            "class": type(self).__name__,
            "name": manifest.get("name", root.name),
            "root": str(root),
            "layout": {"fixed": fixed},
            "sequences": [d.name for d in self._sequence_dirs()],
            "splits": self.splits,
            "calibration": sorted(read_calibration(root)),
            "raw": {"present": present, "missing": missing, "channels": raw_channels},
            "preprocess": preprocess,
        }

    def describe(self, sequence_id: str | None = None) -> dict:  # type: ignore[override]  # instance form supersedes the mixin classmethod
        """Describe this dataset's structure -- identity, sequences, channels.

        Returns a structured dict (and prints a human-readable summary).  Cross-
        references the profile's declared modalities with what is on disk to show
        which raw channels are present or missing and where each lives, plus any
        registered preprocessed channels.  Per-frame facts (counts, shapes) are
        intentionally *not* here -- read them from the loaded dataset
        (``len(ds)``, ``ds[i].data[key].shape``).

        Args:
            sequence_id: Optional identifier used as the printed display label
                only. Channel availability is dataset-wide.

        Returns:
            ``{"class", "name", "root", "layout": {"fixed": [...]},
            "sequences": [...], "splits": [...], "calibration": [...],
            "raw": {"present": [...], "missing": [...], "channels": {key: {...}}},
            "preprocess": {key: meta}}``

        Example::

            ds = Rellis3DDataset("/data/RELLIS")
            ds.describe("00000")
        """
        info = self._structure()
        label = sequence_id if sequence_id is not None else info["name"]
        print(f"\n{info['class']} -- {label}")
        print("─" * 50)
        print("Raw channels")
        if info["raw"]["present"]:
            print("  present  :", ", ".join(info["raw"]["present"]))
        if info["raw"]["missing"]:
            print("  missing  :", ", ".join(info["raw"]["missing"]))
        if not info["raw"]["present"] and not info["raw"]["missing"]:
            print("  (none)")
        print("Preprocessed channels")
        if info["preprocess"]:
            for key, meta in sorted(info["preprocess"].items()):
                ts_info = (
                    f"<- timestamps from {meta['timestamps_from']}"
                    if "timestamps_from" in meta
                    else "<- own timestamps"
                )
                src_info = (
                    f"  sources: {meta['sources']}" if meta.get("sources") else ""
                )
                print(f"  {key:<20} {meta['loader']:<6} {ts_info}{src_info}")
        else:
            print("  (none)")
        print()
        return info

    @property
    def splits(self) -> list[str]:
        if self._splits_spec is not None:
            return list(self._splits_spec.files.keys())
        for layer in self._layers:
            if layer.type == "split" and isinstance(layer.value, list):
                return list(layer.value)
        return []

    def split(self, name: str) -> ProfiledDataset:
        """Return a new dataset *instance* filtered to the named split.

        This re-instantiates the dataset, so it starts with a clean pipeline: any
        registered ``transform()`` is intentionally NOT carried over. To split
        mid-chain while keeping transforms, use :meth:`filter_split` instead."""
        return type(self)(
            self._root,
            keys=list(self._keys),
            split=name,
            sequences=list(self._sequence_ids_filter)
            if self._sequence_ids_filter
            else None,
        )

    @property
    def sequence_ids(self) -> list[str]:
        return list(self._seq_groups.keys())

    @functools.cached_property
    def frame_sequence_ids(self) -> np.ndarray:
        """Sequence ID for every frame, indexed by global frame index.

        Returns a string array of shape ``(len(self),)`` where
        ``frame_sequence_ids[i]`` is the sequence ID that frame ``i`` belongs
        to.  Combined with :attr:`FilteredView.indices`, this lets you split a
        pre-filtered dataset by sequence without a second disk sweep::

            ds_filtered = ds.filter("trav_gt", HasMinPositives(min_pos))
            seq_ids = ds.frame_sequence_ids[ds_filtered.indices]

            for train_seqs, val_seqs in folds:
                train_idx = np.where(np.isin(seq_ids, train_seqs))[0]
                val_idx   = np.where(np.isin(seq_ids, val_seqs))[0]
                ds_train  = ds_filtered.filter(train_idx)
                ds_val    = ds_filtered.filter(val_idx)
        """
        result = np.empty(len(self), dtype=object)
        for seq_id, indices in self._seq_groups.items():
            result[indices] = seq_id
        return result

    @functools.cached_property
    def frame_stems(self) -> np.ndarray:
        """Filename stem for every frame, indexed by global frame index."""
        result = np.empty(len(self), dtype=object)
        anchor = self._files.get(self._ref_key) if self._ref_key is not None else None
        if anchor:
            for i, path in enumerate(anchor):
                result[i] = path.stem
        return result

    def filter_split(self, name: str) -> AbstractDataset:
        """Return a FilteredView restricted to the named predefined split.

        Applies the split without re-instantiating the dataset — registered
        transforms are preserved.  Replaces :meth:`split` for use mid-chain::

            ds.transform("lidar", RobotFilter())
            ds_train = ds.filter_split("train")  # transforms kept

        Works for both LST-based splits (SemanticKITTI/Rellis) and directory-layer
        splits (GOOSE), so there is a transform-preserving split for every dataset.
        """
        if self._splits_spec is not None and self._splits_spec.type == "lst":
            frame_set = self._lst_frame_filter(name)
        else:
            frame_set = self._split_layer_frame_filter(name)
        return _apply_lst_filter(self, frame_set)

    def _split_layer_frame_filter(self, name: str) -> set[tuple[str, str]]:
        """Frames belonging to a directory-layer split (its dir name in the path)."""
        if name not in self.splits:
            raise ValueError(
                f"Split '{name}' not found. Available: {self.splits or '(none)'}."
            )
        anchor = self._files.get(self._ref_key) if self._ref_key is not None else None
        if not anchor:
            raise ValueError(
                f"{type(self).__name__}.filter_split needs a per-frame anchor channel."
            )
        return {
            (self._seq_root(f).name, f.stem)
            for f in anchor
            if name in f.relative_to(self._root).parts
        }

    def _lst_frame_filter(self, name: str) -> set[tuple[str, str]]:
        if self._splits_spec is None or self._splits_spec.type != "lst":
            raise ValueError(f"{type(self).__name__} has no LST-based splits defined.")
        lst_rel = self._splits_spec.files.get(name)
        if lst_rel is None:
            available = list(self._splits_spec.files)
            raise ValueError(f"Split '{name}' not found. Available: {available}")
        return _read_lst_frame_set(self._root / lst_rel)

    def sequences(self) -> list[SequenceView]:
        from apairo.core.sequence_view import SequenceView  # noqa: F401

        return [self.sequence(sid) for sid in self.sequence_ids]

    def sequence(self, seq_id: str) -> SequenceView:
        if seq_id not in self._seq_groups:
            raise KeyError(
                f"Sequence '{seq_id}' not found. Available: {self.sequence_ids}"
            )
        from apairo.core.sequence_view import SequenceView

        return SequenceView(self, self._seq_groups[seq_id], seq_id)

    def _load(self, idx) -> Sample:
        if isinstance(idx, tuple):
            seq_id, local_idx = idx
            view = self.sequence(seq_id)
            return self._load(view._indices[local_idx])
        if not 0 <= idx < len(self):
            raise IndexError(f"Index {idx} out of range [0, {len(self)})")
        return Sample(
            data={key: self._loaders[key][idx] for key in self._keys},
            timestamp=None if self.timestamps is None else float(self.timestamps[idx]),
        )

loaders property

loaders: dict

Per-channel loaders, indexed by global frame index.

frame_sequence_ids cached property

frame_sequence_ids: ndarray

Sequence ID for every frame, indexed by global frame index.

Returns a string array of shape (len(self),) where frame_sequence_ids[i] is the sequence ID that frame i belongs to. Combined with :attr:FilteredView.indices, this lets you split a pre-filtered dataset by sequence without a second disk sweep::

ds_filtered = ds.filter("trav_gt", HasMinPositives(min_pos))
seq_ids = ds.frame_sequence_ids[ds_filtered.indices]

for train_seqs, val_seqs in folds:
    train_idx = np.where(np.isin(seq_ids, train_seqs))[0]
    val_idx   = np.where(np.isin(seq_ids, val_seqs))[0]
    ds_train  = ds_filtered.filter(train_idx)
    ds_val    = ds_filtered.filter(val_idx)

frame_stems cached property

frame_stems: ndarray

Filename stem for every frame, indexed by global frame index.

init classmethod

init(directory: str | Path, *, merge: bool = False, overwrite: bool = False, name: str | None = None) -> Path

Write .apairo/channels.yaml from the dataset profile.

Maps the profile's canonical channel names onto the raw directories present on disk (e.g. os1_cloud_node_kitti_bin -> lidar). One config at directory covers every sequence the profile spans.

Preprocessed channels are not auto-registered: after init, inspect :meth:unregistered_channels (the CLI prints them) and declare the ones you want with :meth:register_channel.

Parameters:

Name	Type	Description	Default
directory	str | Path	Dataset root directory.	required
merge	bool	Add profile channels to an existing config, leaving channels already declared untouched.	False
overwrite	bool	Discard any existing .apairo and rebuild from scratch.	False
name	str | None	Dataset name recorded in the root manifest (.apairo/dataset.yaml); defaults to the directory name.	None

Returns:

Type	Description
Path	Path of the written channels.yaml (the manifest
Path	dataset.yaml, recording the dataset class, is written alongside).

Source code in apairo/core/profiled_dataset.py

@classmethod
def init(
    cls,
    directory: str | Path,
    *,
    merge: bool = False,
    overwrite: bool = False,
    name: str | None = None,
) -> Path:
    """Write ``.apairo/channels.yaml`` from the dataset profile.

    Maps the profile's canonical channel names onto the raw directories
    present on disk (e.g. ``os1_cloud_node_kitti_bin`` -> ``lidar``).  One
    config at *directory* covers every sequence the profile spans.

    Preprocessed channels are *not* auto-registered: after init, inspect
    :meth:`unregistered_channels` (the CLI prints them) and declare the ones
    you want with :meth:`register_channel`.

    Args:
        directory: Dataset root directory.
        merge: Add profile channels to an existing config, leaving channels
            already declared untouched.
        overwrite: Discard any existing ``.apairo`` and rebuild from scratch.
        name: Dataset name recorded in the root manifest
            (``.apairo/dataset.yaml``); defaults to the directory name.

    Returns:
        Path of the written ``channels.yaml`` (the manifest
        ``dataset.yaml``, recording the dataset class, is written alongside).
    """
    if overwrite and merge:
        raise ValueError("overwrite and merge are mutually exclusive.")

    self = cls.__new__(cls)
    self._load_profile(directory)
    root = Path(directory)

    config = self._bootstrap_config(root)
    if not config["channels"]:
        raise FileNotFoundError(
            f"No {cls.__name__} profile channels found under '{root}'."
        )

    if config_exists(root) and not overwrite:
        if not merge:
            raise FileExistsError(
                f"{root / CONFIG_DIR} already exists -- pass overwrite to "
                f"rebuild or merge to add profile channels."
            )
        existing = read_config(root)
        channels = dict(existing.get("channels", {}))
        for key, meta in config["channels"].items():
            channels.setdefault(key, meta)
        config = {
            **existing,
            "version": existing.get("version", 1),
            "channels": channels,
        }

    write_config(root, config)
    # Record the dataset identity in the root manifest so tooling (e.g.
    # `apairo status`) can dispatch through this profile instead of falling
    # back to the profile-unaware generic reading of channels.yaml.
    manifest = read_manifest(root)
    manifest["class"] = cls.__name__
    if name is not None:
        manifest["name"] = name
    manifest.setdefault("name", root.name)
    write_manifest(root, manifest)
    return root / CONFIG_DIR / CHANNELS_FILE

unregistered_channels classmethod

unregistered_channels(directory: str | Path) -> dict[str, str]

Directories that look like channels but are in neither the profile nor the current config -- candidate preprocessed channels.

Best-effort and report-only: returns {name: loader} for sub-directories at the profile's modality depth that hold loadable files. Nothing is registered; declare the ones you want with :meth:register_channel.

Source code in apairo/core/profiled_dataset.py

@classmethod
def unregistered_channels(cls, directory: str | Path) -> dict[str, str]:
    """Directories that look like channels but are in neither the profile nor
    the current config -- candidate preprocessed channels.

    Best-effort and report-only: returns ``{name: loader}`` for sub-directories
    at the profile's modality depth that hold loadable files.  Nothing is
    registered; declare the ones you want with :meth:`register_channel`.
    """
    self = cls.__new__(cls)
    self._load_profile(directory)
    root = Path(directory)
    config = read_config(root) if config_exists(root) else {"channels": {}}
    return self._preprocessed_candidates(root, config)

remove_channel classmethod

remove_channel(directory: str | Path, key: str, *, data: bool = False) -> dict

Remove a channel, profile-aware (overrides :meth:ConfigurableDataset.remove_channel).

A profiled dataset keeps a single channels.yaml at the root but stores each channel's data per sequence (e.g. Rellis-3D/<seq>/<modality>), so the generic root/key deletion would miss the sequence directories. This drops the (root) declaration and, with data=True, deletes the channel's storage in every sequence the profile spans.

Parameters:

Name	Type	Description	Default
directory	str | Path	Dataset root directory.	required
key	str	Channel name to remove (the profile's canonical name).	required
data	bool	Also delete the channel's per-sequence data (destructive).	False

Source code in apairo/core/profiled_dataset.py

@classmethod
def remove_channel(
    cls, directory: str | Path, key: str, *, data: bool = False
) -> dict:
    """Remove a channel, profile-aware (overrides
    :meth:`ConfigurableDataset.remove_channel`).

    A profiled dataset keeps a single ``channels.yaml`` at the root but stores
    each channel's data **per sequence** (e.g. ``Rellis-3D/<seq>/<modality>``),
    so the generic ``root/key`` deletion would miss the sequence directories.
    This drops the (root) declaration and, with ``data=True``, deletes the
    channel's storage in *every* sequence the profile spans.

    Args:
        directory: Dataset root directory.
        key: Channel name to remove (the profile's canonical name).
        data: Also delete the channel's per-sequence data (destructive).
    """
    entry = _remove_channel(directory, key, data=False)
    if data:
        import shutil

        self = cls.__new__(cls)
        self._load_profile(directory)
        for target in self._channel_storage(key):
            if target.is_dir():
                shutil.rmtree(target, ignore_errors=True)
            elif target.exists():
                target.unlink()
    return entry

inventory classmethod

inventory(directory: str | Path) -> dict

Structural self-description from a root path, without loading data.

The path-based form of :meth:describe: builds the profile geometry (no file discovery, no loaders) and reports identity, sequences, channel layout, splits and calibration. Tolerant -- it describes a partial dataset without raising, unlike the constructor. See :meth:describe for the returned schema.

Source code in apairo/core/profiled_dataset.py

@classmethod
def inventory(cls, directory: str | Path) -> dict:
    """Structural self-description from a root path, without loading data.

    The path-based form of :meth:`describe`: builds the profile geometry
    (no file discovery, no loaders) and reports identity, sequences, channel
    layout, splits and calibration.  Tolerant -- it describes a partial
    dataset without raising, unlike the constructor.  See :meth:`describe`
    for the returned schema.
    """
    self = cls.__new__(cls)
    self._load_profile(directory)
    return self._structure()

describe

describe(sequence_id: str | None = None) -> dict

Describe this dataset's structure -- identity, sequences, channels.

Returns a structured dict (and prints a human-readable summary). Cross- references the profile's declared modalities with what is on disk to show which raw channels are present or missing and where each lives, plus any registered preprocessed channels. Per-frame facts (counts, shapes) are intentionally not here -- read them from the loaded dataset (len(ds), ds[i].data[key].shape).

Parameters:

Name	Type	Description	Default
sequence_id	str | None	Optional identifier used as the printed display label only. Channel availability is dataset-wide.	None

Returns:

Type	Description
dict	``{"class", "name", "root", "layout": {"fixed": [...]},
dict	"sequences": [...], "splits": [...], "calibration": [...],
dict	"raw": {"present": [...], "missing": [...], "channels": {key: {...}}},
dict	"preprocess": {key: meta}}``

Example::

ds = Rellis3DDataset("/data/RELLIS")
ds.describe("00000")

Source code in apairo/core/profiled_dataset.py

def describe(self, sequence_id: str | None = None) -> dict:  # type: ignore[override]  # instance form supersedes the mixin classmethod
    """Describe this dataset's structure -- identity, sequences, channels.

    Returns a structured dict (and prints a human-readable summary).  Cross-
    references the profile's declared modalities with what is on disk to show
    which raw channels are present or missing and where each lives, plus any
    registered preprocessed channels.  Per-frame facts (counts, shapes) are
    intentionally *not* here -- read them from the loaded dataset
    (``len(ds)``, ``ds[i].data[key].shape``).

    Args:
        sequence_id: Optional identifier used as the printed display label
            only. Channel availability is dataset-wide.

    Returns:
        ``{"class", "name", "root", "layout": {"fixed": [...]},
        "sequences": [...], "splits": [...], "calibration": [...],
        "raw": {"present": [...], "missing": [...], "channels": {key: {...}}},
        "preprocess": {key: meta}}``

    Example::

        ds = Rellis3DDataset("/data/RELLIS")
        ds.describe("00000")
    """
    info = self._structure()
    label = sequence_id if sequence_id is not None else info["name"]
    print(f"\n{info['class']} -- {label}")
    print("─" * 50)
    print("Raw channels")
    if info["raw"]["present"]:
        print("  present  :", ", ".join(info["raw"]["present"]))
    if info["raw"]["missing"]:
        print("  missing  :", ", ".join(info["raw"]["missing"]))
    if not info["raw"]["present"] and not info["raw"]["missing"]:
        print("  (none)")
    print("Preprocessed channels")
    if info["preprocess"]:
        for key, meta in sorted(info["preprocess"].items()):
            ts_info = (
                f"<- timestamps from {meta['timestamps_from']}"
                if "timestamps_from" in meta
                else "<- own timestamps"
            )
            src_info = (
                f"  sources: {meta['sources']}" if meta.get("sources") else ""
            )
            print(f"  {key:<20} {meta['loader']:<6} {ts_info}{src_info}")
    else:
        print("  (none)")
    print()
    return info

split

split(name: str) -> ProfiledDataset

Return a new dataset instance filtered to the named split.

This re-instantiates the dataset, so it starts with a clean pipeline: any registered transform() is intentionally NOT carried over. To split mid-chain while keeping transforms, use :meth:filter_split instead.

Source code in apairo/core/profiled_dataset.py

def split(self, name: str) -> ProfiledDataset:
    """Return a new dataset *instance* filtered to the named split.

    This re-instantiates the dataset, so it starts with a clean pipeline: any
    registered ``transform()`` is intentionally NOT carried over. To split
    mid-chain while keeping transforms, use :meth:`filter_split` instead."""
    return type(self)(
        self._root,
        keys=list(self._keys),
        split=name,
        sequences=list(self._sequence_ids_filter)
        if self._sequence_ids_filter
        else None,
    )

filter_split

filter_split(name: str) -> AbstractDataset

Return a FilteredView restricted to the named predefined split.

Applies the split without re-instantiating the dataset — registered transforms are preserved. Replaces :meth:split for use mid-chain::

ds.transform("lidar", RobotFilter())
ds_train = ds.filter_split("train")  # transforms kept

Works for both LST-based splits (SemanticKITTI/Rellis) and directory-layer splits (GOOSE), so there is a transform-preserving split for every dataset.

Source code in apairo/core/profiled_dataset.py

def filter_split(self, name: str) -> AbstractDataset:
    """Return a FilteredView restricted to the named predefined split.

    Applies the split without re-instantiating the dataset — registered
    transforms are preserved.  Replaces :meth:`split` for use mid-chain::

        ds.transform("lidar", RobotFilter())
        ds_train = ds.filter_split("train")  # transforms kept

    Works for both LST-based splits (SemanticKITTI/Rellis) and directory-layer
    splits (GOOSE), so there is a transform-preserving split for every dataset.
    """
    if self._splits_spec is not None and self._splits_spec.type == "lst":
        frame_set = self._lst_frame_filter(name)
    else:
        frame_set = self._split_layer_frame_filter(name)
    return _apply_lst_filter(self, frame_set)

---

SynchronousDataset

apairo.core.synchronous_dataset.SynchronousDataset

Bases: AbstractDataset

Base class for datasets where index i returns a complete synchronous frame.

All modalities at index i are co-captured -- one row is one sample across every channel, no interleaving. That co-capture is what synchronous means; it is independent of whether the frame carries a timestamp. Random access and standard PyTorch DataLoader shuffling work without any additional wrappers.

A synchronous dataset may still expose a shared per-frame clock in :attr:timestamps (one entry per global frame, so sample.timestamp is that frame's tick); it defaults to None -- clockless -- and subclasses populate it when the frames carry a timestamp.

Subclasses must implement __len__ and _load.

For new synchronous datasets, prefer extending :class:~apairo.core.profiled_dataset.ProfiledDataset with a YAML profile rather than subclassing this directly.

Attributes:

Name	Type	Description
synchronous		Always True -- marks the co-captured (structural) family.
timestamps	dict | ndarray | None	The shared per-frame clock array, or None when clockless.

Source code in apairo/core/synchronous_dataset.py

class SynchronousDataset(AbstractDataset):
    """Base class for datasets where index ``i`` returns a complete synchronous frame.

    All modalities at index ``i`` are co-captured -- one row is one sample across
    every channel, no interleaving. That co-capture is what *synchronous* means;
    it is independent of whether the frame carries a timestamp. Random access and
    standard PyTorch ``DataLoader`` shuffling work without any additional wrappers.

    A synchronous dataset may still expose a **shared per-frame clock** in
    :attr:`timestamps` (one entry per global frame, so ``sample.timestamp`` is
    that frame's tick); it defaults to ``None`` -- clockless -- and subclasses
    populate it when the frames carry a timestamp.

    Subclasses must implement ``__len__`` and ``_load``.

    For new synchronous datasets, prefer extending
    :class:`~apairo.core.profiled_dataset.ProfiledDataset` with a YAML profile
    rather than subclassing this directly.

    Attributes:
        synchronous: Always ``True`` -- marks the co-captured (structural) family.
        timestamps: The shared per-frame clock array, or ``None`` when clockless.
    """

    synchronous = True
    timestamps: dict | np.ndarray | None = None

    # Provided by the concrete dataset (see ProfiledDataset.__init__).
    _root: Path
    _files: dict[str, list[Path]]

    @property
    def root_dir(self) -> Path:
        return self._root

    def _seq_root(self, path: Path) -> Path:
        """Return the sequence root directory for a native file path.

        Datasets with deeper file structures (e.g. seq/lidar/scan/file.bin)
        should override this to go up the correct number of levels.
        Default: path.parent.parent (one modality directory deep).
        """
        return path.parent.parent

    def derived_path(self, idx: int, key: str, ext: str) -> Path:
        ref = next(iter(self._files.values()))[idx]
        return self._seq_root(ref) / key / f"{ref.stem}.{ext}"

    @abstractmethod
    def __len__(self) -> int: ...

---

SemanticKittiDataset

apairo.dataset.semantic_kitti.SemanticKittiDataset

Bases: ProfiledDataset

SemanticKITTI dataset -- driving LiDAR with dense semantic labels.

Keys: lidar (float32, shape (N, 4)), labels (int64, lower 16 bits = semantic class).

Example::

ds = SemanticKittiDataset("/data/kitti/dataset", keys=["lidar", "labels"])
sample = ds[0]
# sample.data["lidar"]  -> np.ndarray (N, 4)
# sample.data["labels"] -> np.ndarray (N,)

Source code in apairo/dataset/semantic_kitti/dataset.py

class SemanticKittiDataset(ProfiledDataset):
    """SemanticKITTI dataset -- driving LiDAR with dense semantic labels.

    Keys: ``lidar`` (float32, shape (N, 4)), ``labels`` (int64, lower 16 bits = semantic class).

    Example::

        ds = SemanticKittiDataset("/data/kitti/dataset", keys=["lidar", "labels"])
        sample = ds[0]
        # sample.data["lidar"]  -> np.ndarray (N, 4)
        # sample.data["labels"] -> np.ndarray (N,)
    """

    _profile = "semantic_kitti.yaml"

---

Goose3DDataset

apairo.dataset.goose.Goose3DDataset

Bases: ProfiledDataset

GOOSE 3D dataset -- outdoor off-road LiDAR with traversability labels.

Keys: lidar (float32, shape (N, 4)), labels (int64). Split: "train", "val", or "test".

Example::

ds = Goose3DDataset("/data/GOOSE_3D", keys=["lidar", "labels"], split="train")
sample = ds[0]
# sample.data["lidar"]  -> np.ndarray (N, 4)
# sample.data["labels"] -> np.ndarray (N,)

Source code in apairo/dataset/goose/dataset.py

class Goose3DDataset(ProfiledDataset):
    """GOOSE 3D dataset -- outdoor off-road LiDAR with traversability labels.

    Keys: ``lidar`` (float32, shape (N, 4)), ``labels`` (int64).
    Split: ``"train"``, ``"val"``, or ``"test"``.

    Example::

        ds = Goose3DDataset("/data/GOOSE_3D", keys=["lidar", "labels"], split="train")
        sample = ds[0]
        # sample.data["lidar"]  -> np.ndarray (N, 4)
        # sample.data["labels"] -> np.ndarray (N,)
    """

    _profile = "goose.yaml"

---

Rellis3DDataset

apairo.dataset.rellis.Rellis3DDataset

Bases: ProfiledDataset

RELLIS-3D dataset -- off-road LiDAR with semantic labels.

Keys: lidar (float32, shape (N, 4)), labels (int64). Optional: poses (float64, shape (3, 4)) -- one 3x4 pose matrix per frame, loaded from a per-sequence poses.txt file (one row of 12 floats per frame).

Example::

ds = Rellis3DDataset("/data/RELLIS", keys=["lidar", "labels"])
sample = ds[0]
# sample.data["lidar"]  -> np.ndarray (N, 4)
# sample.data["labels"] -> np.ndarray (N,)

ds = Rellis3DDataset("/data/RELLIS", keys=["lidar", "poses"])
# sample.data["poses"]  -> np.ndarray (3, 4)

Source code in apairo/dataset/rellis/dataset.py

class Rellis3DDataset(ProfiledDataset):
    """RELLIS-3D dataset -- off-road LiDAR with semantic labels.

    Keys: ``lidar`` (float32, shape (N, 4)), ``labels`` (int64).
    Optional: ``poses`` (float64, shape (3, 4)) -- one 3x4 pose matrix per frame,
    loaded from a per-sequence ``poses.txt`` file (one row of 12 floats per frame).

    Example::

        ds = Rellis3DDataset("/data/RELLIS", keys=["lidar", "labels"])
        sample = ds[0]
        # sample.data["lidar"]  -> np.ndarray (N, 4)
        # sample.data["labels"] -> np.ndarray (N,)

        ds = Rellis3DDataset("/data/RELLIS", keys=["lidar", "poses"])
        # sample.data["poses"]  -> np.ndarray (3, 4)
    """

    _profile = "rellis.yaml"

---

Asynchronous datasets

RawDataset

apairo.dataset.raw.RawDataset

Bases: RootSequenceMixin, AsyncLayoutDataset, ConfigurableDataset

Generic channels.yaml-driven dataset; single sequence or dataset root.

Parameters:

Name	Type	Description	Default
directory	str | Path	A sequence directory or a dataset root directory -- auto-detected. A directory with no .apairo is bootstrapped on load (loaders inferred from file extensions), so raw data works with no manual :meth:init.	required
keys	list[str] | None	Channels to load. None -> every channel declared in channels.yaml.	None
declare	str | Path | None	Path to a declaration file overlaid onto the channel metadata (per channel, per field) -- see :class:~apairo.dataset.async_layout.AsyncLayoutDataset. On a root, the declaration applies to every sequence.	None

Example::

ds = RawDataset(root, keys=["lidar", "imu"])  # whole dataset
ds.run_preprocess(MyLabeler())                # persisted as a new channel

Source code in apairo/dataset/raw/dataset.py

class RawDataset(RootSequenceMixin, AsyncLayoutDataset, ConfigurableDataset):
    r"""Generic ``channels.yaml``-driven dataset; single sequence or dataset root.

    Args:
        directory: A sequence directory or a dataset root directory --
            auto-detected. A directory with no ``.apairo`` is bootstrapped on
            load (loaders inferred from file extensions), so raw data works with
            no manual :meth:`init`.
        keys: Channels to load. ``None`` -> every channel declared in
            ``channels.yaml``.
        declare: Path to a declaration file overlaid onto the channel metadata
            (per channel, per field) -- see
            :class:`~apairo.dataset.async_layout.AsyncLayoutDataset`. On a
            root, the declaration applies to every sequence.

    Example::

        ds = RawDataset(root, keys=["lidar", "imu"])  # whole dataset
        ds.run_preprocess(MyLabeler())                # persisted as a new channel
    """

    def __init__(
        self,
        directory: str | Path,
        keys: list[str] | None = None,
        declare: str | Path | None = None,
        declare_base: str | Path | None = None,
    ) -> None:
        path = Path(directory)
        # Consulted by _bootstrap_config (which runs before super().__init__).
        self._declare = declare
        self._declare_base = declare_base

        # A sequence loads directly; a bare channel layout (no .apairo) is
        # bootstrapped on the spot, so raw data needs no manual init(). A
        # declaration (in-tree or declare=) also marks a sequence, but only
        # after root detection, so a root carrying a stray apairo.yaml at its
        # top still loads as a root.
        if config_exists(path) or self._is_sequence_layout(path):
            is_sequence = True
        elif _is_dataset_root(path):
            is_sequence = False
        elif declaration_exists(path) or declare is not None:
            is_sequence = True
        else:
            raise FileNotFoundError(
                f"'{path}' is neither a sequence (no recognizable channel "
                f"sub-directories) nor a dataset root (no sequence sub-directories)."
            )

        if is_sequence:
            self._is_root = False
            self._sequence_dir = path
            self._name = path.name
            # A sequence opened standalone inherits its root's apairo.yaml --
            # same contract whether entered from the root or directly.
            if declare_base is None:
                declare_base = inherited_declaration(path)
                self._declare_base = declare_base
            if not config_exists(path):
                self._load_or_create_config(path)
            super().__init__(
                path, keys=keys, declare=declare, declare_base=declare_base
            )
        else:
            self._init_raw_root(path, keys, declare)

    # ------------------------------------------------------------------ init

    @classmethod
    def init(  # type: ignore[override]  # root-aware variant, intentionally different keywords
        cls,
        directory: str | Path,
        *,
        merge: bool = False,
        overwrite: bool = False,
        name: str | None = None,
        declare: str | Path | None = None,
    ) -> Path:
        """Write the ``.apairo`` sidecar(s) by scanning *directory*. Root-aware.

        A **sequence** directory (its sub-directories hold data files) gets a
        ``.apairo/channels.yaml`` with loaders inferred per channel. A **root**
        directory (its sub-directories are sequences) gets each sequence
        initialised, then a ``.apairo/dataset.yaml`` manifest (name + sequence
        order + channel union).

        Args:
            directory: Sequence or dataset-root directory (auto-detected).
            merge: Add newly detected channels without touching existing ones.
            overwrite: Discard existing ``.apairo`` and rebuild from scratch.
                Never touches an ``apairo.yaml`` declaration.
            name: Dataset name for the root manifest (default: directory name).
            declare: External declaration file the scan should respect (the
                in-tree ``apairo.yaml`` is always read). On a root it applies
                to every sequence.

        Returns:
            Path of the file written -- ``channels.yaml`` for a sequence, or
            ``dataset.yaml`` for a root.
        """
        path = Path(directory)
        if declare is None:
            # The effective declaration drives the scan: a root's own
            # apairo.yaml for its sequences, the parent's for a sequence
            # initialised standalone.
            if cls._is_sequence_layout(path):
                declare = inherited_declaration(path)
            else:
                declare = declaration_path(path) if declaration_exists(path) else None

        if cls._is_sequence_layout(path):
            AsyncLayoutDataset.init(
                path, overwrite=overwrite, merge=merge, declare=declare
            )
            return path / CONFIG_DIR / CHANNELS_FILE

        seq_dirs: list[Path] = []
        for d in sorted(path.iterdir()):
            if not d.is_dir() or d.name.startswith("."):
                continue
            if cls._is_sequence_layout(d):
                try:
                    AsyncLayoutDataset.init(
                        d, overwrite=overwrite, merge=merge, declare=declare
                    )
                except (FileExistsError, ValueError):
                    # Already initialised (no overwrite/merge), or merge found
                    # nothing new -- either way the sequence is ready. Idempotent.
                    pass
                seq_dirs.append(d)
            elif config_exists(d):
                seq_dirs.append(d)

        if not seq_dirs:
            raise FileNotFoundError(
                f"'{path}' has no channels and no sequence sub-directories to "
                f"initialise. Point init at a sequence or a dataset root."
            )
        return cls._write_manifest(path, name=name)

    @staticmethod
    def _is_sequence_layout(path: Path) -> bool:
        """True when *path*'s own sub-directories include a recognizable channel."""
        return any(
            _detect_loader(d) is not None
            for d in path.iterdir()
            if d.is_dir() and not d.name.startswith(".")
        )

    @classmethod
    def _write_manifest(cls, root: str | Path, *, name: str | None = None) -> Path:
        """(Re)write ``<root>/.apairo/dataset.yaml`` from the sequences on disk."""
        root = Path(root)
        sequences = sorted(
            d.name
            for d in root.iterdir()
            if d.is_dir() and not d.name.startswith(".") and config_exists(d)
        )
        channels: dict = {}
        for seq in sequences:
            for key, meta in read_config(root / seq).get("channels", {}).items():
                channels.setdefault(key, {"kind": meta.get("kind", "raw")})

        manifest = {
            "version": 1,
            "name": name or root.name,
            "sequences": sequences,
            "channels": channels,
        }
        apairo_dir = root / CONFIG_DIR
        apairo_dir.mkdir(exist_ok=True)
        path = apairo_dir / _MANIFEST_FILE
        with open(path, "w") as f:
            yaml.dump(manifest, f, default_flow_style=False, sort_keys=True)
        return path

    # ------------------------------------------------------------------ root

    def _init_raw_root(
        self,
        root: Path,
        keys: list[str] | None,
        declare: str | Path | None = None,
    ) -> None:
        manifest = _read_manifest(root)
        self._name = manifest.get("name", root.name)

        def is_seq(d: Path) -> bool:  # initialised, or a bare layout to bootstrap
            return config_exists(d) or self._is_sequence_layout(d)

        # Sequence order: manifest order if given, else sorted discovery.
        if manifest.get("sequences"):
            seq_dirs = [root / s for s in manifest["sequences"] if is_seq(root / s)]
        else:
            seq_dirs = sorted(
                d
                for d in root.iterdir()
                if d.is_dir() and not d.name.startswith(".") and is_seq(d)
            )
        if not seq_dirs:
            raise FileNotFoundError(f"No sequences found under '{root}'.")

        # The root's own apairo.yaml propagates to every sequence, below the
        # sequence's own file; an explicit declare= keeps the top slot.
        base = declaration_path(root) if declaration_exists(root) else None

        # type(self) so a profiled subclass (e.g. TartanKittiDataset) builds
        # sequences of its own kind, keeping its channel profile.
        super()._init_root(
            root,
            seq_dirs,
            lambda d: type(self)(d, keys=keys, declare=declare, declare_base=base),
            build_index=True,
        )

    # ------------------------------------------------------------------ hooks

    def _single_available(self) -> frozenset:
        # Declared channels absent from this sequence (a root's union
        # declaration) are not available here -- intersect with what is on disk.
        return frozenset(self._profile) & frozenset(self._files)

    def _set_single_keys(self, keys) -> None:
        if keys == "all":
            keys = sorted(self._profile)
        # Delegate to the layout base's setter (validates + re-inits loaders).
        AsyncLayoutDataset.keys.fset(self, list(keys))  # type: ignore[attr-defined]  # class-level property object

    # ------------------------------------------------------------------ public

    @property
    def name(self) -> str:
        """Dataset name (manifest ``name``, else the directory name)."""
        return self._name

    # ------------------------------------------------------------------ helpers

    def derived_path(self, idx: int, key: str, ext: str) -> Path:
        # On a root, route the global index to the sub-sequence it belongs to
        # (each sub-sequence has its own ``_sequence_dir``); a root has none.
        if self._is_root:
            seq_idx, local_idx = self._locate(idx)
            return self._sequences[seq_idx].derived_path(local_idx, key, ext)
        return self._sequence_dir / key / f"{idx:06d}.{ext}"

    def _bootstrap_config(self, sequence_dir: Path) -> dict:
        """ConfigurableDataset hook: detect raw channels when .apairo is absent."""
        explained = _declared_key_channels(
            sequence_dir,
            getattr(self, "_declare_base", None),
            getattr(self, "_declare", None),
        )
        channels: dict = {}
        for key in sorted(get_files(str(sequence_dir))):
            channel_dir = Path(sequence_dir) / key
            loader = _detect_loader(channel_dir)
            if loader is None:
                continue
            channels[key] = {"loader": loader, "kind": "raw"}
            # A declared key/order regex explains this channel's stems (their
            # '_' is part of the name, not a suffix) -- don't fan them out
            # into suffixed sub-channels.
            if key in explained:
                continue
            for suffix, frag in _suffix_channel_entries(channel_dir, loader).items():
                channels[f"{key}_{suffix}"] = {"kind": "raw", **frag}
        return {"version": 1, "channels": channels}

name property

name: str

Dataset name (manifest name, else the directory name).

init classmethod

init(directory: str | Path, *, merge: bool = False, overwrite: bool = False, name: str | None = None, declare: str | Path | None = None) -> Path

Write the .apairo sidecar(s) by scanning directory. Root-aware.

A sequence directory (its sub-directories hold data files) gets a .apairo/channels.yaml with loaders inferred per channel. A root directory (its sub-directories are sequences) gets each sequence initialised, then a .apairo/dataset.yaml manifest (name + sequence order + channel union).

Parameters:

Name	Type	Description	Default
directory	str | Path	Sequence or dataset-root directory (auto-detected).	required
merge	bool	Add newly detected channels without touching existing ones.	False
overwrite	bool	Discard existing .apairo and rebuild from scratch. Never touches an apairo.yaml declaration.	False
name	str | None	Dataset name for the root manifest (default: directory name).	None
declare	str | Path | None	External declaration file the scan should respect (the in-tree apairo.yaml is always read). On a root it applies to every sequence.	None

Returns:

Type	Description
Path	Path of the file written -- channels.yaml for a sequence, or
Path	dataset.yaml for a root.

Source code in apairo/dataset/raw/dataset.py

@classmethod
def init(  # type: ignore[override]  # root-aware variant, intentionally different keywords
    cls,
    directory: str | Path,
    *,
    merge: bool = False,
    overwrite: bool = False,
    name: str | None = None,
    declare: str | Path | None = None,
) -> Path:
    """Write the ``.apairo`` sidecar(s) by scanning *directory*. Root-aware.

    A **sequence** directory (its sub-directories hold data files) gets a
    ``.apairo/channels.yaml`` with loaders inferred per channel. A **root**
    directory (its sub-directories are sequences) gets each sequence
    initialised, then a ``.apairo/dataset.yaml`` manifest (name + sequence
    order + channel union).

    Args:
        directory: Sequence or dataset-root directory (auto-detected).
        merge: Add newly detected channels without touching existing ones.
        overwrite: Discard existing ``.apairo`` and rebuild from scratch.
            Never touches an ``apairo.yaml`` declaration.
        name: Dataset name for the root manifest (default: directory name).
        declare: External declaration file the scan should respect (the
            in-tree ``apairo.yaml`` is always read). On a root it applies
            to every sequence.

    Returns:
        Path of the file written -- ``channels.yaml`` for a sequence, or
        ``dataset.yaml`` for a root.
    """
    path = Path(directory)
    if declare is None:
        # The effective declaration drives the scan: a root's own
        # apairo.yaml for its sequences, the parent's for a sequence
        # initialised standalone.
        if cls._is_sequence_layout(path):
            declare = inherited_declaration(path)
        else:
            declare = declaration_path(path) if declaration_exists(path) else None

    if cls._is_sequence_layout(path):
        AsyncLayoutDataset.init(
            path, overwrite=overwrite, merge=merge, declare=declare
        )
        return path / CONFIG_DIR / CHANNELS_FILE

    seq_dirs: list[Path] = []
    for d in sorted(path.iterdir()):
        if not d.is_dir() or d.name.startswith("."):
            continue
        if cls._is_sequence_layout(d):
            try:
                AsyncLayoutDataset.init(
                    d, overwrite=overwrite, merge=merge, declare=declare
                )
            except (FileExistsError, ValueError):
                # Already initialised (no overwrite/merge), or merge found
                # nothing new -- either way the sequence is ready. Idempotent.
                pass
            seq_dirs.append(d)
        elif config_exists(d):
            seq_dirs.append(d)

    if not seq_dirs:
        raise FileNotFoundError(
            f"'{path}' has no channels and no sequence sub-directories to "
            f"initialise. Point init at a sequence or a dataset root."
        )
    return cls._write_manifest(path, name=name)

---

TartanKittiDataset

apairo.dataset.tartan_kitti.TartanKittiDataset

Bases: RawDataset

TartanDrive v2 -- a :class:~apairo.dataset.raw.RawDataset whose channel set is the fixed TartanDrive profile.

Loading, multi-sequence roots, synchronization and preprocessing are exactly :class:RawDataset's; this class only pins the canonical TartanDrive channels (profile.yaml). That means a raw sequence with no .apairo bootstraps with the profile's loaders rather than ones merely guessed from file extensions, and :meth:describe can flag profile channels missing on disk.

Single sequence or root directory, auto-detected::

ds = TartanKittiDataset(seq_dir, keys=["velodyne_0", "cmd"])
ds = TartanKittiDataset(root_dir, keys=["velodyne_0"])   # all sequences

Source code in apairo/dataset/tartan_kitti/dataset.py

class TartanKittiDataset(RawDataset):
    r"""TartanDrive v2 -- a :class:`~apairo.dataset.raw.RawDataset` whose channel
    set is the fixed TartanDrive profile.

    Loading, multi-sequence roots, synchronization and preprocessing are exactly
    :class:`RawDataset`'s; this class only pins the canonical TartanDrive
    channels (``profile.yaml``). That means a raw sequence with no ``.apairo``
    bootstraps with the *profile's* loaders rather than ones merely guessed from
    file extensions, and :meth:`describe` can flag profile channels missing on
    disk.

    Single sequence or root directory, auto-detected::

        ds = TartanKittiDataset(seq_dir, keys=["velodyne_0", "cmd"])
        ds = TartanKittiDataset(root_dir, keys=["velodyne_0"])   # all sequences
    """

    available_keys: ClassVar[frozenset] = frozenset(_PROFILE)

    def _bootstrap_config(self, sequence_dir: Path) -> dict:
        """Declare the on-disk profile channels, pinning each loader from the
        TartanDrive profile instead of inferring it from file extensions.

        A ``"npys"`` channel may also hold suffixed sub-channel variants (e.g.
        ``velodyne_0/000000_intensity.npy``); those fan out into sibling
        channels (``velodyne_0_intensity``) the same way as an unprofiled
        :class:`~apairo.dataset.raw.RawDataset`.
        """
        channels: dict = {}
        for key in sorted(get_files(str(sequence_dir))):
            if key not in _PROFILE:
                continue
            loader = _PROFILE[key]
            channels[key] = {"loader": loader, "kind": "raw"}
            channel_dir = Path(sequence_dir) / key
            for suffix, frag in _suffix_channel_entries(channel_dir, loader).items():
                channels[f"{key}_{suffix}"] = {"kind": "raw", **frag}
        return {"version": 1, "channels": channels}

---

AsyncLayoutDataset

Abstract per-channel layout base for the asynchronous family. Internal base class — subclass it to add a fixed-channel async dataset; end users load through RawDataset or TartanKittiDataset.

apairo.dataset.async_layout.AsyncLayoutDataset

Bases: AbstractDataset

Abstract asynchronous layout loader (one subdirectory per channel).

This is the format primitive of the asynchronous dataset family. It is not a concrete dataset -- the synchronous KITTI-style datasets are :class:~apairo.core.profiled_dataset.ProfiledDataset subclasses (e.g. :class:~apairo.dataset.semantic_kitti.SemanticKittiDataset).

It describes how channels are stored, never which channels exist: each channel is a subdirectory with its own timestamps.txt and data files in a format known to the loader registry (npys, npy, bin, img, zarr). A channel may instead carry its alignment key in its filenames -- a key: {name: <regex>} / {file: <name>} spec parses it in memory at read time (nothing written), with an optional order enumeration policy; see docs/datasets/bring-your-own-dataset.md. The set of channels is per-instance state, read from .apairo/channels.yaml (or an explicit dataset_profile). Datasets with a fixed channel set layer a profile on top (e.g. :class:~apairo.dataset.tartan_kitti.TartanKittiDataset); datasets with dynamic channels (e.g. apairo-extractor output) use :class:~apairo.dataset.raw.RawDataset, which reads the channel set from .apairo with no profile.

Usage with an explicit profile (original API)::

ds = AsyncLayoutDataset(seq_dir, keys=["lidar", "cam"], dataset_profile="my.yaml")

Usage with .apairo (after :meth:init has been called)::

AsyncLayoutDataset.init(seq_dir)          # once, auto-detects channels
ds = AsyncLayoutDataset(seq_dir)          # keys and loaders come from .apairo
ds = AsyncLayoutDataset(seq_dir, keys=["lidar"])  # restrict to a subset

Parameters:

Name	Type	Description	Default
directory	str | Path	Path to the dataset root / sequence directory.	required
keys	list[str] | None	Modality names to load. None → all channels declared in .apairo (requires .apairo to exist).	None
dataset_profile	str | Path | None	YAML profile filename or absolute Path mapping keys to loader types. None → loaders are read from .apairo (requires .apairo to exist).	None
declare	str | Path | None	Path to a declaration file (the channels.yaml schema minus machine provenance) overlaid onto the channel metadata, per channel and per field. Precedence: declare= > <directory>/apairo.yaml declare_base > .apairo/channels.yaml. Lets a read-only tree be declared from outside; apairo never writes a declaration.	None
declare_base	str | Path | None	Lower-precedence declaration overlaid before the in-tree apairo.yaml -- how a dataset root propagates its own <root>/apairo.yaml to every sequence (the sequence's own file stays the more specific word). Rarely passed by hand.	None

Source code in apairo/dataset/async_layout/dataset.py

class AsyncLayoutDataset(AbstractDataset):
    r"""Abstract *asynchronous layout* loader (one subdirectory per channel).

    This is the format primitive of the asynchronous dataset family. It is not
    a concrete dataset -- the synchronous KITTI-style datasets are
    :class:`~apairo.core.profiled_dataset.ProfiledDataset` subclasses (e.g.
    :class:`~apairo.dataset.semantic_kitti.SemanticKittiDataset`).

    It describes *how* channels are stored, never *which* channels exist: each
    channel is a subdirectory with its own ``timestamps.txt`` and data files in
    a format known to the loader registry (``npys``, ``npy``, ``bin``, ``img``,
    ``zarr``). A channel may instead carry its alignment key in its filenames --
    a ``key: {name: <regex>}`` / ``{file: <name>}`` spec parses it in memory at
    read time (nothing written), with an optional ``order`` enumeration policy;
    see ``docs/datasets/bring-your-own-dataset.md``. The set of channels is
    per-instance state, read from ``.apairo/channels.yaml`` (or an explicit
    ``dataset_profile``). Datasets
    with a *fixed* channel set layer a profile on top (e.g.
    :class:`~apairo.dataset.tartan_kitti.TartanKittiDataset`); datasets with
    *dynamic* channels (e.g. ``apairo-extractor`` output) use
    :class:`~apairo.dataset.raw.RawDataset`, which reads the channel set from
    ``.apairo`` with no profile.

    **Usage with an explicit profile (original API)**::

        ds = AsyncLayoutDataset(seq_dir, keys=["lidar", "cam"], dataset_profile="my.yaml")

    **Usage with** ``.apairo`` **(after** :meth:`init` **has been called)**::

        AsyncLayoutDataset.init(seq_dir)          # once, auto-detects channels
        ds = AsyncLayoutDataset(seq_dir)          # keys and loaders come from .apairo
        ds = AsyncLayoutDataset(seq_dir, keys=["lidar"])  # restrict to a subset

    Args:
        directory: Path to the dataset root / sequence directory.
        keys: Modality names to load.  ``None`` → all channels declared in
            ``.apairo`` (requires ``.apairo`` to exist).
        dataset_profile: YAML profile filename **or** absolute Path mapping keys
            to loader types.  ``None`` → loaders are read from ``.apairo``
            (requires ``.apairo`` to exist).
        declare: Path to a declaration file (the ``channels.yaml`` schema minus
            machine provenance) overlaid onto the channel metadata, per channel
            and per field.  Precedence: ``declare=`` > ``<directory>/apairo.yaml``
            > ``declare_base`` > ``.apairo/channels.yaml``.  Lets a read-only
            tree be declared from outside; apairo never writes a declaration.
        declare_base: Lower-precedence declaration overlaid *before* the
            in-tree ``apairo.yaml`` -- how a dataset root propagates its own
            ``<root>/apairo.yaml`` to every sequence (the sequence's own file
            stays the more specific word). Rarely passed by hand.
    """

    synchronous: bool = False

    def __init__(
        self,
        directory: str | Path,
        keys: list[str] | None = None,
        dataset_profile: str | Path | None = None,
        declare: str | Path | None = None,
        declare_base: str | Path | None = None,
    ) -> None:
        directory = Path(directory)
        keys_defaulted = False  # True when keys=None resolved to "everything"

        # Channel metadata from .apairo (empty when a dataset_profile is passed
        # and no sidecar exists). alias_of maps an on-disk directory name to the
        # public name it is exposed under; timestamp_aliases maps a channel to the
        # one it borrows its clock from (its `timestamps_from`). Everything below
        # is keyed by the public name; the directory name only locates files.
        # _config_fallback is set by ConfigurableDataset when the directory is
        # read-only and the bootstrapped sidecar could not be written.
        fallback = getattr(self, "_config_fallback", None)
        if fallback is not None:
            channels = fallback.get("channels", {})
        elif config_exists(directory):
            channels = read_config(directory).get("channels", {})
        else:
            channels = {}
        # Human declarations overlay the machine registry, per channel and per
        # field, least specific first: the root's propagated declaration, the
        # sequence's own apairo.yaml, then an explicit declare= file.
        if declare_base is not None:
            channels = merge_declared_channels(channels, read_declaration(declare_base))
        if declaration_exists(directory):
            channels = merge_declared_channels(
                channels, read_declaration(declaration_path(directory))
            )
        if declare is not None:
            channels = merge_declared_channels(channels, read_declaration(declare))
        self._alias_of: dict[str, str] = {
            k: v["alias"] for k, v in channels.items() if v.get("alias")
        }
        # Honour the request's language: a channel explicitly asked for by its
        # real (directory) name is exposed under that name, one asked for by its
        # alias under the alias. Rewriting the alias table up front keeps every
        # table below in the request's vocabulary.
        if keys is not None:
            to_real = {alias: real for real, alias in self._alias_of.items()}
            for k in keys:
                real = to_real.get(k, k)
                if real in self._alias_of:
                    self._alias_of[real] = k
        self._timestamp_aliases: dict[str, str] = {
            self._public(k): self._resolve_key(v["timestamps_from"])
            for k, v in channels.items()
            if v.get("timestamps_from")
        }
        # A suffixed sub-channel (e.g. velodyne_0_intensity) has no directory of
        # its own -- it reads a suffix-filtered subset of another channel's files.
        self._suffix_of: dict[str, str] = {
            self._public(k): v["suffix"] for k, v in channels.items() if v.get("suffix")
        }
        # A channel may declare that its alignment key is parsed from its own
        # filenames (e.g. Rellis camera frame<N>-<epoch>_<ms>.jpg) instead of a
        # timestamps.txt -- the key is then computed in memory, nothing is written.
        self._key_spec: dict[str, dict] = {
            self._public(k): v["key"] for k, v in channels.items() if v.get("key")
        }
        # ...and how those files are enumerated/ordered, when the default frame-file
        # convention doesn't fit its naming. Separate from `key`; when absent it
        # defaults to the key's own regex.
        self._order_spec: dict[str, dict] = {
            self._public(k): v["order"] for k, v in channels.items() if v.get("order")
        }
        # A stacked `npy` channel may name the exact `.npy` it loads when its
        # directory colocates several (poses.npy beside valid_mask.npy) -- the
        # whole-array analogue of `suffix`. The name lives here, in the layout.
        self._array_file_of: dict[str, str] = {
            self._public(k): safe_config_name(
                v["array_file"], label=f"channel '{k}' array_file"
            )
            for k, v in channels.items()
            if v.get("array_file")
        }
        # A `pcd` channel's field contract. A PCD header is self-describing, so
        # the field set is a per-file property; declaring it here is what makes
        # the channel's width a layout decision instead of a per-file accident.
        self._fields_of: dict[str, list[str]] = {
            self._public(k): list(v["fields"])
            for k, v in channels.items()
            if v.get("fields")
        }

        if dataset_profile is not None:
            self._profile: dict[str, str] = load_profile(dataset_profile)
        elif channels:
            self._profile = {
                self._public(k): v["loader"]
                for k, v in channels.items()
                if "loader" in v
            }
            if keys is None:
                keys = sorted(self._profile.keys())
                keys_defaulted = True
        else:
            raise FileNotFoundError(
                f"No dataset_profile given and no .apairo or apairo.yaml found "
                f"in '{directory}'. Either pass dataset_profile=... or "
                f"declare=..., write an apairo.yaml declaration, or initialize "
                f"with {type(self).__name__}.init('{directory}')."
            )

        if keys is None:
            raise ValueError(
                "keys must be specified when dataset_profile is given. "
                "Pass keys=[...] or use .apairo (call init() first)."
            )

        # Re-key the on-disk directories by their public name so a request, a
        # loader and a sample all speak the same (aliased) language.
        self._files: dict[str, str] = {}
        for real, path in get_files(str(directory)).items():
            public = self._public(real)
            if public in self._files:
                raise ValueError(
                    f"Alias collision in '{directory}': the public name '{public}' "
                    f"is claimed by more than one channel. Clear one alias with "
                    f"`apairo alias <channel> --remove` (see `apairo status`)."
                )
            self._files[public] = path

        # A plain channel whose `directory` names where its files live, when
        # that is not the channel's own key: another top-level directory, or a
        # nested relative path (per_object_gt/pcd -- an annotation tool's
        # export). The explicit declaration wins over the same-name scan.
        for real, meta in channels.items():
            sub = meta.get("directory")
            if not sub or meta.get("suffix") or meta.get("array_file"):
                continue
            resolved = directory / safe_config_name(
                str(sub), label=f"channel '{real}' directory"
            )
            if resolved.is_dir():
                self._files[self._public(real)] = str(resolved)

        # Sub-channels that share another channel's directory instead of owning
        # one: a suffixed per-frame variant (*_intensity.npy), or a colocated
        # stacked array named by `array_file` (valid_mask.npy beside poses.npy).
        # Both read out of the directory named by their "directory" field.
        for real, meta in channels.items():
            if not (meta.get("suffix") or meta.get("array_file")):
                continue
            public = self._public(real)
            source_public = self._public(meta.get("directory", real))
            if source_public in self._files:
                self._files[public] = self._files[source_public]

        keys = [self._resolve_key(k) for k in keys]
        if keys_defaulted:
            # Load-everything mode: a dataset-wide declaration (a root's file
            # is the union over its sequences) may name channels this sequence
            # does not hold -- skip them here instead of failing the sequence.
            # An explicitly requested channel still errors below.
            keys = [k for k in keys if k in self._files]
        missing = set(keys) - set(self._files)
        if missing:
            raise KeyError(f"Keys not found in dataset directory: {missing}")

        self._keys: list[str] = []
        self._set_keys(keys)
        self._init()

    # ------------------------------------------------------------------ alias

    def _public(self, real_name: str) -> str:
        """Public name a directory is exposed under (its alias, else itself)."""
        return self._alias_of.get(real_name, real_name)

    def _resolve_key(self, key: str) -> str:
        """Normalize a requested key (alias *or* real directory name) to its
        public name. Unknown keys pass through unchanged so the usual
        not-found error still fires."""
        if key in self._alias_of:  # a real name that has an alias -> its alias
            return self._alias_of[key]
        return key  # already a public name (an alias, or an unaliased real name)

    @classmethod
    def init(
        cls,
        directory: str | Path,
        *,
        raw_keys: list[str] | None = None,
        overwrite: bool = False,
        merge: bool = False,
        declare: str | Path | None = None,
    ) -> Path:
        """Scan an async-layout directory and write ``.apairo/channels.yaml``.

        All detected subdirectories are registered as raw channels.  Loader
        type is inferred from file extensions:

        * ``.bin`` → ``bin``
        * ``.pcd`` → ``pcd``
        * ``.png`` / ``.jpg`` / … → ``img``
        * multiple ``.npy`` files → ``npys``
        * single ``.npy`` file → ``npy``

        For ambiguous cases (e.g. a single-frame ``.npy`` that is actually
        per-frame), call :func:`~apairo.core.config.register_raw_channel`
        afterwards to override the detected loader.

        Args:
            directory: Dataset root / sequence directory to initialize.
            raw_keys: Subdirectory names to include.  ``None`` → all detected
                subdirectories with recognizable file types.
            overwrite: Discard the existing ``.apairo`` and rebuild from
                scratch.  Incompatible with ``merge``.
            merge: Add newly detected raw channels to an existing ``.apairo``
                without touching channels already declared (raw or
                preprocessed).  If ``.apairo`` does not yet exist, behaves
                like a normal init.  Incompatible with ``overwrite``.
            declare: Path to an external declaration file the scan should
                respect, in addition to the in-tree ``apairo.yaml`` (which is
                always read).  The scan only *reads* declarations -- it writes
                ``.apairo/`` and nothing else.

        Returns:
            Path of the written ``channels.yaml``.

        Raises:
            ValueError: If both ``overwrite`` and ``merge`` are ``True``.
            FileExistsError: If ``.apairo`` already exists and both
                ``overwrite`` and ``merge`` are ``False``.
            ValueError: If no new recognizable channels are found.
        """
        if overwrite and merge:
            raise ValueError("overwrite and merge are mutually exclusive.")

        directory = Path(directory)

        explained = _declared_key_channels(directory, declare)

        if merge and config_exists(directory):
            existing = read_config(directory).get("channels", {})
            added = 0
            for channel_dir in sorted(directory.iterdir()):
                if not channel_dir.is_dir() or channel_dir.name.startswith("."):
                    continue
                if raw_keys is not None and channel_dir.name not in raw_keys:
                    continue
                loader = _detect_loader(channel_dir)
                if loader is None:
                    continue
                if channel_dir.name not in existing:
                    _register_raw_channel(directory, channel_dir.name, loader)
                    added += 1
                if channel_dir.name in explained:
                    continue
                # A directory's base channel may already be registered while a
                # suffix that only appeared later (e.g. *_intensity.npy) is not
                # -- check independently so re-running merge picks it up.
                for suffix, frag in _suffix_channel_entries(
                    channel_dir, loader
                ).items():
                    key = f"{channel_dir.name}_{suffix}"
                    if key in existing:
                        continue
                    _register_raw_channel(
                        directory,
                        key,
                        frag["loader"],
                        directory=frag["directory"],
                        suffix=frag["suffix"],
                    )
                    added += 1
            if added == 0:
                detail = f" (checked: {raw_keys})" if raw_keys else ""
                raise ValueError(
                    f"No new recognizable channels found in '{directory}'{detail}."
                )
            return directory / CONFIG_DIR / CHANNELS_FILE

        if config_exists(directory) and not overwrite:
            raise FileExistsError(
                f".apairo already exists in '{directory}'. "
                f"Pass overwrite=True to reinitialize, or merge=True to add new channels."
            )

        channels: dict = {}
        for channel_dir in sorted(directory.iterdir()):
            if not channel_dir.is_dir() or channel_dir.name.startswith("."):
                continue
            if raw_keys is not None and channel_dir.name not in raw_keys:
                continue
            loader = _detect_loader(channel_dir)
            if loader is None:
                continue
            channels[channel_dir.name] = {
                "kind": "raw",
                "loader": loader,
            }
            if channel_dir.name in explained:
                continue
            for suffix, frag in _suffix_channel_entries(channel_dir, loader).items():
                channels[f"{channel_dir.name}_{suffix}"] = {"kind": "raw", **frag}

        if not channels:
            detail = f" (checked: {raw_keys})" if raw_keys else ""
            raise ValueError(
                f"No recognizable channels found in '{directory}'{detail}. "
                f"Expected subdirectories containing .bin, .pcd, .npy, or image files."
            )

        write_config(directory, {"version": 1, "channels": channels})
        return directory / CONFIG_DIR / CHANNELS_FILE

    # ------------------------------------------------------------------ keys

    @property
    def keys(self) -> list[str]:
        return self._keys

    @keys.setter
    def keys(self, keys: list[str]) -> None:
        keys = [self._resolve_key(k) for k in keys]
        missing = set(keys) - set(self._files)
        if missing:
            raise KeyError(f"Keys not found in dataset directory: {missing}")
        self._set_keys(list(keys))
        self._init()

    # ----------------------------------------------------------------- shape

    @property
    def shape(self) -> dict[str, tuple[int, ...]]:
        return {key: self.loaders[key].shape for key in self.keys}

    # ----------------------------------------------------------------- init

    def _init(self) -> None:
        if not self._keys:
            return
        self._init_loaders()
        self._init_timeline()

    def _init_loaders(self) -> None:
        loaders: dict[str, AbstractLoader] = {}
        for key in self._keys:
            loader_cls = str_to_loader[self._profile[key]]
            directory = self._files[key]
            suffix = self._suffix_of.get(key)
            order_provider = getattr(self, "_order_providers", {}).get(key)
            enumerate_by_regex = key in self._order_spec or (
                key in self._key_spec and "name" in self._key_spec[key]
            )
            # The `pcd` field contract travels with every construction path: it is
            # the channel's declared width, not an artefact of how files are named.
            extra: dict = {}
            if self._profile[key] == "pcd" and key in self._fields_of:
                extra["fields"] = self._fields_of[key]
            if order_provider is not None:  # subclass callable: directory -> filenames
                loaders[key] = loader_cls(
                    directory, files=list(order_provider(directory)), **extra
                )
            elif enumerate_by_regex:
                # Declarative enumeration policy (the `order` regex, else the `key`
                # regex): a channel whose names carry a '_' (a Rellis <epoch>_<ms>),
                # which the default frame-file convention reserves for suffixes, still
                # enumerates, and the loader's own name sort is bypassed.
                if self._profile[key] not in {"npys", "img", "bin", "pcd"}:
                    raise ValueError(
                        f"Channel '{key}' declares a filename key/order but its loader "
                        f"'{self._profile[key]}' has no per-frame files -- filename "
                        f"keys/order need a per-frame loader (npys, img, bin, pcd)."
                    )
                loaders[key] = loader_cls(
                    directory, files=self._enumerate(key, directory), **extra
                )
            elif suffix:
                loaders[key] = loader_cls(
                    directory, files=suffixed_frame_files(directory, suffix), **extra
                )
            elif self._array_file_of.get(key) and self._profile[key] == "npy":
                # A colocated stacked array named explicitly (valid_mask.npy in a
                # shared gicp_poses/): load that file, not the directory's glob[0].
                loaders[key] = loader_cls(directory, file=self._array_file_of[key])
            else:
                loaders[key] = loader_cls(directory, **extra)
        self.loaders: dict[str, AbstractLoader] = loaders
        self.timestamps: dict[str, np.ndarray] = self._collect_timestamps()
        self._check_suffix_coverage()
        self.end_of_time: float = get_end_of_time(self.timestamps) + 1.0

    def _enumerate(self, key: str, directory: str) -> list[str]:
        """Ordered filenames for a channel with a declarative enumeration policy:
        the loader-extension files whose stem matches its ``order`` regex (else its
        ``key`` regex), sorted by the numeric value of the regex's first capture
        group (else lexicographically). This is the ``order`` contract -- it lets a
        channel whose names carry a '_' (a Rellis ``<epoch>_<ms>``, which the default
        frame-file convention reserves for suffixes) enumerate anyway, filters out
        strays (a ``timestamps.txt``, a dotfile, a wrong-extension note), and orders
        even non-zero-padded frame indices correctly."""
        import re

        spec = self._order_spec.get(key) or self._key_spec.get(key, {})
        pattern = spec.get("name")
        if pattern is None:
            raise ValueError(
                f"Channel '{key}' needs an 'order' or 'key' regex ('name') to "
                f"enumerate by; got {spec!r}."
            )
        regex = re.compile(pattern)
        exts = {
            "npys": {".npy"},
            "npy": {".npy"},
            "bin": {".bin"},
            "pcd": {".pcd"},
            "img": {".png", ".jpg", ".jpeg", ".bmp"},
        }.get(self._profile[key])

        def matched(p: Path) -> bool:
            if not p.is_file() or p.name == "timestamps.txt" or p.name.startswith("."):
                return False
            if exts is not None and p.suffix.lower() not in exts:
                return False
            return regex.search(p.stem) is not None

        def order_key(name: str) -> tuple[int, str]:
            match = regex.search(Path(name).stem)
            first = match.groups()[0] if (match and match.groups()) else None
            return (int(first) if (first and first.isdigit()) else 0, name)

        names = sorted(
            (p.name for p in Path(directory).iterdir() if matched(p)), key=order_key
        )
        if not names:
            raise FileNotFoundError(
                f"Channel '{key}': no files in '{directory}' match the enumeration "
                f"regex {pattern!r}."
            )
        return names

    def _as_key_array(self, key: str, values) -> np.ndarray:
        """Validate + normalize a channel's key array: 1-D float, one value per
        frame, non-decreasing -- the timeline and ``synchronize()`` need each
        channel's keys in ascending order."""
        arr = np.atleast_1d(np.asarray(values, dtype=float)).ravel()
        n = len(self.loaders[key])
        if len(arr) != n:
            raise ValueError(
                f"Channel '{key}': its key provider returned {len(arr)} value(s) for "
                f"{n} frame(s)."
            )
        if arr.size > 1 and np.any(np.diff(arr) < 0):
            raise ValueError(
                f"Channel '{key}': keys are not non-decreasing. The timeline and "
                f"synchronize() need each channel's keys ascending -- check the "
                f"key/order regex captures the frame-ordering field."
            )
        return arr

    def _check_suffix_coverage(self) -> None:
        """A suffixed sub-channel borrows the base channel's clock (shared
        directory), so the timeline gives it one slot per base frame. If its
        ``*_<suffix>.npy`` files don't cover every base frame, ``_load`` would
        index past the loader -- fail here, at construction, with a clear message
        instead of a cryptic ``IndexError`` later."""
        for key in self._keys:
            suffix = self._suffix_of.get(key)
            if suffix is None:
                continue
            n_files = len(self.loaders[key])
            n_clock = len(self.timestamps[key])
            if n_files != n_clock:
                shared = Path(self._files[key]).name
                raise ValueError(
                    f"Suffixed sub-channel '{key}' has {n_files} '*_{suffix}.npy' "
                    f"file(s) in '{shared}/' but shares that channel's clock of "
                    f"{n_clock} frame(s): a suffixed variant must cover every base "
                    f"frame. Check for missing or extra '_{suffix}.npy' files."
                )
        # A colocated `array_file` sub-channel borrows its shared directory's
        # clock the same way a suffix channel does, but its stacked array's row
        # count is never validated -- a short array would only surface as a
        # cryptic IndexError deep in _load. Fail here at construction instead.
        for key in self._keys:
            array_file = self._array_file_of.get(key)
            if array_file is None:
                continue
            n_rows = len(self.loaders[key])
            n_clock = len(self.timestamps[key])
            if n_rows != n_clock:
                raise ValueError(
                    f"Colocated array_file sub-channel '{key}' has {n_rows} row(s) "
                    f"in '{array_file}' but shares a clock of {n_clock} frame(s): a "
                    f"colocated array must cover every frame."
                )

    def _collect_timestamps(self) -> dict[str, np.ndarray]:
        """Timestamps per loaded key: its own clock (a ``_key_providers`` callable,
        a declarative ``key`` spec, or a ``timestamps.txt``), else the clock of the
        channel named by its ``timestamps_from`` -- resolved through the *same*
        precedence, so borrowing works whatever the source's clock origin and
        regardless of the order channels are processed in."""
        timestamps: dict[str, np.ndarray] = {}
        fallback: list[str] = []

        def own_clock(key: str) -> np.ndarray | None:
            """A channel's own clock (provider > key spec > timestamps.txt), memoized
            in ``timestamps``; ``None`` if it has none of the three."""
            if key in timestamps:
                return timestamps[key]
            provider = getattr(self, "_key_providers", {}).get(key)
            if provider is not None:  # subclass callable: filenames -> key array
                timestamps[key] = self._as_key_array(
                    key, provider(getattr(self.loaders[key], "files", None))
                )
                return timestamps[key]
            if key in self._key_spec:  # declarative key, parsed in memory
                timestamps[key] = self._as_key_array(key, self._parse_key(key))
                return timestamps[key]
            ts_path = Path(self._files[key]) / "timestamps.txt"
            if ts_path.exists():
                timestamps[key] = load_timestamps(ts_path)
                return timestamps[key]
            return None

        for key in self._keys:
            if own_clock(key) is not None:
                continue
            if key in self._timestamp_aliases:  # timestamps_from a source channel
                src = self._timestamp_aliases[key]
                src_clock = own_clock(src)
                if src_clock is None:
                    raise ValueError(
                        f"'{key}' shares timestamps with '{src}' (timestamps_from), "
                        f"but '{src}' has no resolvable clock (no key spec, provider, "
                        f"or timestamps.txt)."
                    )
                timestamps[key] = src_clock
            else:
                fallback.append(key)
        if fallback:
            timestamps.update(loads_timestamps(fallback, self._files))
        return timestamps

    def _parse_key(self, key: str) -> np.ndarray:
        r"""A channel's alignment key from its ``key`` spec, computed in memory --
        nothing is written. Two forms:

        - ``{name: '<regex>'}``: parse the key from each filename stem. Capture
          groups become a number: with ``scale: [s0, s1, ...]`` as
          ``sum(int(group_i) * s_i)`` (e.g. ``<sec>_<ms>`` with ``scale [1, 0.001]``),
          else ``float('.'.join(groups))`` (one group = an index; two = ``<int>.<frac>``).
        - ``{file: '<name>'}``: read the keys from a named sidecar in the channel
          directory (one float per line -- a differently-named ``timestamps.txt``).
        """
        from apairo.core.keys import parse_filename_key

        spec = self._key_spec[key]
        directory = Path(self._files[key])
        files = getattr(self.loaders[key], "files", None)
        if "file" not in spec and files is None:
            raise ValueError(
                f"Channel '{key}' declares a filename-parsed key but its loader "
                f"('{self._profile[key]}') is stacked and has no per-frame filenames. "
                f"Filename keys need a per-frame loader (npys/img/bin)."
            )
        return parse_filename_key(
            files or [], spec, directory=directory, label=f"Channel '{key}'"
        )

    def _init_timeline(self) -> None:
        """Build the interleaved timeline as two parallel numpy arrays."""
        from apairo.utils.timestamps import merge_timeline

        self._tl_key_idxs, self._tl_frame_idxs = merge_timeline(
            self.timestamps, self._keys
        )

    # ------------------------------------------------------------ dunder

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

    def _load(self, idx: int) -> Sample:
        if not 0 <= idx < len(self):
            raise IndexError(f"Index {idx} out of range [0, {len(self)})")
        key = self._keys[self._tl_key_idxs[idx]]
        frame = int(self._tl_frame_idxs[idx])
        return Sample(
            data={key: self.loaders[key][frame]},
            timestamp=float(self.timestamps[key][frame]),
        )

    # ------------------------------------------------------ frame provenance

    def _sequence_name(self) -> str | None:
        """This (single) sequence's directory name, or ``None`` if unknown."""
        d = getattr(self, "_sequence_dir", None)
        return d.name if d is not None else None

    def frame_info(self, idx: int) -> FrameRef:
        """Channel + row each interleaved event came from. See
        :meth:`AbstractDataset.frame_info`."""
        if not 0 <= idx < len(self):
            raise IndexError(f"Index {idx} out of range [0, {len(self)})")
        return FrameRef(
            sequence=self._sequence_name(),
            channel=self._keys[self._tl_key_idxs[idx]],
            row=int(self._tl_frame_idxs[idx]),
        )

    @property
    def frame_sequence_ids(self) -> np.ndarray:
        """Sequence id per global event -- the sequence directory name (a single
        async dataset is one sequence). Object array of shape ``(len(self),)``."""
        return np.full(len(self), self._sequence_name(), dtype=object)

    @property
    def frame_channel_ids(self) -> np.ndarray:
        """Channel that produced each global event. Object array of shape
        ``(len(self),)``, vectorized from the merged timeline."""
        return np.asarray(self._keys, dtype=object)[self._tl_key_idxs]

    @property
    def frame_stems(self) -> np.ndarray:
        """Filename stem backing each global event: the per-frame data file's
        stem, or the zero-padded row for stacked (single-file) channels."""
        result = np.empty(len(self), dtype=object)
        for i in range(len(self)):
            key = self._keys[self._tl_key_idxs[i]]
            row = int(self._tl_frame_idxs[i])
            files = getattr(self.loaders[key], "files", None)
            result[i] = Path(files[row]).stem if files else f"{row:06d}"
        return result

frame_sequence_ids property

frame_sequence_ids: ndarray

Sequence id per global event -- the sequence directory name (a single async dataset is one sequence). Object array of shape (len(self),).

frame_channel_ids property

frame_channel_ids: ndarray

Channel that produced each global event. Object array of shape (len(self),), vectorized from the merged timeline.

frame_stems property

frame_stems: ndarray

Filename stem backing each global event: the per-frame data file's stem, or the zero-padded row for stacked (single-file) channels.

init classmethod

init(directory: str | Path, *, raw_keys: list[str] | None = None, overwrite: bool = False, merge: bool = False, declare: str | Path | None = None) -> Path

Scan an async-layout directory and write .apairo/channels.yaml.

All detected subdirectories are registered as raw channels. Loader type is inferred from file extensions:

• .bin → bin
• .pcd → pcd
• .png / .jpg / … → img
• multiple .npy files → npys
• single .npy file → npy

For ambiguous cases (e.g. a single-frame .npy that is actually per-frame), call :func:~apairo.core.config.register_raw_channel afterwards to override the detected loader.

Parameters:

Name	Type	Description	Default
directory	str | Path	Dataset root / sequence directory to initialize.	required
raw_keys	list[str] | None	Subdirectory names to include. None → all detected subdirectories with recognizable file types.	None
overwrite	bool	Discard the existing .apairo and rebuild from scratch. Incompatible with merge.	False
merge	bool	Add newly detected raw channels to an existing .apairo without touching channels already declared (raw or preprocessed). If .apairo does not yet exist, behaves like a normal init. Incompatible with overwrite.	False
declare	str | Path | None	Path to an external declaration file the scan should respect, in addition to the in-tree apairo.yaml (which is always read). The scan only reads declarations -- it writes .apairo/ and nothing else.	None

Returns:

Type	Description
Path	Path of the written channels.yaml.

Raises:

Type	Description
ValueError	If both overwrite and merge are True.
FileExistsError	If .apairo already exists and both overwrite and merge are False.
ValueError	If no new recognizable channels are found.

Source code in apairo/dataset/async_layout/dataset.py

@classmethod
def init(
    cls,
    directory: str | Path,
    *,
    raw_keys: list[str] | None = None,
    overwrite: bool = False,
    merge: bool = False,
    declare: str | Path | None = None,
) -> Path:
    """Scan an async-layout directory and write ``.apairo/channels.yaml``.

    All detected subdirectories are registered as raw channels.  Loader
    type is inferred from file extensions:

    * ``.bin`` → ``bin``
    * ``.pcd`` → ``pcd``
    * ``.png`` / ``.jpg`` / … → ``img``
    * multiple ``.npy`` files → ``npys``
    * single ``.npy`` file → ``npy``

    For ambiguous cases (e.g. a single-frame ``.npy`` that is actually
    per-frame), call :func:`~apairo.core.config.register_raw_channel`
    afterwards to override the detected loader.

    Args:
        directory: Dataset root / sequence directory to initialize.
        raw_keys: Subdirectory names to include.  ``None`` → all detected
            subdirectories with recognizable file types.
        overwrite: Discard the existing ``.apairo`` and rebuild from
            scratch.  Incompatible with ``merge``.
        merge: Add newly detected raw channels to an existing ``.apairo``
            without touching channels already declared (raw or
            preprocessed).  If ``.apairo`` does not yet exist, behaves
            like a normal init.  Incompatible with ``overwrite``.
        declare: Path to an external declaration file the scan should
            respect, in addition to the in-tree ``apairo.yaml`` (which is
            always read).  The scan only *reads* declarations -- it writes
            ``.apairo/`` and nothing else.

    Returns:
        Path of the written ``channels.yaml``.

    Raises:
        ValueError: If both ``overwrite`` and ``merge`` are ``True``.
        FileExistsError: If ``.apairo`` already exists and both
            ``overwrite`` and ``merge`` are ``False``.
        ValueError: If no new recognizable channels are found.
    """
    if overwrite and merge:
        raise ValueError("overwrite and merge are mutually exclusive.")

    directory = Path(directory)

    explained = _declared_key_channels(directory, declare)

    if merge and config_exists(directory):
        existing = read_config(directory).get("channels", {})
        added = 0
        for channel_dir in sorted(directory.iterdir()):
            if not channel_dir.is_dir() or channel_dir.name.startswith("."):
                continue
            if raw_keys is not None and channel_dir.name not in raw_keys:
                continue
            loader = _detect_loader(channel_dir)
            if loader is None:
                continue
            if channel_dir.name not in existing:
                _register_raw_channel(directory, channel_dir.name, loader)
                added += 1
            if channel_dir.name in explained:
                continue
            # A directory's base channel may already be registered while a
            # suffix that only appeared later (e.g. *_intensity.npy) is not
            # -- check independently so re-running merge picks it up.
            for suffix, frag in _suffix_channel_entries(
                channel_dir, loader
            ).items():
                key = f"{channel_dir.name}_{suffix}"
                if key in existing:
                    continue
                _register_raw_channel(
                    directory,
                    key,
                    frag["loader"],
                    directory=frag["directory"],
                    suffix=frag["suffix"],
                )
                added += 1
        if added == 0:
            detail = f" (checked: {raw_keys})" if raw_keys else ""
            raise ValueError(
                f"No new recognizable channels found in '{directory}'{detail}."
            )
        return directory / CONFIG_DIR / CHANNELS_FILE

    if config_exists(directory) and not overwrite:
        raise FileExistsError(
            f".apairo already exists in '{directory}'. "
            f"Pass overwrite=True to reinitialize, or merge=True to add new channels."
        )

    channels: dict = {}
    for channel_dir in sorted(directory.iterdir()):
        if not channel_dir.is_dir() or channel_dir.name.startswith("."):
            continue
        if raw_keys is not None and channel_dir.name not in raw_keys:
            continue
        loader = _detect_loader(channel_dir)
        if loader is None:
            continue
        channels[channel_dir.name] = {
            "kind": "raw",
            "loader": loader,
        }
        if channel_dir.name in explained:
            continue
        for suffix, frag in _suffix_channel_entries(channel_dir, loader).items():
            channels[f"{channel_dir.name}_{suffix}"] = {"kind": "raw", **frag}

    if not channels:
        detail = f" (checked: {raw_keys})" if raw_keys else ""
        raise ValueError(
            f"No recognizable channels found in '{directory}'{detail}. "
            f"Expected subdirectories containing .bin, .pcd, .npy, or image files."
        )

    write_config(directory, {"version": 1, "channels": channels})
    return directory / CONFIG_DIR / CHANNELS_FILE

frame_info

frame_info(idx: int) -> FrameRef

Channel + row each interleaved event came from. See :meth:AbstractDataset.frame_info.

Source code in apairo/dataset/async_layout/dataset.py

def frame_info(self, idx: int) -> FrameRef:
    """Channel + row each interleaved event came from. See
    :meth:`AbstractDataset.frame_info`."""
    if not 0 <= idx < len(self):
        raise IndexError(f"Index {idx} out of range [0, {len(self)})")
    return FrameRef(
        sequence=self._sequence_name(),
        channel=self._keys[self._tl_key_idxs[idx]],
        row=int(self._tl_frame_idxs[idx]),
    )

---

StreamDataset

apairo.dataset.stream.StreamDataset

Bases: AbstractDataset

In-memory asynchronous dataset built from timestamped event streams.

The bridge between live or freshly-decoded data (ROS messages, queue items, arrays in RAM) and the apairo API: give it one (timestamps, items) pair per channel and it behaves exactly like a file-backed asynchronous dataset -- merged timeline, single-event samples, and above all :meth:~apairo.core.abstract_dataset.AbstractDataset.synchronize::

ds = StreamDataset({
    "image": (img_ts, img_msgs),      # any indexable items
    "lidar": (lidar_ts, lidar_msgs),
    "odom":  (odom_ts,  odom_msgs),
})

ds[0]                                  # one event, timestamp-ordered
frames = ds.synchronize(reference=clock, method="previous")

Items are stored as given -- they can be numpy arrays, ROS messages, or any Python objects; apairo never copies or converts them.

Parameters:

Name	Type	Description	Default
streams	dict[str, tuple[ndarray, Sequence]]	{channel: (timestamps, items)}. Timestamps must be ascending 1-D arrays; len(items) must match.	required

Raises:

Type	Description
ValueError	On empty streams, length mismatch, or non-ascending timestamps.

Source code in apairo/dataset/stream.py

class StreamDataset(AbstractDataset):
    """In-memory asynchronous dataset built from timestamped event streams.

    The bridge between live or freshly-decoded data (ROS messages, queue
    items, arrays in RAM) and the apairo API: give it one ``(timestamps,
    items)`` pair per channel and it behaves exactly like a file-backed
    asynchronous dataset -- merged timeline, single-event samples, and above
    all :meth:`~apairo.core.abstract_dataset.AbstractDataset.synchronize`::

        ds = StreamDataset({
            "image": (img_ts, img_msgs),      # any indexable items
            "lidar": (lidar_ts, lidar_msgs),
            "odom":  (odom_ts,  odom_msgs),
        })

        ds[0]                                  # one event, timestamp-ordered
        frames = ds.synchronize(reference=clock, method="previous")

    Items are stored as given -- they can be numpy arrays, ROS messages, or
    any Python objects; apairo never copies or converts them.

    Args:
        streams: ``{channel: (timestamps, items)}``.  Timestamps must be
            ascending 1-D arrays; ``len(items)`` must match.

    Raises:
        ValueError: On empty streams, length mismatch, or non-ascending
            timestamps.
    """

    def __init__(
        self,
        streams: dict[str, tuple[np.ndarray, Sequence]],
    ) -> None:
        if not streams:
            raise ValueError("StreamDataset requires at least one stream.")

        self.loaders = {}
        self.timestamps: dict[str, np.ndarray] = {}
        for key, (ts, items) in streams.items():
            ts = np.asarray(ts, dtype=np.float64)
            if ts.ndim != 1 or len(ts) == 0:
                raise ValueError(
                    f"Stream {key!r}: timestamps must be a non-empty 1-D "
                    f"array, got shape {ts.shape}."
                )
            if len(ts) != len(items):
                raise ValueError(
                    f"Stream {key!r}: {len(ts)} timestamps for {len(items)} items."
                )
            if np.any(np.diff(ts) < 0):
                raise ValueError(f"Stream {key!r}: timestamps must be ascending.")
            self.timestamps[key] = ts
            # Duck-typed: any indexable per-frame container works as a loader.
            self.loaders[key] = items  # type: ignore[assignment]

        self._set_keys(list(streams))

        from apairo.utils.timestamps import merge_timeline

        self._tl_key_idxs, self._tl_frame_idxs = merge_timeline(
            self.timestamps, self._keys
        )

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

    def _load(self, idx: int) -> Sample:
        if not 0 <= idx < len(self):
            raise IndexError(f"Index {idx} out of range [0, {len(self)})")
        key = self._keys[self._tl_key_idxs[idx]]
        frame = int(self._tl_frame_idxs[idx])
        return Sample(
            data={key: self.loaders[key][frame]},
            timestamp=float(self.timestamps[key][frame]),
        )

    @property
    def frame_channel_ids(self) -> np.ndarray:
        """Channel that produced each global event. Object array of shape
        ``(len(self),)``, vectorized from the merged timeline."""
        return np.asarray(self._keys, dtype=object)[self._tl_key_idxs]

    def frame_info(self, idx: int) -> FrameRef:
        """Channel + per-channel row each interleaved event came from -- async, so
        it mirrors AsyncLayoutDataset rather than the synchronous default."""
        if not 0 <= idx < len(self):
            raise IndexError(f"Index {idx} out of range [0, {len(self)})")
        return FrameRef(
            sequence=None,
            channel=self._keys[self._tl_key_idxs[idx]],
            row=int(self._tl_frame_idxs[idx]),
        )

    def __repr__(self) -> str:
        sizes = {k: len(v) for k, v in self.loaders.items()}
        return f"StreamDataset(events={len(self)}, streams={sizes})"

frame_channel_ids property

frame_channel_ids: ndarray

Channel that produced each global event. Object array of shape (len(self),), vectorized from the merged timeline.

frame_info

frame_info(idx: int) -> FrameRef

Channel + per-channel row each interleaved event came from -- async, so it mirrors AsyncLayoutDataset rather than the synchronous default.

Source code in apairo/dataset/stream.py

def frame_info(self, idx: int) -> FrameRef:
    """Channel + per-channel row each interleaved event came from -- async, so
    it mirrors AsyncLayoutDataset rather than the synchronous default."""
    if not 0 <= idx < len(self):
        raise IndexError(f"Index {idx} out of range [0, {len(self)})")
    return FrameRef(
        sequence=None,
        channel=self._keys[self._tl_key_idxs[idx]],
        row=int(self._tl_frame_idxs[idx]),
    )

---

SynchronizedView

apairo.core.synchronized_view.SynchronizedView

Bases: AbstractDataset

A synchronous view over an asynchronous dataset.

Created by :meth:~apairo.core.abstract_dataset.AbstractDataset.synchronize. Index i returns a complete sample built around the i-th tick of the reference clock: every channel contributes either an existing event matched by timestamp, or a value synthesized at the tick by an :class:~apairo.core.interpolator.Interpolator. The matching is a pure index computation (one np.searchsorted per channel at construction time) -- no data is read until access.

Because the view is synchronous (timestamps is None), the full chaining API applies: .filter(), .select(), .cache(), .join(), and map-style PyTorch DataLoader with shuffling.

.. note:: The view reads channel data directly from the parent's loaders -- transforms registered on the parent are not applied (they were written for single-event samples). Register transforms on the view::

    ds_sync = ds.synchronize().transform("velodyne_0", RangeFilter(50))

Parameters:

Name	Type	Description	Default
parent	AbstractDataset	Asynchronous dataset exposing per-channel timestamps and loaders.	required
reference	str | ndarray | None	The clock to resample onto. Three forms: channel name -- that channel's timestamps drive the view; None -- the lowest-frequency channel is used; array of timestamps -- an external clock. Enables fixed-rate resampling (np.arange(t0, t1, 1/hz)) or distance-based resampling (see :func:~apairo.utils.timestamps.clock_from_distance).	None
method	ChannelStrategy | dict[str, ChannelStrategy]	Strategy applied to every channel, or a dict mapping channel names to per-channel strategies (unlisted channels default to "previous"). A strategy is one of: "previous" -- last event with t <= t_ref (zero-order hold, never looks into the future; "latest" is a deprecated alias); "next" -- first event with t >= t_ref; "nearest" -- event closest in time to t_ref, either side (ties favour the earlier event); a callable (channel_ts, ref_ts) -> indices returning, for each reference tick, the event index to use (negative = no match, the frame is dropped); an :class:~apairo.core.interpolator.Interpolator -- the value is synthesized at t_ref from the two bracketing events (continuous signals only: poses, IMU, commands).	'previous'
tolerance	float | None	Maximum |t - t_ref| in seconds. For interpolated channels, both bracketing events must lie within tolerance. Reference ticks where any channel has no match are dropped.	None

Example::

from apairo_transform.interp import Se3Interp

ds = TartanKittiDataset(seq, keys=["velodyne_0", "gicp_poses"])
ds_sync = ds.synchronize(
    reference="velodyne_0",
    method={"gicp_poses": Se3Interp()},   # velodyne_0 -> "previous"
    tolerance=0.05,
)

s = ds_sync[0]
s.data["gicp_poses"]    # pose interpolated at s.timestamp

Source code in apairo/core/synchronized_view.py

class SynchronizedView(AbstractDataset):
    """A synchronous view over an asynchronous dataset.

    Created by :meth:`~apairo.core.abstract_dataset.AbstractDataset.synchronize`.
    Index ``i`` returns a complete sample built around the *i*-th tick of the
    reference clock: every channel contributes either an existing event
    matched by timestamp, or a value synthesized at the tick by an
    :class:`~apairo.core.interpolator.Interpolator`.  The matching is a pure
    index computation (one ``np.searchsorted`` per channel at construction
    time) -- no data is read until access.

    Because the view is synchronous (``timestamps`` is ``None``), the full
    chaining API applies: ``.filter()``, ``.select()``, ``.cache()``,
    ``.join()``, and map-style PyTorch ``DataLoader`` with shuffling.

    .. note::
        The view reads channel data directly from the parent's loaders --
        transforms registered on the *parent* are not applied (they were
        written for single-event samples).  Register transforms on the view::

            ds_sync = ds.synchronize().transform("velodyne_0", RangeFilter(50))

    Args:
        parent: Asynchronous dataset exposing per-channel ``timestamps`` and
            ``loaders``.
        reference: The clock to resample onto.  Three forms:

            * channel name -- that channel's timestamps drive the view;
            * ``None`` -- the lowest-frequency channel is used;
            * **array of timestamps** -- an external clock.  Enables
              fixed-rate resampling (``np.arange(t0, t1, 1/hz)``) or
              distance-based resampling (see
              :func:`~apairo.utils.timestamps.clock_from_distance`).
        method: Strategy applied to every channel, or a dict mapping channel
            names to per-channel strategies (unlisted channels default to
            ``"previous"``).  A strategy is one of:

            * ``"previous"`` -- last event with ``t <= t_ref`` (zero-order
              hold, never looks into the future; ``"latest"`` is a
              deprecated alias);
            * ``"next"`` -- first event with ``t >= t_ref``;
            * ``"nearest"`` -- event closest in time to ``t_ref``, either
              side (ties favour the earlier event);
            * a **callable** ``(channel_ts, ref_ts) -> indices`` returning,
              for each reference tick, the event index to use (negative = no
              match, the frame is dropped);
            * an :class:`~apairo.core.interpolator.Interpolator` -- the value
              is synthesized at ``t_ref`` from the two bracketing events
              (continuous signals only: poses, IMU, commands).
        tolerance: Maximum ``|t - t_ref|`` in seconds.  For interpolated
            channels, *both* bracketing events must lie within tolerance.
            Reference ticks where any channel has no match are dropped.

    Example::

        from apairo_transform.interp import Se3Interp

        ds = TartanKittiDataset(seq, keys=["velodyne_0", "gicp_poses"])
        ds_sync = ds.synchronize(
            reference="velodyne_0",
            method={"gicp_poses": Se3Interp()},   # velodyne_0 -> "previous"
            tolerance=0.05,
        )

        s = ds_sync[0]
        s.data["gicp_poses"]    # pose interpolated at s.timestamp
    """

    timestamps = None  # the view itself is synchronous

    def __init__(
        self,
        parent: AbstractDataset,
        reference: str | np.ndarray | None = None,
        method: ChannelStrategy | dict[str, ChannelStrategy] = "previous",
        tolerance: float | None = None,
    ) -> None:
        parent_ts = getattr(parent, "timestamps", None)
        if not isinstance(parent_ts, dict) or not parent_ts:
            raise ValueError(
                f"synchronize() requires an asynchronous dataset with "
                f"per-channel timestamps; {parent.__class__.__name__} is "
                f"already synchronous."
            )

        keys = list(parent.keys)
        strategies = self._resolve_strategies(method, keys)
        ref_name, ref_ts = self._resolve_clock(reference, parent_ts, keys)

        valid = np.ones(len(ref_ts), dtype=bool)
        index_map: dict[str, np.ndarray] = {}
        channel_ts: dict[str, np.ndarray] = {}

        for key in keys:
            ts = np.asarray(parent_ts[key], dtype=float)
            channel_ts[key] = ts
            idx, ok = self._match(strategies[key], ts, ref_ts, tolerance)
            valid &= ok
            index_map[key] = idx

        keep = np.where(valid)[0]
        self._parent = parent
        self._reference = ref_name
        self._method = method
        self._strategies = strategies
        self._tolerance = tolerance
        self._ref_timestamps = ref_ts[keep]
        self._index_map = {k: v[keep].astype(np.intp) for k, v in index_map.items()}
        self._channel_ts = channel_ts
        self._keys = keys

    # ------------------------------------------------------------- resolution

    @staticmethod
    def _resolve_strategies(method, keys: list[str]) -> dict[str, ChannelStrategy]:
        """Normalize *method* into one validated strategy per channel."""
        if isinstance(method, (set, frozenset)):
            raise TypeError(
                f"method must be a single strategy or a dict {{channel: strategy}}, "
                f"not a set -- did you write {{a, b}} instead of {{a: b}}? "
                f"Got {method!r}."
            )
        if isinstance(method, dict):
            unknown = set(method) - set(keys)
            if unknown:
                raise KeyError(
                    f"method maps unknown channels {sorted(unknown)}; "
                    f"dataset keys are {keys}."
                )
            strategies = {k: method.get(k, "previous") for k in keys}
        else:
            strategies = {k: method for k in keys}

        if any(s == "latest" for s in strategies.values()):
            warnings.warn(
                "method='latest' is deprecated, use 'previous' (same "
                "semantics: last event with t <= t_ref).",
                DeprecationWarning,
                stacklevel=4,  # _resolve_strategies <- __init__ <- synchronize()
            )
            strategies = {
                k: ("previous" if s == "latest" else s) for k, s in strategies.items()
            }

        for key, strat in strategies.items():
            if isinstance(strat, Interpolator) or callable(strat):
                continue
            if strat not in ("previous", "next", "nearest"):
                raise ValueError(
                    f"Strategy for {key!r} must be 'previous', 'next', "
                    f"'nearest', a callable (channel_ts, ref_ts) -> indices, "
                    f"or an Interpolator, got {strat!r}"
                )
        return strategies

    @staticmethod
    def _resolve_clock(reference, parent_ts: dict, keys: list[str]):
        """Resolve *reference* into ``(name_or_None, timestamp_array)``.

        Accepts a channel name, ``None`` (lowest-frequency channel), or an
        explicit array of timestamps — an external clock (e.g. fixed-rate
        ticks or distance-based ticks from odometry).
        """
        if reference is None:
            from apairo.utils.timestamps import get_reference_timestamps

            name = get_reference_timestamps({k: parent_ts[k] for k in keys})
            return name, np.asarray(parent_ts[name], dtype=float)

        if isinstance(reference, str):
            if reference not in keys:
                raise KeyError(
                    f"Reference channel {reference!r} not in dataset keys {keys}."
                )
            return reference, np.asarray(parent_ts[reference], dtype=float)

        ref_ts = np.asarray(reference, dtype=float)
        if ref_ts.ndim != 1 or len(ref_ts) == 0:
            raise ValueError(
                f"An external clock must be a non-empty 1-D array of "
                f"timestamps, got shape {ref_ts.shape}."
            )
        if np.any(np.diff(ref_ts) < 0):
            raise ValueError("External clock timestamps must be ascending.")
        return None, ref_ts

    # --------------------------------------------------------------- matching

    @staticmethod
    def _match(
        strat: ChannelStrategy,
        ts: np.ndarray,
        ref_ts: np.ndarray,
        tolerance: float | None,
    ) -> tuple[np.ndarray, np.ndarray]:
        """Compute event indices and per-tick validity for one channel.

        Returns ``(idx, valid)`` where ``idx`` has shape ``(N,)`` for matching
        strategies or ``(N, 2)`` (bracketing pair) for interpolators.
        """
        right = np.searchsorted(ts, ref_ts, side="right")
        latest = right - 1  # last event with t <= t_ref; -1 when none yet

        if isinstance(strat, Interpolator):
            i0 = np.clip(latest, 0, len(ts) - 1)
            i1 = np.clip(right, 0, len(ts) - 1)
            # exact matches collapse to a single index -- the stored value is
            # returned directly, the interpolator is never called
            i1 = np.where(ts[i0] == ref_ts, i0, i1)
            # bracketed: an event at or before the tick, and one at or after
            valid = (latest >= 0) & (ts[i1] >= ref_ts)
            if tolerance is not None:
                valid &= np.maximum(ref_ts - ts[i0], ts[i1] - ref_ts) <= tolerance
            return np.stack([i0, i1], axis=1), valid

        if callable(strat):
            idx = np.asarray(strat(ts, ref_ts))
            if idx.shape != ref_ts.shape:
                raise ValueError(
                    f"Custom method returned shape {idx.shape}, expected "
                    f"{ref_ts.shape} (one index per reference tick; "
                    f"negative = no match)."
                )
            valid = (idx >= 0) & (idx < len(ts))
        elif strat == "previous":
            idx = latest
            valid = latest >= 0
        elif strat == "next":
            # first event with t >= t_ref; len(ts) when none remains
            idx = np.searchsorted(ts, ref_ts, side="left")
            valid = idx < len(ts)
        else:  # nearest -- either side; ties favour the earlier event
            prev = np.clip(latest, 0, len(ts) - 1)
            nxt = np.clip(right, 0, len(ts) - 1)
            idx = np.where(
                np.abs(ts[prev] - ref_ts) <= np.abs(ts[nxt] - ref_ts),
                prev,
                nxt,
            )
            valid = np.ones(len(ref_ts), dtype=bool)

        idx = np.clip(idx, 0, len(ts) - 1)
        if tolerance is not None:
            valid = valid & (np.abs(ts[idx] - ref_ts) <= tolerance)
        return idx, valid

    # ------------------------------------------------------------- properties

    @property
    def root_dir(self) -> Path | None:
        """Delegates to the parent's ``root_dir`` (or ``None`` if it has
        none), so ``.calibration`` still resolves on a synchronized view --
        without this, every synchronized view read an empty ``Calibration``
        regardless of what its parent exposed."""
        return getattr(self._parent, "root_dir", None)

    @property
    def reference(self) -> str | None:
        """Channel providing the clock, or ``None`` for an external clock."""
        return self._reference

    @property
    def reference_timestamps(self) -> np.ndarray:
        """Timestamp of each frame in the view (reference clock)."""
        return self._ref_timestamps

    @property
    def frame_indices(self) -> dict[str, np.ndarray]:
        """Per-channel event indices backing each frame.

        Shape ``(n,)`` for matched channels (the event used), ``(n, 2)`` for
        interpolated channels (the bracketing pair).
        """
        return self._index_map

    def time_offsets(self, key: str) -> np.ndarray:
        """Signed ``t_event - t_ref`` per frame for *key*, in seconds.

        Interpolated channels return zeros: their values are synthesized at
        the reference instant.
        """
        if isinstance(self._strategies[key], Interpolator):
            return np.zeros(len(self), dtype=float)
        ts = self._channel_ts[key]
        return ts[self._index_map[key]] - self._ref_timestamps

    def frame_info(self, idx: int) -> FrameRef:
        """Provenance of synchronised frame *idx*.

        A synchronised frame is *composite*: each channel is backed by its own
        source event (see :attr:`frame_indices`), so there is no single origin
        channel -- ``channel`` is ``None`` and ``row`` is the **view index**, not
        an on-disk row (unlike a profiled dataset, where ``row`` addresses a file
        on disk). ``sequence`` is filled when the parent belongs to a single
        sequence (see :attr:`frame_sequence_ids`), ``None`` otherwise.

        For real per-frame provenance use :attr:`frame_indices`;
        ``frame_indices[self.reference][idx]`` is the reference-clock event this
        tick was resampled onto.
        """
        return super().frame_info(idx)

    @functools.cached_property
    def frame_sequence_ids(self) -> np.ndarray:
        """Sequence id per synchronised frame, shape ``(n,)``.

        A synchronised frame mixes events from several channels, so it only
        carries a sequence id when the whole parent belongs to one sequence;
        over a multi-sequence parent this raises ``AttributeError``, keeping
        the availability probe of the base class."""
        parent_ids = self._parent.frame_sequence_ids
        if len(parent_ids) and (parent_ids != parent_ids[0]).any():
            raise AttributeError(
                f"{type(self).__name__} exposes frame_sequence_ids only over "
                f"a single-sequence parent; this parent spans "
                f"{len(set(parent_ids))} sequences."
            )
        seq = parent_ids[0] if len(parent_ids) else None
        return np.full(len(self), seq, dtype=object)

    @property
    def frame_channel_ids(self) -> np.ndarray:
        """Unavailable: a synchronised frame composites every channel (see
        :meth:`frame_info`), so there is no single origin channel -- unlike
        :attr:`frame_sequence_ids`, this raises unconditionally, even over a
        single-sequence parent. Use :attr:`frame_indices` for real per-channel
        provenance."""
        raise AttributeError(
            f"{type(self).__name__} exposes no frame_channel_ids: a "
            "synchronised frame composites every channel, so there is no "
            "single origin. Use frame_indices for per-channel provenance."
        )

    # ----------------------------------------------------------------- access

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

    def _load(self, idx: int) -> Sample:
        from apairo.core.sample import Sample

        if not 0 <= idx < len(self):
            raise IndexError(f"Index {idx} out of range [0, {len(self)})")

        t_ref = float(self._ref_timestamps[idx])
        data = {}
        for key in self._keys:
            strat = self._strategies[key]
            if isinstance(strat, Interpolator):
                i0, i1 = (int(i) for i in self._index_map[key][idx])
                v0 = self._parent.loaders[key][i0]
                if i0 == i1:  # exact match -- no synthesis needed
                    data[key] = v0
                else:
                    ts = self._channel_ts[key]
                    data[key] = strat(
                        t_ref,
                        float(ts[i0]),
                        v0,
                        float(ts[i1]),
                        self._parent.loaders[key][i1],
                    )
            else:
                data[key] = self._parent.loaders[key][int(self._index_map[key][idx])]
        return Sample(data=data, timestamp=t_ref)

    def __repr__(self) -> str:
        ref = self._reference if self._reference is not None else "<external clock>"
        method: object
        if isinstance(self._method, dict):
            method = "per-channel"
        else:
            m: Any = self._method
            method = getattr(m, "__name__", m)
        return (
            f"SynchronizedView(n={len(self)}, reference={ref!r}, "
            f"method={method!r}, keys={self._keys})"
        )

root_dir property

root_dir: Path | None

Delegates to the parent's root_dir (or None if it has none), so .calibration still resolves on a synchronized view -- without this, every synchronized view read an empty Calibration regardless of what its parent exposed.

reference property

reference: str | None

Channel providing the clock, or None for an external clock.

reference_timestamps property

reference_timestamps: ndarray

Timestamp of each frame in the view (reference clock).

frame_indices property

frame_indices: dict[str, ndarray]

Per-channel event indices backing each frame.

Shape (n,) for matched channels (the event used), (n, 2) for interpolated channels (the bracketing pair).

frame_sequence_ids cached property

frame_sequence_ids: ndarray

Sequence id per synchronised frame, shape (n,).

A synchronised frame mixes events from several channels, so it only carries a sequence id when the whole parent belongs to one sequence; over a multi-sequence parent this raises AttributeError, keeping the availability probe of the base class.

frame_channel_ids property

frame_channel_ids: ndarray

Unavailable: a synchronised frame composites every channel (see :meth:frame_info), so there is no single origin channel -- unlike :attr:frame_sequence_ids, this raises unconditionally, even over a single-sequence parent. Use :attr:frame_indices for real per-channel provenance.

time_offsets

time_offsets(key: str) -> np.ndarray

Signed t_event - t_ref per frame for key, in seconds.

Interpolated channels return zeros: their values are synthesized at the reference instant.

Source code in apairo/core/synchronized_view.py

def time_offsets(self, key: str) -> np.ndarray:
    """Signed ``t_event - t_ref`` per frame for *key*, in seconds.

    Interpolated channels return zeros: their values are synthesized at
    the reference instant.
    """
    if isinstance(self._strategies[key], Interpolator):
        return np.zeros(len(self), dtype=float)
    ts = self._channel_ts[key]
    return ts[self._index_map[key]] - self._ref_timestamps

frame_info

frame_info(idx: int) -> FrameRef

Provenance of synchronised frame idx.

A synchronised frame is composite: each channel is backed by its own source event (see :attr:frame_indices), so there is no single origin channel -- channel is None and row is the view index, not an on-disk row (unlike a profiled dataset, where row addresses a file on disk). sequence is filled when the parent belongs to a single sequence (see :attr:frame_sequence_ids), None otherwise.

For real per-frame provenance use :attr:frame_indices; frame_indices[self.reference][idx] is the reference-clock event this tick was resampled onto.

Source code in apairo/core/synchronized_view.py

def frame_info(self, idx: int) -> FrameRef:
    """Provenance of synchronised frame *idx*.

    A synchronised frame is *composite*: each channel is backed by its own
    source event (see :attr:`frame_indices`), so there is no single origin
    channel -- ``channel`` is ``None`` and ``row`` is the **view index**, not
    an on-disk row (unlike a profiled dataset, where ``row`` addresses a file
    on disk). ``sequence`` is filled when the parent belongs to a single
    sequence (see :attr:`frame_sequence_ids`), ``None`` otherwise.

    For real per-frame provenance use :attr:`frame_indices`;
    ``frame_indices[self.reference][idx]`` is the reference-clock event this
    tick was resampled onto.
    """
    return super().frame_info(idx)

---

Interpolator

apairo.core.interpolator.Interpolator

Bases: ABC

Synthesize a channel value at the reference instant from its two bracketing events.

This is the value-level counterpart of the index-level matching strategies ("previous", "next", "nearest"): instead of picking an existing event, an interpolator builds a new value at t_ref. Use it for continuous signals -- poses, IMU, commands -- never for data that cannot be blended (point clouds, images).

Contract, as orchestrated by :class:~apairo.core.synchronized_view.SynchronizedView:

• a channel whose strategy is an Interpolator receives, for each reference tick t, its two bracketing events (t0, v0) and (t1, v1) with t0 <= t <= t1 and t0 < t1;
• ticks not bracketed by two events (before the first event or after the last) are dropped from the view;
• exact matches (t == t0) bypass the interpolator -- the stored value is returned directly, so implementations never see t0 == t1;
• with tolerance, both neighbours must lie within tolerance of t (max(t - t0, t1 - t) <= tolerance).

Concrete implementations (LinearInterp, Se3Interp, ...) live in apairo_transform.interp.

Example::

class LinearInterp(Interpolator):
    def __call__(self, t, t0, v0, t1, v1):
        a = (t - t0) / (t1 - t0)
        return (1.0 - a) * v0 + a * v1

Source code in apairo/core/interpolator.py

class Interpolator(ABC):
    """Synthesize a channel value at the reference instant from its two
    bracketing events.

    This is the *value-level* counterpart of the index-level matching
    strategies (``"previous"``, ``"next"``, ``"nearest"``): instead of
    picking an existing
    event, an interpolator builds a new value at ``t_ref``.  Use it for
    continuous signals -- poses, IMU, commands -- never for data that cannot
    be blended (point clouds, images).

    Contract, as orchestrated by
    :class:`~apairo.core.synchronized_view.SynchronizedView`:

    * a channel whose strategy is an ``Interpolator`` receives, for each
      reference tick ``t``, its two bracketing events ``(t0, v0)`` and
      ``(t1, v1)`` with ``t0 <= t <= t1`` and ``t0 < t1``;
    * ticks not bracketed by two events (before the first event or after the
      last) are dropped from the view;
    * exact matches (``t == t0``) bypass the interpolator -- the stored
      value is returned directly, so implementations never see ``t0 == t1``;
    * with ``tolerance``, *both* neighbours must lie within tolerance of
      ``t`` (``max(t - t0, t1 - t) <= tolerance``).

    Concrete implementations (``LinearInterp``, ``Se3Interp``, ...) live in
    ``apairo_transform.interp``.

    Example::

        class LinearInterp(Interpolator):
            def __call__(self, t, t0, v0, t1, v1):
                a = (t - t0) / (t1 - t0)
                return (1.0 - a) * v0 + a * v1
    """

    @abstractmethod
    def __call__(self, t: float, t0: float, v0: Any, t1: float, v1: Any) -> Any:
        """Return the channel value at time *t*, ``t0 <= t <= t1``, ``t0 < t1``."""
        ...

__call__ abstractmethod

__call__(t: float, t0: float, v0: Any, t1: float, v1: Any) -> Any

Return the channel value at time t, t0 <= t <= t1, t0 < t1.

Source code in apairo/core/interpolator.py

@abstractmethod
def __call__(self, t: float, t0: float, v0: Any, t1: float, v1: Any) -> Any:
    """Return the channel value at time *t*, ``t0 <= t <= t1``, ``t0 < t1``."""
    ...

---

Dataset composition

ConcatDataset

apairo.dataset.concat.ConcatDataset

Bases: AbstractDataset

Concatenates multiple dataset instances into one.

Takes the intersection of keys across all datasets so every index returns the same set of modalities regardless of which underlying dataset is hit. Indexing is O(log n) via binary search over cumulative lengths.

Parameters:

Name	Type	Description	Default
datasets	list[AbstractDataset]	Non-empty list of dataset instances to concatenate.	required

Example::

sequences = [
    SemanticKittiDataset(f"/data/kitti/seq_{i:02d}", keys=["lidar", "labels"])
    for i in range(11)
]
combined = ConcatDataset(sequences)
sample = combined[0]

Raises:

Type	Description
ValueError	If datasets is empty.

Source code in apairo/dataset/concat.py

class ConcatDataset(AbstractDataset):
    """Concatenates multiple dataset instances into one.

    Takes the intersection of keys across all datasets so every index returns
    the same set of modalities regardless of which underlying dataset is hit.
    Indexing is O(log n) via binary search over cumulative lengths.

    Args:
        datasets: Non-empty list of dataset instances to concatenate.

    Example::

        sequences = [
            SemanticKittiDataset(f"/data/kitti/seq_{i:02d}", keys=["lidar", "labels"])
            for i in range(11)
        ]
        combined = ConcatDataset(sequences)
        sample = combined[0]

    Raises:
        ValueError: If ``datasets`` is empty.
    """

    def __init__(self, datasets: list[AbstractDataset]) -> None:
        if not datasets:
            raise ValueError("datasets must be non-empty")
        self.datasets = datasets
        self._resolve_keys()
        self._lengths = np.array([len(ds) for ds in self.datasets], dtype=np.intp)
        self._cumulative = np.cumsum(self._lengths)

    def _resolve_keys(self) -> None:
        keys = set(self.datasets[0].keys)
        for ds in self.datasets[1:]:
            keys &= set(ds.keys)
        self._keys = sorted(keys)

    @property
    def keys(self) -> list[str]:
        return self._keys

    @keys.setter
    def keys(self, keys) -> None:
        self._set_keys(list(keys))
        self.__dict__.pop("timestamps", None)

    @functools.cached_property
    def timestamps(self) -> dict[str, np.ndarray] | np.ndarray | None:  # type: ignore[override]
        """The concatenated clock. Gated on the structural ``is_synchronous``
        protocol -- every view implements it, whereas the view wrappers do not
        all define a bare ``.timestamps`` attribute. A synchronous child exposes a
        shared per-frame ndarray clock (or ``None``); an asynchronous child a
        per-channel dict."""
        if self.is_synchronous:
            clocks = []
            for ds in self.datasets:
                ts = getattr(ds, "timestamps", None)
                if ts is None:  # a clockless synchronous child -> clockless concat
                    return None
                clocks.append(ts)
            return np.concatenate(clocks)
        result: dict[str, list[np.ndarray]] = {k: [] for k in self._keys}
        for ds in self.datasets:
            ts = getattr(ds, "timestamps", None)
            if ts is None:  # an async child without a clock -> clockless concat
                return None
            for k in self._keys:
                result[k].append(ts[k])
        return {k: np.concatenate(v) for k, v in result.items()}

    @property
    def is_synchronous(self) -> bool:
        # Structural, delegated to the children (mirrors ZipDataset) -- NOT
        # `timestamps is None`, which misreads a clocked synchronous dataset as
        # asynchronous now that synchronous datasets can carry a per-frame clock.
        return all(ds.is_synchronous for ds in self.datasets)

    def _dataset_idx_and_offset(self, idx: int) -> tuple[int, int]:
        if idx < 0 or idx >= self._cumulative[-1]:
            raise IndexError(f"Index {idx} out of range [0, {self._cumulative[-1]})")
        ds_idx = int(np.searchsorted(self._cumulative, idx, side="right"))
        offset = int(self._cumulative[ds_idx - 1]) if ds_idx > 0 else 0
        return ds_idx, offset

    def __len__(self) -> int:
        return int(self._cumulative[-1])

    def frame_info(self, idx: int) -> FrameRef:
        """Provenance of frame *idx*, delegated to the dataset that owns it."""
        ds_idx, offset = self._dataset_idx_and_offset(idx)
        return self.datasets[ds_idx].frame_info(idx - offset)

    @functools.cached_property
    def frame_sequence_ids(self) -> np.ndarray:
        """Sequence id per global frame, concatenated from the children.

        Ids are forwarded verbatim -- two children exposing the same id stay
        indistinguishable. Raises ``AttributeError`` when any child exposes
        none, so the availability probe keeps working."""
        return np.concatenate([ds.frame_sequence_ids for ds in self.datasets])

    @functools.cached_property
    def frame_stems(self) -> np.ndarray:
        """Filename stem per global frame, concatenated from the children."""
        return np.concatenate([ds.frame_stems for ds in self.datasets])

    @functools.cached_property
    def frame_channel_ids(self) -> np.ndarray:
        """Channel that produced each global frame, concatenated from the
        children."""
        return np.concatenate([ds.frame_channel_ids for ds in self.datasets])

    def _load(self, idx: int) -> Sample:
        ds_idx, offset = self._dataset_idx_and_offset(idx)
        sample = self.datasets[ds_idx][idx - offset]
        return Sample(
            data={k: sample.data[k] for k in self._keys if k in sample.data},
            timestamp=sample.timestamp,
        )

timestamps cached property

timestamps: dict[str, ndarray] | ndarray | None

The concatenated clock. Gated on the structural is_synchronous protocol -- every view implements it, whereas the view wrappers do not all define a bare .timestamps attribute. A synchronous child exposes a shared per-frame ndarray clock (or None); an asynchronous child a per-channel dict.

frame_sequence_ids cached property

frame_sequence_ids: ndarray

Sequence id per global frame, concatenated from the children.

Ids are forwarded verbatim -- two children exposing the same id stay indistinguishable. Raises AttributeError when any child exposes none, so the availability probe keeps working.

frame_stems cached property

frame_stems: ndarray

Filename stem per global frame, concatenated from the children.

frame_channel_ids cached property

frame_channel_ids: ndarray

Channel that produced each global frame, concatenated from the children.

frame_info

frame_info(idx: int) -> FrameRef

Provenance of frame idx, delegated to the dataset that owns it.

Source code in apairo/dataset/concat.py

def frame_info(self, idx: int) -> FrameRef:
    """Provenance of frame *idx*, delegated to the dataset that owns it."""
    ds_idx, offset = self._dataset_idx_and_offset(idx)
    return self.datasets[ds_idx].frame_info(idx - offset)

---

split_sequences

apairo.dataset.split_sequences

split_sequences(datasets: list, ratios: tuple[float, float, float] = (0.8, 0.1, 0.1)) -> tuple[list, list, list]

Split datasets into train/val/test at the sequence level.

Splitting at sequence level avoids temporal leakage between splits.

Parameters:

Name	Type	Description	Default
datasets	list	Ordered list (one entry per recording session).	required
ratios	tuple[float, float, float]	(train, val, test) fractions, must sum to 1.0.	(0.8, 0.1, 0.1)

Source code in apairo/dataset/__init__.py

def split_sequences(
    datasets: list,
    ratios: tuple[float, float, float] = (0.8, 0.1, 0.1),
) -> tuple[list, list, list]:
    """Split datasets into train/val/test at the sequence level.

    Splitting at sequence level avoids temporal leakage between splits.

    Args:
        datasets: Ordered list (one entry per recording session).
        ratios: (train, val, test) fractions, must sum to 1.0.
    """
    if abs(sum(ratios) - 1.0) > 1e-6:
        raise ValueError(f"Ratios must sum to 1.0, got {sum(ratios):.4f}")
    n = len(datasets)
    i1 = int(n * ratios[0])
    i2 = i1 + int(n * ratios[1])
    return datasets[:i1], datasets[i1:i2], datasets[i2:]

---

Extensibility

ConfigurableDataset

apairo.core.configurable_dataset.ConfigurableDataset

Mixin for datasets that support preprocessed-channel extensibility via .apairo.

Any dataset class that wants to be extensible at runtime (i.e. allow users to register new preprocessed channels without touching source code) should inherit from this mixin alongside its normal base class.

Concrete subclasses must implement :meth:_bootstrap_config, which describes how to auto-discover the dataset's raw channels when .apairo does not yet exist.

Usage pattern for preprocessing scripts::

MyDataset.register_channel(
    seq_dir, "my_channel", "npys",
    timestamps_from="lidar",
    sources=["lidar"],
)

Usage in dataset __init__::

config = self._load_or_create_config(sequence_dir)

Source code in apairo/core/configurable_dataset.py

class ConfigurableDataset:
    """Mixin for datasets that support preprocessed-channel extensibility via ``.apairo``.

    Any dataset class that wants to be extensible at runtime (i.e. allow users to
    register new preprocessed channels without touching source code) should inherit
    from this mixin alongside its normal base class.

    Concrete subclasses must implement :meth:`_bootstrap_config`, which describes
    how to auto-discover the dataset's raw channels when ``.apairo`` does not yet
    exist.

    Usage pattern for preprocessing scripts::

        MyDataset.register_channel(
            seq_dir, "my_channel", "npys",
            timestamps_from="lidar",
            sources=["lidar"],
        )

    Usage in dataset ``__init__``::

        config = self._load_or_create_config(sequence_dir)
    """

    @classmethod
    def register_channel(
        cls,
        sequence_dir: str | Path,
        key: str,
        loader: str,
        *,
        timestamps_from: str | None = None,
        sources: list[str] | None = None,
        recipe: str | None = None,
    ) -> None:
        """Register a preprocessed channel in ``sequence_dir/.apairo``.

        Args:
            sequence_dir: Dataset sequence directory.
            key: Channel name -- must match its subdirectory name.
            loader: Data format: ``"npy"``, ``"npys"``, ``"bin"``, or ``"img"``.
            timestamps_from: Channel whose timestamps to share when this channel
                has no ``timestamps.txt`` of its own.
            sources: Provenance -- channels this channel was derived from.
            recipe: Content hash of the producing preprocessor's declared config
                (set by ``run_preprocess``; enables ``reuse=True`` skip/regenerate).
        """
        _register_channel(
            sequence_dir,
            key,
            loader,
            timestamps_from=timestamps_from,
            sources=sources,
            recipe=recipe,
        )

    @classmethod
    def remove_channel(
        cls, sequence_dir: str | Path, key: str, *, data: bool = False
    ) -> dict:
        """Remove a channel from ``sequence_dir/.apairo`` (inverse of
        :meth:`register_channel`).

        By default only the declaration is dropped, which is reversible; pass
        ``data=True`` to also delete the channel's directory on disk. Returns the
        removed metadata entry. For an interactive guard on *raw* channels and
        data deletion, prefer the CLI (``apairo channel remove``).

        Args:
            sequence_dir: Dataset sequence directory.
            key: Channel name to remove (its on-disk directory name).
            data: Also delete the channel's directory from disk (destructive).
        """
        return _remove_channel(sequence_dir, key, data=data)

    @abstractmethod
    def _bootstrap_config(self, sequence_dir: Path) -> dict:
        """Return an initial ``.apairo`` config for this dataset.

        Called when no ``.apairo`` exists yet.  Should auto-discover all raw
        channels present in ``sequence_dir`` and return a config dict of the form::

            {
                "version": 1,
                "channels": {
                    "channel_name": {"loader": "npys", "kind": "raw"},
                    ...
                },
            }
        """
        ...

    run_preprocess = _RunPreprocessDescriptor()
    """Run a preprocessor and persist the output channel.

    Can be called on the class or on an existing instance:

    **Class form** -- root_dir required::

        Goose3DDataset.run_preprocess(preprocessor, "/data/GOOSE_3D", split="train")

    **Instance form** -- root_dir inferred from the dataset::

        ds = Goose3DDataset("/data/GOOSE_3D", keys=["lidar"], split="train")
        ds.run_preprocess(preprocessor)

    Extra keyword arguments are forwarded to the dataset constructor (class form)
    or ignored (instance form, since the instance is already configured).
    ``overwrite=True`` recomputes even if the output already exists.

    To preview a :class:`~apairo.core.preprocessor.FramePreprocessor` without
    writing anything, run the same instance lazily first:
    ``ds.transform(preprocessor)``.
    """

    @classmethod
    def describe(cls, sequence_dir: str | Path) -> dict:
        """Describe what is available in a sequence directory.

        Reads ``.apairo`` (creating it if absent) and cross-references it with
        the class's :attr:`available_keys` to produce a three-way breakdown:

        - **raw / present** -- raw channels on disk and registered
        - **raw / missing** -- raw channels known from the profile but not on disk
        - **preprocess** -- channels produced by a preprocessing pipeline

        Returns the breakdown as a dict and prints a human-readable summary.

        Example::

            MyDataset.describe("/data/my_dataset/sequence_01")
        """
        sequence_dir = Path(sequence_dir)
        config = (
            read_config(sequence_dir)
            if config_exists(sequence_dir)
            else cls(sequence_dir)._load_or_create_config(sequence_dir)  # type: ignore[call-arg]
        )
        channels = config.get("channels", {})

        raw_present = sorted(
            k for k, v in channels.items() if v.get("kind", "raw") == "raw"
        )
        preprocess = {
            k: v for k, v in channels.items() if v.get("kind") == "preprocess"
        }
        raw_missing = sorted(
            k for k in getattr(cls, "available_keys", frozenset()) if k not in channels
        )

        # --- pretty print ---
        print(f"\n{cls.__name__} -- {sequence_dir.name}")
        print("─" * 50)

        print("Raw channels")
        if raw_present:
            print("  present  :", ", ".join(raw_present))
        if raw_missing:
            print("  missing  :", ", ".join(raw_missing))
        if not raw_present and not raw_missing:
            print("  (none)")

        print("Preprocessed channels")
        if preprocess:
            for key, meta in sorted(preprocess.items()):
                ts_info = (
                    f"<- timestamps from {meta['timestamps_from']}"
                    if "timestamps_from" in meta
                    else "<- own timestamps"
                )
                src_info = (
                    f"  sources: {meta['sources']}" if meta.get("sources") else ""
                )
                print(f"  {key:<20} {meta['loader']:<6} {ts_info}{src_info}")
        else:
            print("  (none)")
        print()

        return {
            "raw": {"present": raw_present, "missing": raw_missing},
            "preprocess": preprocess,
        }

    @classmethod
    def register_raw_channel(
        cls,
        sequence_dir: str | Path,
        key: str,
        loader: str,
        *,
        alias: str | None = None,
        directory: str | None = None,
        suffix: str | None = None,
    ) -> None:
        """Declare a raw channel in ``sequence_dir/.apairo``.

        Use this to manually add or override a raw channel declaration, for
        example after :meth:`init` detected the wrong loader type.

        Args:
            sequence_dir: Dataset sequence directory.
            key: Channel name -- must match its subdirectory name.
            loader: Data format: ``"npy"``, ``"npys"``, ``"bin"``, ``"img"``, or
                ``"zarr"``.
            alias: Public name to expose the channel under at load time (the
                directory keeps its real name). See
                :func:`~apairo.core.config.set_alias`.
            directory: On-disk subdirectory this channel's files live in, when
                different from *key* (a suffixed sub-channel sharing another
                channel's directory). Defaults to *key*.
            suffix: Frame-file suffix to load from *directory* (e.g.
                ``"intensity"`` for ``000000_intensity.npy``). Pairs with
                *directory*.
        """
        _register_raw_channel(
            sequence_dir, key, loader, alias=alias, directory=directory, suffix=suffix
        )

    @classmethod
    def verify(cls, sequence_dir: str | Path) -> bool:
        """Verify that ``.apairo`` is coherent with what is on disk.

        Prints any issues found and returns ``True`` when the config is clean.

        Args:
            sequence_dir: Dataset sequence directory containing ``.apairo``.

        Returns:
            ``True`` if no issues were found, ``False`` otherwise.

        Example::

            ok = MyDataset.verify("/data/my_dataset/seq_01")
        """
        issues = _verify_config(sequence_dir)
        if not issues:
            print(f"OK  {Path(sequence_dir)}/.apairo")
            return True
        print(f"{len(issues)} issue(s) in {Path(sequence_dir)}/.apairo :")
        for issue in issues:
            print(f"  - {issue}")
        return False

    def _load_or_create_config(self, root_dir: Path) -> dict:
        """Read ``.apairo/channels.yaml`` if it exists, otherwise bootstrap it.

        The bootstrapped config is written back so the next load skips
        detection. Loading must never *require* write access (shared datasets
        often sit on read-only mounts): when the sidecar cannot be written, the
        config is kept on the instance (``_config_fallback``) and picked up by
        :class:`~apairo.dataset.async_layout.AsyncLayoutDataset` in place of
        the on-disk file.
        """
        if not config_exists(root_dir):
            config = self._bootstrap_config(root_dir)
            try:
                write_config(root_dir, config)
            except OSError as exc:
                logger.warning(
                    "Cannot write the .apairo sidecar in '%s' (%s); using the "
                    "bootstrapped config in memory for this instance. Run "
                    "`apairo init` on a writable copy to persist it.",
                    root_dir,
                    exc,
                )
                self._config_fallback = config
        else:
            config = read_config(root_dir)
        return config

run_preprocess class-attribute instance-attribute

run_preprocess = _RunPreprocessDescriptor()

Run a preprocessor and persist the output channel.

Can be called on the class or on an existing instance:

Class form -- root_dir required::

Goose3DDataset.run_preprocess(preprocessor, "/data/GOOSE_3D", split="train")

Instance form -- root_dir inferred from the dataset::

ds = Goose3DDataset("/data/GOOSE_3D", keys=["lidar"], split="train")
ds.run_preprocess(preprocessor)

Extra keyword arguments are forwarded to the dataset constructor (class form) or ignored (instance form, since the instance is already configured). overwrite=True recomputes even if the output already exists.

To preview a :class:~apairo.core.preprocessor.FramePreprocessor without writing anything, run the same instance lazily first: ds.transform(preprocessor).

register_channel classmethod

register_channel(sequence_dir: str | Path, key: str, loader: str, *, timestamps_from: str | None = None, sources: list[str] | None = None, recipe: str | None = None) -> None

Register a preprocessed channel in sequence_dir/.apairo.

Parameters:

Name	Type	Description	Default
sequence_dir	str | Path	Dataset sequence directory.	required
key	str	Channel name -- must match its subdirectory name.	required
loader	str	Data format: "npy", "npys", "bin", or "img".	required
timestamps_from	str | None	Channel whose timestamps to share when this channel has no timestamps.txt of its own.	None
sources	list[str] | None	Provenance -- channels this channel was derived from.	None
recipe	str | None	Content hash of the producing preprocessor's declared config (set by run_preprocess; enables reuse=True skip/regenerate).	None

Source code in apairo/core/configurable_dataset.py

@classmethod
def register_channel(
    cls,
    sequence_dir: str | Path,
    key: str,
    loader: str,
    *,
    timestamps_from: str | None = None,
    sources: list[str] | None = None,
    recipe: str | None = None,
) -> None:
    """Register a preprocessed channel in ``sequence_dir/.apairo``.

    Args:
        sequence_dir: Dataset sequence directory.
        key: Channel name -- must match its subdirectory name.
        loader: Data format: ``"npy"``, ``"npys"``, ``"bin"``, or ``"img"``.
        timestamps_from: Channel whose timestamps to share when this channel
            has no ``timestamps.txt`` of its own.
        sources: Provenance -- channels this channel was derived from.
        recipe: Content hash of the producing preprocessor's declared config
            (set by ``run_preprocess``; enables ``reuse=True`` skip/regenerate).
    """
    _register_channel(
        sequence_dir,
        key,
        loader,
        timestamps_from=timestamps_from,
        sources=sources,
        recipe=recipe,
    )

remove_channel classmethod

remove_channel(sequence_dir: str | Path, key: str, *, data: bool = False) -> dict

Remove a channel from sequence_dir/.apairo (inverse of :meth:register_channel).

By default only the declaration is dropped, which is reversible; pass data=True to also delete the channel's directory on disk. Returns the removed metadata entry. For an interactive guard on raw channels and data deletion, prefer the CLI (apairo channel remove).

Parameters:

Name	Type	Description	Default
sequence_dir	str | Path	Dataset sequence directory.	required
key	str	Channel name to remove (its on-disk directory name).	required
data	bool	Also delete the channel's directory from disk (destructive).	False

Source code in apairo/core/configurable_dataset.py

@classmethod
def remove_channel(
    cls, sequence_dir: str | Path, key: str, *, data: bool = False
) -> dict:
    """Remove a channel from ``sequence_dir/.apairo`` (inverse of
    :meth:`register_channel`).

    By default only the declaration is dropped, which is reversible; pass
    ``data=True`` to also delete the channel's directory on disk. Returns the
    removed metadata entry. For an interactive guard on *raw* channels and
    data deletion, prefer the CLI (``apairo channel remove``).

    Args:
        sequence_dir: Dataset sequence directory.
        key: Channel name to remove (its on-disk directory name).
        data: Also delete the channel's directory from disk (destructive).
    """
    return _remove_channel(sequence_dir, key, data=data)

describe classmethod

describe(sequence_dir: str | Path) -> dict

Describe what is available in a sequence directory.

Reads .apairo (creating it if absent) and cross-references it with the class's :attr:available_keys to produce a three-way breakdown:

• raw / present -- raw channels on disk and registered
• raw / missing -- raw channels known from the profile but not on disk
• preprocess -- channels produced by a preprocessing pipeline

Returns the breakdown as a dict and prints a human-readable summary.

Example::

MyDataset.describe("/data/my_dataset/sequence_01")

Source code in apairo/core/configurable_dataset.py

@classmethod
def describe(cls, sequence_dir: str | Path) -> dict:
    """Describe what is available in a sequence directory.

    Reads ``.apairo`` (creating it if absent) and cross-references it with
    the class's :attr:`available_keys` to produce a three-way breakdown:

    - **raw / present** -- raw channels on disk and registered
    - **raw / missing** -- raw channels known from the profile but not on disk
    - **preprocess** -- channels produced by a preprocessing pipeline

    Returns the breakdown as a dict and prints a human-readable summary.

    Example::

        MyDataset.describe("/data/my_dataset/sequence_01")
    """
    sequence_dir = Path(sequence_dir)
    config = (
        read_config(sequence_dir)
        if config_exists(sequence_dir)
        else cls(sequence_dir)._load_or_create_config(sequence_dir)  # type: ignore[call-arg]
    )
    channels = config.get("channels", {})

    raw_present = sorted(
        k for k, v in channels.items() if v.get("kind", "raw") == "raw"
    )
    preprocess = {
        k: v for k, v in channels.items() if v.get("kind") == "preprocess"
    }
    raw_missing = sorted(
        k for k in getattr(cls, "available_keys", frozenset()) if k not in channels
    )

    # --- pretty print ---
    print(f"\n{cls.__name__} -- {sequence_dir.name}")
    print("─" * 50)

    print("Raw channels")
    if raw_present:
        print("  present  :", ", ".join(raw_present))
    if raw_missing:
        print("  missing  :", ", ".join(raw_missing))
    if not raw_present and not raw_missing:
        print("  (none)")

    print("Preprocessed channels")
    if preprocess:
        for key, meta in sorted(preprocess.items()):
            ts_info = (
                f"<- timestamps from {meta['timestamps_from']}"
                if "timestamps_from" in meta
                else "<- own timestamps"
            )
            src_info = (
                f"  sources: {meta['sources']}" if meta.get("sources") else ""
            )
            print(f"  {key:<20} {meta['loader']:<6} {ts_info}{src_info}")
    else:
        print("  (none)")
    print()

    return {
        "raw": {"present": raw_present, "missing": raw_missing},
        "preprocess": preprocess,
    }

register_raw_channel classmethod

register_raw_channel(sequence_dir: str | Path, key: str, loader: str, *, alias: str | None = None, directory: str | None = None, suffix: str | None = None) -> None

Declare a raw channel in sequence_dir/.apairo.

Use this to manually add or override a raw channel declaration, for example after :meth:init detected the wrong loader type.

Parameters:

Name	Type	Description	Default
sequence_dir	str | Path	Dataset sequence directory.	required
key	str	Channel name -- must match its subdirectory name.	required
loader	str	Data format: "npy", "npys", "bin", "img", or "zarr".	required
alias	str | None	Public name to expose the channel under at load time (the directory keeps its real name). See :func:~apairo.core.config.set_alias.	None
directory	str | None	On-disk subdirectory this channel's files live in, when different from key (a suffixed sub-channel sharing another channel's directory). Defaults to key.	None
suffix	str | None	Frame-file suffix to load from directory (e.g. "intensity" for 000000_intensity.npy). Pairs with directory.	None

Source code in apairo/core/configurable_dataset.py

@classmethod
def register_raw_channel(
    cls,
    sequence_dir: str | Path,
    key: str,
    loader: str,
    *,
    alias: str | None = None,
    directory: str | None = None,
    suffix: str | None = None,
) -> None:
    """Declare a raw channel in ``sequence_dir/.apairo``.

    Use this to manually add or override a raw channel declaration, for
    example after :meth:`init` detected the wrong loader type.

    Args:
        sequence_dir: Dataset sequence directory.
        key: Channel name -- must match its subdirectory name.
        loader: Data format: ``"npy"``, ``"npys"``, ``"bin"``, ``"img"``, or
            ``"zarr"``.
        alias: Public name to expose the channel under at load time (the
            directory keeps its real name). See
            :func:`~apairo.core.config.set_alias`.
        directory: On-disk subdirectory this channel's files live in, when
            different from *key* (a suffixed sub-channel sharing another
            channel's directory). Defaults to *key*.
        suffix: Frame-file suffix to load from *directory* (e.g.
            ``"intensity"`` for ``000000_intensity.npy``). Pairs with
            *directory*.
    """
    _register_raw_channel(
        sequence_dir, key, loader, alias=alias, directory=directory, suffix=suffix
    )

verify classmethod

verify(sequence_dir: str | Path) -> bool

Verify that .apairo is coherent with what is on disk.

Prints any issues found and returns True when the config is clean.

Parameters:

Name	Type	Description	Default
sequence_dir	str | Path	Dataset sequence directory containing .apairo.	required

Returns:

Type	Description
bool	True if no issues were found, False otherwise.

Example::

ok = MyDataset.verify("/data/my_dataset/seq_01")

Source code in apairo/core/configurable_dataset.py

@classmethod
def verify(cls, sequence_dir: str | Path) -> bool:
    """Verify that ``.apairo`` is coherent with what is on disk.

    Prints any issues found and returns ``True`` when the config is clean.

    Args:
        sequence_dir: Dataset sequence directory containing ``.apairo``.

    Returns:
        ``True`` if no issues were found, ``False`` otherwise.

    Example::

        ok = MyDataset.verify("/data/my_dataset/seq_01")
    """
    issues = _verify_config(sequence_dir)
    if not issues:
        print(f"OK  {Path(sequence_dir)}/.apairo")
        return True
    print(f"{len(issues)} issue(s) in {Path(sequence_dir)}/.apairo :")
    for issue in issues:
        print(f"  - {issue}")
    return False

---

RootSequenceMixin

Shared single-sequence vs. dataset-root handling for the asynchronous family (flat indexing, per-sequence access, per-sequence synchronize + concat). Reused by RawDataset and TartanKittiDataset.

apairo.core.root_sequence.RootSequenceMixin

Bases: AbstractDataset

Flat-indexed root over several same-typed sequence datasets.

Source code in apairo/core/root_sequence.py

class RootSequenceMixin(_MixinBase):
    """Flat-indexed root over several same-typed sequence datasets."""

    _is_root: bool = False
    # Single-sequence instances expose their directory (subclass contract).
    _sequence_dir: Path
    # Root instances hold one single-sequence instance of the concrete class
    # (built by the _init_root factory) per sequence directory.
    _sequences: list[Any]

    # ------------------------------------------------------------------ build

    def _init_root(
        self,
        root: str | Path,
        seq_dirs: list[Path],
        make_sequence: Callable[[Path], RootSequenceMixin],
        *,
        build_index: bool = True,
    ) -> None:
        """Populate the root from *seq_dirs*, one sub-dataset per directory.

        Args:
            root: The dataset root directory.
            seq_dirs: Sequence directories, in load order.
            make_sequence: Factory building one single-sequence instance of the
                concrete dataset class from a sequence directory.
            build_index: Build the flat index now.  Pass ``False`` for lazy
                datasets whose sequences have no keys loaded yet (the index is
                built later, when ``keys`` is set).
        """
        self._is_root = True
        self._root_dir = Path(root)
        self._sequences = [make_sequence(d) for d in seq_dirs]
        if build_index:
            self._build_flat_index()

    def _build_flat_index(self) -> None:
        lengths = [len(s) for s in self._sequences]
        self._cumulative_lengths = np.array([0, *np.cumsum(lengths)], dtype=np.intp)

    def _locate(self, idx: int) -> tuple[int, int]:
        """Map a global frame index to ``(sequence index, local row)``.

        Root datasets only. Shared by ``_load``, ``frame_info`` and
        ``derived_path`` so the flat-index arithmetic lives in one place.
        """
        if not hasattr(self, "_cumulative_lengths"):
            raise RuntimeError("No keys loaded. Set ds.keys = [...] first.")
        if not 0 <= idx < len(self):
            raise IndexError(f"Index {idx} out of range [0, {len(self)})")
        seq_idx = int(np.searchsorted(self._cumulative_lengths[1:], idx, side="right"))
        return seq_idx, idx - int(self._cumulative_lengths[seq_idx])

    # ------------------------------------------------------------ subclass hooks

    def _single_available(self) -> frozenset:
        """Channels available in a single sequence -- implemented by subclasses."""
        raise NotImplementedError

    def _set_single_keys(self, keys) -> None:
        """Apply *keys* to a single sequence -- implemented by subclasses."""
        raise NotImplementedError

    # ------------------------------------------------------------- public API

    @property
    def root_dir(self) -> Path:
        return self._root_dir if self._is_root else self._sequence_dir

    @property
    def calibration(self) -> Calibration:
        """Static extrinsics from ``.apairo/calibration.yaml`` (e.g. written from
        ``/tf_static``). On a root, sequences' tables are merged -- each sequence
        carries its own, so the calibration follows the data, not the root."""
        if not self._is_root:
            return read_calibration(self._sequence_dir)
        merged = Calibration()
        for seq in self._sequences:
            cal = read_calibration(seq._sequence_dir)
            merged.update(cal)
            merged.cameras.update(cal.cameras)
        return merged

    @property
    def available(self) -> frozenset:
        """Channels available -- intersection across sequences for a root dataset."""
        if not self._is_root:
            return self._single_available()
        if not self._sequences:
            return frozenset()
        common = frozenset(self._sequences[0].available)
        for seq in self._sequences[1:]:
            common &= frozenset(seq.available)
        return common

    @property
    def sequences(self) -> list:
        """Per-sequence datasets (root datasets only)."""
        if not self._is_root:
            raise AttributeError("'sequences' is only available on root datasets.")
        return self._sequences

    @property
    def sequence_ids(self) -> list[str]:
        """Sequence directory names, in load order (root datasets only)."""
        if not self._is_root:
            raise AttributeError("'sequence_ids' is only available on root datasets.")
        return [seq._sequence_dir.name for seq in self._sequences]

    @property
    def _seq_groups(self) -> dict[str, list[int]] | None:
        """Per-sequence global frame indices ``{seq_name: [i, ...]}`` -- what the
        preprocess runner partitions on, so a ``SequencePreprocessor`` runs once
        per sequence (never across a seam) and writes one output per sequence.
        ``None`` for a single-sequence dataset (the runner treats it as one group)
        or before ``keys`` are set."""
        if not getattr(self, "_is_root", False) or not hasattr(
            self, "_cumulative_lengths"
        ):
            return None
        cum = self._cumulative_lengths
        return {
            seq._sequence_dir.name: list(range(int(cum[i]), int(cum[i + 1])))
            for i, seq in enumerate(self._sequences)
        }

    def sequence(self, seq_id: str) -> SequenceView:
        """Return a :class:`~apairo.core.sequence_view.SequenceView` for *seq_id*."""
        if not self._is_root:
            raise AttributeError("'sequence()' is only available on root datasets.")
        from apairo.core.sequence_view import SequenceView

        for seq in self._sequences:
            if seq._sequence_dir.name == seq_id:
                return SequenceView(seq, range(len(seq)), seq_id)
        raise KeyError(f"Sequence '{seq_id}' not found. Available: {self.sequence_ids}")

    def synchronize(self, reference=None, method="previous", tolerance=None):
        """Resample onto a reference clock -- see :meth:`AbstractDataset.synchronize`.

        On a root dataset each sequence is synchronized independently (clocks are
        not comparable across recordings) and the results concatenated, so an
        external clock array is only valid on a single sequence.
        """
        if not self._is_root:
            return super().synchronize(
                reference=reference, method=method, tolerance=tolerance
            )
        if reference is not None and not isinstance(reference, str):
            raise ValueError(
                "An external clock array cannot be applied to a root dataset: each "
                "sequence has its own time base. Synchronize sequences individually "
                "(ds.sequences[i].synchronize(...)) and concat the results."
            )
        from apairo.dataset.concat import ConcatDataset

        return ConcatDataset(
            [
                seq.synchronize(reference=reference, method=method, tolerance=tolerance)
                for seq in self._sequences
            ]
        )

    # ------------------------------------------------------------------ keys

    @property
    def keys(self) -> list[str]:
        if self._is_root:
            return self._sequences[0].keys if self._sequences else []
        return super().keys

    @keys.setter
    def keys(self, keys) -> None:
        if not self._is_root:
            self._set_single_keys(keys)
            return
        if keys == "all":
            keys = sorted(self.available)
        for seq in self._sequences:
            seq.keys = list(keys)
        self._build_flat_index()

    # ------------------------------------------------------------------ dunder

    def __len__(self) -> int:
        if not self._is_root:
            return super().__len__()  # type: ignore[safe-super]  # layout base implements it
        if not hasattr(self, "_cumulative_lengths"):
            raise RuntimeError("No keys loaded. Set ds.keys = [...] first.")
        return int(self._cumulative_lengths[-1])

    def _load(self, idx):
        if isinstance(idx, tuple):
            seq_id, local_idx = idx
            view = self.sequence(seq_id)
            return self._load(view._indices[local_idx])
        if not self._is_root:
            return super()._load(idx)
        seq_idx, local_idx = self._locate(idx)
        return self._sequences[seq_idx]._load(local_idx)

    # ------------------------------------------------------ frame provenance

    def frame_info(self, idx: int) -> FrameRef:
        """Channel + row each event came from, plus the sub-sequence it belongs
        to (root datasets). See :meth:`AbstractDataset.frame_info`."""
        if not self._is_root:
            return super().frame_info(idx)
        seq_idx, local_idx = self._locate(idx)
        return (
            self._sequences[seq_idx]
            .frame_info(local_idx)
            ._replace(sequence=self.sequence_ids[seq_idx])
        )

    @property
    def frame_sequence_ids(self) -> np.ndarray:
        """Sequence id per global frame index (object array). On a root, the
        sub-sequence each frame belongs to; on a single sequence, delegated."""
        if not self._is_root:
            return super().frame_sequence_ids
        result = np.empty(len(self), dtype=object)
        for seq_idx in range(len(self._sequences)):
            a = int(self._cumulative_lengths[seq_idx])
            b = int(self._cumulative_lengths[seq_idx + 1])
            result[a:b] = self.sequence_ids[seq_idx]
        return result

    @property
    def frame_stems(self) -> np.ndarray:
        """Filename stem per global frame index, concatenated over sequences."""
        if not self._is_root:
            return super().frame_stems
        if not self._sequences:
            return np.empty(0, dtype=object)
        return np.concatenate([s.frame_stems for s in self._sequences])

    @property
    def frame_channel_ids(self) -> np.ndarray:
        """Channel that produced each global frame index, concatenated over
        sequences."""
        if not self._is_root:
            return super().frame_channel_ids
        if not self._sequences:
            return np.empty(0, dtype=object)
        return np.concatenate([s.frame_channel_ids for s in self._sequences])

calibration property

calibration: Calibration

Static extrinsics from .apairo/calibration.yaml (e.g. written from /tf_static). On a root, sequences' tables are merged -- each sequence carries its own, so the calibration follows the data, not the root.

available property

available: frozenset

Channels available -- intersection across sequences for a root dataset.

sequences property

sequences: list

Per-sequence datasets (root datasets only).

sequence_ids property

sequence_ids: list[str]

Sequence directory names, in load order (root datasets only).

frame_sequence_ids property

frame_sequence_ids: ndarray

Sequence id per global frame index (object array). On a root, the sub-sequence each frame belongs to; on a single sequence, delegated.

frame_stems property

frame_stems: ndarray

Filename stem per global frame index, concatenated over sequences.

frame_channel_ids property

frame_channel_ids: ndarray

Channel that produced each global frame index, concatenated over sequences.

sequence

sequence(seq_id: str) -> SequenceView

Return a :class:~apairo.core.sequence_view.SequenceView for seq_id.

Source code in apairo/core/root_sequence.py

def sequence(self, seq_id: str) -> SequenceView:
    """Return a :class:`~apairo.core.sequence_view.SequenceView` for *seq_id*."""
    if not self._is_root:
        raise AttributeError("'sequence()' is only available on root datasets.")
    from apairo.core.sequence_view import SequenceView

    for seq in self._sequences:
        if seq._sequence_dir.name == seq_id:
            return SequenceView(seq, range(len(seq)), seq_id)
    raise KeyError(f"Sequence '{seq_id}' not found. Available: {self.sequence_ids}")

synchronize

synchronize(reference=None, method='previous', tolerance=None)

Resample onto a reference clock -- see :meth:AbstractDataset.synchronize.

On a root dataset each sequence is synchronized independently (clocks are not comparable across recordings) and the results concatenated, so an external clock array is only valid on a single sequence.

Source code in apairo/core/root_sequence.py

def synchronize(self, reference=None, method="previous", tolerance=None):
    """Resample onto a reference clock -- see :meth:`AbstractDataset.synchronize`.

    On a root dataset each sequence is synchronized independently (clocks are
    not comparable across recordings) and the results concatenated, so an
    external clock array is only valid on a single sequence.
    """
    if not self._is_root:
        return super().synchronize(
            reference=reference, method=method, tolerance=tolerance
        )
    if reference is not None and not isinstance(reference, str):
        raise ValueError(
            "An external clock array cannot be applied to a root dataset: each "
            "sequence has its own time base. Synchronize sequences individually "
            "(ds.sequences[i].synchronize(...)) and concat the results."
        )
    from apairo.dataset.concat import ConcatDataset

    return ConcatDataset(
        [
            seq.synchronize(reference=reference, method=method, tolerance=tolerance)
            for seq in self._sequences
        ]
    )

frame_info

frame_info(idx: int) -> FrameRef

Channel + row each event came from, plus the sub-sequence it belongs to (root datasets). See :meth:AbstractDataset.frame_info.

Source code in apairo/core/root_sequence.py

def frame_info(self, idx: int) -> FrameRef:
    """Channel + row each event came from, plus the sub-sequence it belongs
    to (root datasets). See :meth:`AbstractDataset.frame_info`."""
    if not self._is_root:
        return super().frame_info(idx)
    seq_idx, local_idx = self._locate(idx)
    return (
        self._sequences[seq_idx]
        .frame_info(local_idx)
        ._replace(sequence=self.sequence_ids[seq_idx])
    )

---

register_channel

apairo.core.config.register_channel

register_channel(root_dir: str | Path, key: str, loader: str, *, timestamps_from: str | None = None, sources: list[str] | None = None, frame: str | None = None, recipe: str | None = None) -> None

Register a preprocessed channel in root_dir/.apairo/channels.yaml.

This is the low-level standalone function. Most users will prefer the classmethod :meth:ConfigurableDataset.register_channel so that the call site names the dataset type explicitly.

Existing channels (raw or preprocessed) are preserved -- only key is updated.

Parameters:

Name	Type	Description	Default
root_dir	str | Path	Dataset root directory.	required
key	str	Channel name -- must match its subdirectory name.	required
loader	str	Data format: "npy", "npys", "bin", or "img".	required
timestamps_from	str | None	Source channel whose timestamps this channel shares (provenance only -- the channel always has its own timestamps.txt).	None
sources	list[str] | None	Provenance -- raw channels this channel was derived from.	None
frame	str | None	Coordinate frame the channel's data is expressed in (descriptive metadata only; apairo does not apply transforms).	None
recipe	str | None	Content hash of the producing preprocessor's declared config, so a later run_preprocess(..., reuse=True) can tell an identical recipe (skip) from a changed one (regenerate). Provenance only.	None

Source code in apairo/core/config.py

def register_channel(
    root_dir: str | Path,
    key: str,
    loader: str,
    *,
    timestamps_from: str | None = None,
    sources: list[str] | None = None,
    frame: str | None = None,
    recipe: str | None = None,
) -> None:
    """Register a preprocessed channel in ``root_dir/.apairo/channels.yaml``.

    This is the low-level standalone function.  Most users will prefer the
    classmethod :meth:`ConfigurableDataset.register_channel` so that the call
    site names the dataset type explicitly.

    Existing channels (raw or preprocessed) are preserved -- only ``key`` is
    updated.

    Args:
        root_dir: Dataset root directory.
        key: Channel name -- must match its subdirectory name.
        loader: Data format: ``"npy"``, ``"npys"``, ``"bin"``, or ``"img"``.
        timestamps_from: Source channel whose timestamps this channel shares
            (provenance only -- the channel always has its own ``timestamps.txt``).
        sources: Provenance -- raw channels this channel was derived from.
        frame: Coordinate frame the channel's data is expressed in (descriptive
            metadata only; apairo does not apply transforms).
        recipe: Content hash of the producing preprocessor's declared config, so
            a later ``run_preprocess(..., reuse=True)`` can tell an identical
            recipe (skip) from a changed one (regenerate). Provenance only.
    """
    root_dir = Path(root_dir)
    # Read existing config to preserve all other channels (raw + preprocessed).
    config: dict = (
        read_config(root_dir)
        if config_exists(root_dir)
        else {"version": 1, "channels": {}}
    )

    entry: dict = {"kind": "preprocess", "loader": loader}
    if timestamps_from is not None:
        entry["timestamps_from"] = timestamps_from
    if sources:
        entry["sources"] = list(sources)
    if frame is not None:
        entry["frame"] = frame
    if recipe is not None:
        entry["recipe"] = recipe

    config["channels"][key] = entry
    write_config(root_dir, config)

---

WRITERS

Format writers used by the preprocessing runner. Keyed by loader name ("npy", "npys", "bin", "zarr", "img").

from apairo import WRITERS

writer = WRITERS["npy"]()
writer.write(my_array, Path("/data/output/000000.npy"))

apairo.writer.WRITERS module-attribute

WRITERS: dict[str, type] = {'npy': NPYWriter, 'npys': NPYWriter, 'bin': BINWriter, 'zarr': ZarrWriter, 'img': TarImageWriter}

---

DERIVED_LOADERS

File-level loaders for derived/preprocessed keys. Keyed by loader name ("npy", "bin", "img"). Each entry is a Callable[[Path], np.ndarray].

from apairo import DERIVED_LOADERS

tensor = DERIVED_LOADERS["npy"](Path("/data/output/000000.npy"))

apairo.loader.DERIVED_LOADERS module-attribute

DERIVED_LOADERS: dict[str, Callable[[Path], ndarray]] = {'npy': lambda path: np.load(path), 'bin': lambda path: np.fromfile(path, dtype=np.float32), 'img': _load_img, 'pcd': lambda path: read_pcd(str(path))}

---

Preprocessing

FramePreprocessor

apairo.core.preprocessor.FramePreprocessor

Bases: Preprocessor

Preprocessor that operates frame-by-frame.

The runner calls the instance once per input frame. Use this for per-scan operations (label inference, feature extraction, …).

Output is stored as one file per frame (000000.npy, 000001.npy, …) when output_loader is "npys" or "bin".

Because a frame preprocessor is just a Sample -> value callable, it can also run lazily -- ds.transform(preprocessor) publishes its result under output_key at access time, nothing is written. Preview a preprocess this way before materializing it with run_preprocess.

Example::

class TravLabel(FramePreprocessor):
    output_key    = "trav_label"
    output_loader = "npys"
    input_keys    = ["velodyne_0"]
    timestamps_from = "velodyne_0"   # no own timestamps.txt

    def __call__(self, sample: Sample) -> np.ndarray:
        pts = sample.data["velodyne_0"]
        return my_model(pts)

Source code in apairo/core/preprocessor.py

class FramePreprocessor(Preprocessor):
    """Preprocessor that operates frame-by-frame.

    The runner calls the instance once per input frame.  Use this for
    per-scan operations (label inference, feature extraction, …).

    Output is stored as one file per frame (``000000.npy``, ``000001.npy``,
    …) when ``output_loader`` is ``"npys"`` or ``"bin"``.

    Because a frame preprocessor is just a ``Sample -> value`` callable, it
    can also run lazily -- ``ds.transform(preprocessor)`` publishes its
    result under ``output_key`` at access time, nothing is written.  Preview
    a preprocess this way before materializing it with ``run_preprocess``.

    Example::

        class TravLabel(FramePreprocessor):
            output_key    = "trav_label"
            output_loader = "npys"
            input_keys    = ["velodyne_0"]
            timestamps_from = "velodyne_0"   # no own timestamps.txt

            def __call__(self, sample: Sample) -> np.ndarray:
                pts = sample.data["velodyne_0"]
                return my_model(pts)
    """

    @abstractmethod
    def __call__(self, sample: Sample) -> Any:
        """Process one frame.

        Args:
            sample: A :class:`~apairo.core.sample.Sample` whose ``data`` dict
                contains at least the keys declared in :attr:`input_keys`.

        Returns:
            A ``numpy.ndarray`` representing the output for this frame.
        """
        ...

__call__ abstractmethod

__call__(sample: Sample) -> Any

Process one frame.

Parameters:

Name	Type	Description	Default
sample	Sample	A :class:~apairo.core.sample.Sample whose data dict contains at least the keys declared in :attr:input_keys.	required

Returns:

Type	Description
Any	A numpy.ndarray representing the output for this frame.

Source code in apairo/core/preprocessor.py

@abstractmethod
def __call__(self, sample: Sample) -> Any:
    """Process one frame.

    Args:
        sample: A :class:`~apairo.core.sample.Sample` whose ``data`` dict
            contains at least the keys declared in :attr:`input_keys`.

    Returns:
        A ``numpy.ndarray`` representing the output for this frame.
    """
    ...

---

SequencePreprocessor

apairo.core.preprocessor.SequencePreprocessor

Bases: Preprocessor

Preprocessor that operates on the full sequence at once.

The runner calls the instance with an iterator over all input frames. Use this for algorithms that need global context (ICP, trajectory smoothing, …). Global context is also why a sequence preprocessor cannot run lazily: it must be materialized via run_preprocess.

Output is stored as a single {output_key}.npy file when output_loader is "npy".

Example::

class GICPPoses(SequencePreprocessor):
    output_key    = "gicp_poses"
    output_loader = "npy"
    input_keys    = ["velodyne_0"]
    sources       = ["velodyne_0"]   # has its own timestamps.txt

    def __call__(self, frames: Iterator[Sample]) -> np.ndarray:
        poses = []
        for sample in frames:
            pts = sample.data["velodyne_0"]
            poses.append(register(pts))
        return np.stack(poses)           # (N, 4, 4)

Source code in apairo/core/preprocessor.py

class SequencePreprocessor(Preprocessor):
    """Preprocessor that operates on the full sequence at once.

    The runner calls the instance with an iterator over all input frames.
    Use this for algorithms that need global context (ICP, trajectory
    smoothing, …).  Global context is also why a sequence preprocessor
    cannot run lazily: it must be materialized via ``run_preprocess``.

    Output is stored as a single ``{output_key}.npy`` file when
    ``output_loader`` is ``"npy"``.

    Example::

        class GICPPoses(SequencePreprocessor):
            output_key    = "gicp_poses"
            output_loader = "npy"
            input_keys    = ["velodyne_0"]
            sources       = ["velodyne_0"]   # has its own timestamps.txt

            def __call__(self, frames: Iterator[Sample]) -> np.ndarray:
                poses = []
                for sample in frames:
                    pts = sample.data["velodyne_0"]
                    poses.append(register(pts))
                return np.stack(poses)           # (N, 4, 4)
    """

    @abstractmethod
    def __call__(self, frames: Iterator[Sample]) -> Any:
        """Process all frames.

        Args:
            frames: Iterator of :class:`~apairo.core.sample.Sample` objects.

        Returns:
            A ``numpy.ndarray`` of shape ``(N, ...)``.
        """
        ...

__call__ abstractmethod

__call__(frames: Iterator[Sample]) -> Any

Process all frames.

Parameters:

Name	Type	Description	Default
frames	Iterator[Sample]	Iterator of :class:~apairo.core.sample.Sample objects.	required

Returns:

Type	Description
Any	A numpy.ndarray of shape (N, ...).

Source code in apairo/core/preprocessor.py

@abstractmethod
def __call__(self, frames: Iterator[Sample]) -> Any:
    """Process all frames.

    Args:
        frames: Iterator of :class:`~apairo.core.sample.Sample` objects.

    Returns:
        A ``numpy.ndarray`` of shape ``(N, ...)``.
    """
    ...

---

Data structures

Sample

apairo.core.sample.Sample dataclass

A single dataset sample -- one timeline event or one complete frame.

timestamp follows the frame's clock, not merely whether the data is synchronous:

• Asynchronous event -- data has one key; timestamp is that event's.
• Synchronous clocked frame (a synchronize() result) -- data has all requested keys; timestamp is the reference-clock tick the frame was resampled onto.
• Synchronous clockless frame (a profiled dataset) -- data has all keys; timestamp is None.

Source code in apairo/core/sample.py

@dataclass
class Sample:
    """A single dataset sample -- one timeline event or one complete frame.

    ``timestamp`` follows the frame's *clock*, not merely whether the data is
    synchronous:

    - Asynchronous event -- ``data`` has one key; ``timestamp`` is that event's.
    - Synchronous *clocked* frame (a ``synchronize()`` result) -- ``data`` has
      all requested keys; ``timestamp`` is the reference-clock tick the frame
      was resampled onto.
    - Synchronous *clockless* frame (a profiled dataset) -- ``data`` has all
      keys; ``timestamp`` is ``None``.
    """

    data: dict[str, Any]
    timestamp: float | None = None

---

FrameRef

Returned by AbstractDataset.frame_info(idx) -- where a global frame index comes from (sequence, channel, row).

apairo.core.abstract_dataset.FrameRef

Bases: NamedTuple

Where a global frame index comes from -- for layout-aware tooling.

Returned by :meth:AbstractDataset.frame_info. Lets a visualizer or splitter map a flat index back to its origin without reaching into private timeline state.

Attributes:

Name	Type	Description
sequence	str | None	Sub-sequence the frame belongs to (None for a single, unnamed sequence).
channel	str | None	Channel that produced the event -- asynchronous datasets interleave channels, so one event is one channel. None for a synchronous frame, which is all channels at the same row.
row	int	Frame index within that channel/sequence.

Source code in apairo/core/abstract_dataset.py

class FrameRef(NamedTuple):
    """Where a global frame index comes from -- for layout-aware tooling.

    Returned by :meth:`AbstractDataset.frame_info`. Lets a visualizer or splitter
    map a flat index back to its origin without reaching into private timeline
    state.

    Attributes:
        sequence: Sub-sequence the frame belongs to (``None`` for a single,
            unnamed sequence).
        channel: Channel that produced the event -- asynchronous datasets
            interleave channels, so one event is one channel. ``None`` for a
            synchronous frame, which is *all* channels at the same row.
        row: Frame index within that channel/sequence.
    """

    sequence: str | None
    channel: str | None
    row: int

---

ModalitySpec

apairo.core.profiled_dataset.ModalitySpec dataclass

Source code in apairo/core/profiled_dataset.py

@dataclass
class ModalitySpec:
    ext: str
    dtype: str | None = None
    reshape: list | None = None
    mask: int | None = None
    cast_dtype: str | None = None
    loader: str | None = None
    subpath: list[str] = field(default_factory=list)
    optional: bool = False
    resolved_dtype: type | None = field(default=None, compare=False, repr=False)

    @classmethod
    def from_dict(cls, key: str, d: dict) -> ModalitySpec:
        ext = d.get("ext", "")
        if ext and not ext.startswith("."):
            ext = f".{ext}"
        # ``cast_dtype``: target NumPy dtype for a final ``.astype()`` after
        # loading (e.g. int32 labels -> int64).
        cast_dtype = d.get("cast_dtype")
        return cls(
            ext=ext,
            dtype=d.get("dtype"),
            reshape=d.get("reshape"),
            mask=d.get("mask"),
            cast_dtype=cast_dtype,
            loader=d.get("loader"),
            subpath=d.get("subpath", []),
            optional=d.get("optional", False),
            resolved_dtype=_NUMPY_DTYPE.get(cast_dtype) if cast_dtype else None,
        )

    @property
    def is_sequence_file(self) -> bool:
        return self.loader in _SEQUENCE_LOADERS

    def effective_subpath(self, key: str) -> list[str]:
        return self.subpath if self.subpath else [key]