Skip to content

ds_msp.data

The neutral data layer — observation/correspondence containers and the dataset abstraction consumed by every calibration service (ds_msp.calib, ds_msp.rig) and the detection/IO layers. Depends only on ds_msp.core and NumPy.

ds_msp.data

Neutral data layer: observation/correspondence containers + the dataset abstraction.

Depends only on core and NumPy. Consumed by every calibration service (calib, rig) and the IO/detection adapters, so shared record types never force one service to import another.

BoardObs dataclass

One planar board seen by one camera in one frame (cf. BoardObs.cpp).

Source code in ds_msp/data/observations.py 
62
63
64
65
66
67
68
69
70
71
72
73
@dataclass
class BoardObs:
    """One planar board seen by one camera in one frame (cf. ``BoardObs.cpp``)."""

    cam_id: int
    frame_id: int
    board_id: int
    corner_ids: np.ndarray      # (K,) int  — board-local corner ids that were detected
    pts_2d: np.ndarray          # (K, 2)     — detected pixels
    T_c_b: Optional[np.ndarray] = None   # (4,4) board->camera from robust PnP
    valid: bool = True          # False if PnP inliers < 4 (BoardObs.cpp:149)
    image_path: Optional[str] = None     # source image this was detected in (for overlays)

CalibDataset dataclass

An ordered collection of :class:Observation records.

Source code in ds_msp/data/dataset.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@dataclass
class CalibDataset:
    """An ordered collection of :class:`Observation` records."""

    observations: List[Observation]

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

    def __getitem__(self, i: int) -> Observation:
        return self.observations[i]

    def __iter__(self):
        return iter(self.observations)

    def by_camera(self) -> Dict[int, List[Observation]]:
        """Group observations by ``cam_id`` (insertion order preserved per camera)."""
        out: Dict[int, List[Observation]] = {}
        for o in self.observations:
            out.setdefault(o.cam_id, []).append(o)
        return out

    def by_frame(self) -> Dict[int, List[Observation]]:
        """Group observations by ``frame_id``."""
        out: Dict[int, List[Observation]] = {}
        for o in self.observations:
            out.setdefault(o.frame_id, []).append(o)
        return out

    def as_parallel_lists(self) -> Tuple[List[np.ndarray], List[np.ndarray], List[np.ndarray]]:
        """Return ``(X_world_list, keypoints_list, visibility_list)`` — the legacy
        single-camera view that :func:`ds_msp.calib.calibrate` consumes."""
        X = [o.points_3d for o in self.observations]
        kp = [o.pixels for o in self.observations]
        vis = [o.visibility for o in self.observations]
        return X, kp, vis

    @classmethod
    def from_parallel_lists(cls,
                            X_world_list: Sequence[np.ndarray],
                            keypoints_list: Sequence[np.ndarray],
                            visibility_list: Sequence[np.ndarray],
                            *, cam_id: int = 0) -> "CalibDataset":
        """Build a dataset from the legacy parallel lists (one frame per element)."""
        obs = [
            Observation(points_3d=X, pixels=kp, visibility=vis,
                        cam_id=cam_id, frame_id=i)
            for i, (X, kp, vis) in enumerate(zip(X_world_list, keypoints_list, visibility_list))
        ]
        return cls(obs)

as_parallel_lists 

as_parallel_lists() -> Tuple[List[np.ndarray], List[np.ndarray], List[np.ndarray]]

Return (X_world_list, keypoints_list, visibility_list) — the legacy single-camera view that :func:ds_msp.calib.calibrate consumes.

Source code in ds_msp/data/dataset.py 
51
52
53
54
55
56
57
def as_parallel_lists(self) -> Tuple[List[np.ndarray], List[np.ndarray], List[np.ndarray]]:
    """Return ``(X_world_list, keypoints_list, visibility_list)`` — the legacy
    single-camera view that :func:`ds_msp.calib.calibrate` consumes."""
    X = [o.points_3d for o in self.observations]
    kp = [o.pixels for o in self.observations]
    vis = [o.visibility for o in self.observations]
    return X, kp, vis

by_camera 

by_camera() -> Dict[int, List[Observation]]

Group observations by cam_id (insertion order preserved per camera).

Source code in ds_msp/data/dataset.py 
37
38
39
40
41
42
def by_camera(self) -> Dict[int, List[Observation]]:
    """Group observations by ``cam_id`` (insertion order preserved per camera)."""
    out: Dict[int, List[Observation]] = {}
    for o in self.observations:
        out.setdefault(o.cam_id, []).append(o)
    return out

by_frame 

by_frame() -> Dict[int, List[Observation]]

Group observations by frame_id.

Source code in ds_msp/data/dataset.py 
44
45
46
47
48
49
def by_frame(self) -> Dict[int, List[Observation]]:
    """Group observations by ``frame_id``."""
    out: Dict[int, List[Observation]] = {}
    for o in self.observations:
        out.setdefault(o.frame_id, []).append(o)
    return out

from_parallel_lists classmethod 

from_parallel_lists(X_world_list: Sequence[ndarray], keypoints_list: Sequence[ndarray], visibility_list: Sequence[ndarray], *, cam_id: int = 0) -> 'CalibDataset'

Build a dataset from the legacy parallel lists (one frame per element).

Source code in ds_msp/data/dataset.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
@classmethod
def from_parallel_lists(cls,
                        X_world_list: Sequence[np.ndarray],
                        keypoints_list: Sequence[np.ndarray],
                        visibility_list: Sequence[np.ndarray],
                        *, cam_id: int = 0) -> "CalibDataset":
    """Build a dataset from the legacy parallel lists (one frame per element)."""
    obs = [
        Observation(points_3d=X, pixels=kp, visibility=vis,
                    cam_id=cam_id, frame_id=i)
        for i, (X, kp, vis) in enumerate(zip(X_world_list, keypoints_list, visibility_list))
    ]
    return cls(obs)

Object3D dataclass

Several planar boards fused into one rigid 3D point cloud (cf. Object3D.cpp).

Source code in ds_msp/data/observations.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
@dataclass
class Object3D:
    """Several planar boards fused into one rigid 3D point cloud (cf. ``Object3D.cpp``)."""

    object_id: int
    board_ids: List[int]
    ref_board_id: int                                   # min(board_ids) (McCalib.cpp:898)
    T_co_b: Dict[int, np.ndarray]                       # board_id -> (4,4) board->object
    pts_3d: np.ndarray                                  # (P, 3) all corners in object frame
    pts_obj_2_board: np.ndarray                         # (P, 2) [board_id, corner_id]
    pts_board_2_obj: Dict[Tuple[int, int], int]         # (board_id, corner_id) -> row

    def row_of(self, board_id: int, corner_id: int) -> int:
        """Look up the row into :attr:`pts_3d` for a ``(board_id, corner_id)`` pair.

        Parameters
        ----------
        board_id : int
            One of :attr:`board_ids`.
        corner_id : int
            Board-local corner id (as detected by the board's own numbering).

        Returns
        -------
        int
            Row index into :attr:`pts_3d` / :attr:`pts_obj_2_board`.

        Raises
        ------
        KeyError
            If ``(board_id, corner_id)`` was never fused into this object.
        """
        return self.pts_board_2_obj[(int(board_id), int(corner_id))]

row_of 

row_of(board_id: int, corner_id: int) -> int

Look up the row into :attr:pts_3d for a (board_id, corner_id) pair.

Parameters:

Name Type Description Default
board_id int

One of :attr:board_ids.

 required
corner_id int

Board-local corner id (as detected by the board's own numbering).

 required

Returns:

Type Description
int

Row index into :attr:pts_3d / :attr:pts_obj_2_board.

Raises:

Type Description
KeyError

If (board_id, corner_id) was never fused into this object.
Source code in ds_msp/data/observations.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def row_of(self, board_id: int, corner_id: int) -> int:
    """Look up the row into :attr:`pts_3d` for a ``(board_id, corner_id)`` pair.

    Parameters
    ----------
    board_id : int
        One of :attr:`board_ids`.
    corner_id : int
        Board-local corner id (as detected by the board's own numbering).

    Returns
    -------
    int
        Row index into :attr:`pts_3d` / :attr:`pts_obj_2_board`.

    Raises
    ------
    KeyError
        If ``(board_id, corner_id)`` was never fused into this object.
    """
    return self.pts_board_2_obj[(int(board_id), int(corner_id))]

ObjectObs dataclass

One fused object seen by one camera in one frame (cf. Object3DObs.cpp).

Source code in ds_msp/data/observations.py 
111
112
113
114
115
116
117
118
119
120
121
@dataclass
class ObjectObs:
    """One fused object seen by one camera in one frame (cf. ``Object3DObs.cpp``)."""

    cam_id: int
    frame_id: int
    object_id: int
    point_rows: np.ndarray      # (K,) int  — rows into Object3D.pts_3d
    pts_2d: np.ndarray          # (K, 2)
    T_c_o: Optional[np.ndarray] = None   # (4,4) object->camera from robust PnP
    image_path: Optional[str] = None     # source image this was detected in (for overlays)

Observation dataclass

One view's 3D<->2D correspondences for a single camera in a single frame.

The atomic unit both single-camera calibration and multi-camera rig calibration build on. points_3d are object/board-frame points; pixels the detected image points; visibility masks which rows are usable (e.g. decoded + in-bounds).

Source code in ds_msp/data/observations.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
@dataclass
class Observation:
    """One view's 3D<->2D correspondences for a single camera in a single frame.

    The atomic unit both single-camera calibration and multi-camera rig calibration
    build on. ``points_3d`` are object/board-frame points; ``pixels`` the detected image
    points; ``visibility`` masks which rows are usable (e.g. decoded + in-bounds).
    """

    points_3d: np.ndarray        # (N, 3) object/board-frame points
    pixels: np.ndarray           # (N, 2) detected pixels
    visibility: np.ndarray       # (N,) bool
    cam_id: int = 0
    frame_id: int = 0

    def __post_init__(self) -> None:
        self.points_3d = np.asarray(self.points_3d, dtype=np.float64)
        self.pixels = np.asarray(self.pixels, dtype=np.float64)
        if self.visibility is None:
            self.visibility = np.ones(len(self.points_3d), dtype=bool)
        else:
            self.visibility = np.asarray(self.visibility, dtype=bool)
        n = len(self.points_3d)
        if self.points_3d.shape != (n, 3):
            raise ValueError(f"points_3d must be (N,3), got {self.points_3d.shape}")
        if self.pixels.shape != (n, 2):
            raise ValueError(f"pixels must be (N,2), got {self.pixels.shape}")
        if self.visibility.shape != (n,):
            raise ValueError(f"visibility must be (N,), got {self.visibility.shape}")

RigState dataclass

The optimization variable mutated by the staged global BA (rig.bundle).

Source code in ds_msp/data/observations.py 
124
125
126
127
128
129
130
131
132
133
@dataclass
class RigState:
    """The optimization variable mutated by the staged global BA (``rig.bundle``)."""

    cameras: Dict[int, CameraModel]                     # per-camera intrinsics
    T_c_g: Dict[int, np.ndarray]                        # camera-in-group; ref cam = identity
    ref_cam_id: int
    object_poses: Dict[Tuple[int, int], np.ndarray]     # (object_id, frame_id) -> T_g_o
    objects: Dict[int, Object3D]                        # holds T_co_b board poses
    img_size: Dict[int, Tuple[int, int]] = field(default_factory=dict)  # cam_id -> (w, h)