ds_msp.io

Camera I/O: Kalibr camchain YAML, MC-Calib, COLMAP, and nerfstudio interop. See Read/write Kalibr YAML for a task recipe.

ds_msp.io

Camera I/O: Kalibr camchain YAML, COLMAP & nerfstudio interop.

colmap_to_model

colmap_to_model(colmap_model: str, params: Sequence[float])

Reconstruct a DS-MSP camera model from a COLMAP (model, params) pair.

Source code in ds_msp/io/colmap.py

def colmap_to_model(colmap_model: str, params: Sequence[float]):
    """Reconstruct a DS-MSP camera model from a COLMAP ``(model, params)`` pair."""
    p = [float(v) for v in params]
    if colmap_model == "OPENCV_FISHEYE":
        fx, fy, cx, cy, k1, k2, k3, k4 = p
        return KannalaBrandtModel(fx, fy, cx, cy, k1, k2, k3, k4)
    if colmap_model == "OPENCV":
        fx, fy, cx, cy, k1, k2, p1, p2 = p
        return RadTanModel(fx, fy, cx, cy, k1, k2, p1, p2, 0.0)
    if colmap_model == "PINHOLE":
        fx, fy, cx, cy = p
        return RadTanModel(fx, fy, cx, cy)
    if colmap_model == "SIMPLE_PINHOLE":
        f, cx, cy = p
        return RadTanModel(f, f, cx, cy)
    raise NotImplementedError(f"Unsupported COLMAP camera model '{colmap_model}'.")

export_colmap

export_colmap(out_dir: str, model, width: int, height: int, poses: ndarray, image_names: Sequence[str], points3d: Optional[ndarray] = None, point_colors: Optional[ndarray] = None, camera_id: int = 1) -> str

Write a COLMAP text sparse model from a DS-MSP calibration + poses + points.

Parameters:

Name	Type	Description	Default
out_dir	str	Directory to write cameras.txt / images.txt / points3D.txt into.	required
model	CameraModel	DS-MSP camera model (KB / RadTan / pinhole; convert others to KB first).	required
width	int	Image resolution.	required
height	int	Image resolution.	required
poses	(N, 4, 4) array	T_cam_world (world-to-camera) per image, matching image_names.	required
image_names	sequence of str	File names, one per pose.	required
points3d	(M, 3) array	Sparse point cloud (world frame). Many Gaussian-Splatting trainers require a sparse point cloud (they do not support random initialization).	None
point_colors	(M, 3) uint8 array	Per-point RGB; defaults to mid-grey.	None
camera_id	int	Single shared camera id (all images share one intrinsic model).	1

Returns:

Name	Type	Description
out_dir	str

Source code in ds_msp/io/colmap.py

def export_colmap(
    out_dir: str,
    model,
    width: int,
    height: int,
    poses: np.ndarray,
    image_names: Sequence[str],
    points3d: Optional[np.ndarray] = None,
    point_colors: Optional[np.ndarray] = None,
    camera_id: int = 1,
) -> str:
    """Write a COLMAP text sparse model from a DS-MSP calibration + poses + points.

    Parameters
    ----------
    out_dir : str
        Directory to write ``cameras.txt`` / ``images.txt`` / ``points3D.txt`` into.
    model : CameraModel
        DS-MSP camera model (KB / RadTan / pinhole; convert others to KB first).
    width, height : int
        Image resolution.
    poses : (N, 4, 4) array
        ``T_cam_world`` (world-to-camera) per image, matching ``image_names``.
    image_names : sequence of str
        File names, one per pose.
    points3d : (M, 3) array, optional
        Sparse point cloud (world frame). Many Gaussian-Splatting trainers require a
        sparse point cloud (they do not support random initialization).
    point_colors : (M, 3) uint8 array, optional
        Per-point RGB; defaults to mid-grey.
    camera_id : int
        Single shared camera id (all images share one intrinsic model).

    Returns
    -------
    out_dir : str
    """
    poses = np.asarray(poses, dtype=np.float64).reshape(-1, 4, 4)
    if len(poses) != len(image_names):
        raise ValueError(f"poses ({len(poses)}) and image_names ({len(image_names)}) differ")

    colmap_model, params = model_to_colmap(model)
    cameras = [ColmapCamera(camera_id, colmap_model, int(width), int(height), params)]

    images = []
    for i, (T, name) in enumerate(zip(poses, image_names), start=1):
        images.append(ColmapImage(
            id=i, qvec=_qvec_from_R(T[:3, :3]), tvec=T[:3, 3].copy(),
            camera_id=camera_id, name=name,
        ))

    points: List[ColmapPoint3D] = []
    if points3d is not None:
        xyz = np.asarray(points3d, dtype=np.float64).reshape(-1, 3)
        if point_colors is None:
            rgb = np.full((len(xyz), 3), 128, dtype=np.uint8)
        else:
            rgb = np.asarray(point_colors).reshape(-1, 3).astype(np.uint8)
        for j, (p, c) in enumerate(zip(xyz, rgb), start=1):
            points.append(ColmapPoint3D(id=j, xyz=p, rgb=c, error=0.0))

    os.makedirs(out_dir, exist_ok=True)
    _write_cameras(os.path.join(out_dir, "cameras.txt"), cameras)
    _write_images(os.path.join(out_dir, "images.txt"), images)
    _write_points3D(os.path.join(out_dir, "points3D.txt"), points)
    return out_dir

export_nerfstudio

export_nerfstudio(path: str, model, width: int, height: int, poses: ndarray, image_names: Sequence[str]) -> str

Write a nerfstudio transforms.json.

Parameters:

Name	Type	Description	Default
path	str	Output .json path.	required
model	CameraModel	DS-MSP camera model (KB / RadTan / pinhole; convert others to KB first).	required
width	int	Image resolution.	required
height	int	Image resolution.	required
poses	(N, 4, 4) array	T_cam_world (world-to-camera, OpenCV) per image.	required
image_names	sequence of str	File names (used as file_path), one per pose.	required

Returns:

Name	Type	Description
path	str

Source code in ds_msp/io/nerfstudio.py

def export_nerfstudio(
    path: str,
    model,
    width: int,
    height: int,
    poses: np.ndarray,
    image_names: Sequence[str],
) -> str:
    """Write a nerfstudio ``transforms.json``.

    Parameters
    ----------
    path : str
        Output ``.json`` path.
    model : CameraModel
        DS-MSP camera model (KB / RadTan / pinhole; convert others to KB first).
    width, height : int
        Image resolution.
    poses : (N, 4, 4) array
        ``T_cam_world`` (world-to-camera, OpenCV) per image.
    image_names : sequence of str
        File names (used as ``file_path``), one per pose.

    Returns
    -------
    path : str
    """
    poses = np.asarray(poses, dtype=np.float64).reshape(-1, 4, 4)
    if len(poses) != len(image_names):
        raise ValueError(f"poses ({len(poses)}) and image_names ({len(image_names)}) differ")

    colmap_model, params = model_to_colmap(model)
    fx, fy, cx, cy = params[:4]
    out: dict = {
        "camera_model": colmap_model,
        "w": int(width), "h": int(height),
        "fl_x": fx, "fl_y": fy, "cx": cx, "cy": cy,
    }
    for key, val in zip(_DISTORTION_KEYS[colmap_model], params[4:]):
        out[key] = val

    out["frames"] = [
        {"file_path": name, "transform_matrix": _c2w_gl_from_Tcw(T).tolist()}
        for T, name in zip(poses, image_names)
    ]

    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
    with open(path, "w") as f:
        json.dump(out, f, indent=2)
    return path

from_kalibr_cam

from_kalibr_cam(block: dict)

Reconstruct a model from a Kalibr camN stanza (dict).

Source code in ds_msp/io/kalibr.py

def from_kalibr_cam(block: dict):
    """Reconstruct a model from a Kalibr ``camN`` stanza (dict)."""
    cm = block["camera_model"]
    dm = block.get("distortion_model", "none")
    intr = [float(v) for v in block["intrinsics"]]
    D = [float(v) for v in block.get("distortion_coeffs", [])]

    if cm == "ds":
        xi, alpha, fx, fy, cx, cy = intr
        return DoubleSphereModel(fx, fy, cx, cy, xi, alpha)
    if cm == "eucm":
        alpha, beta, fx, fy, cx, cy = intr
        return EUCMModel(fx, fy, cx, cy, alpha, beta)
    if cm == "omni":
        xi_mei, fx, fy, cx, cy = intr
        if dm not in ("none", None, ""):
            raise NotImplementedError("omni + distortion is not representable by UCMModel")
        alpha = xi_mei / (1.0 + xi_mei)
        return UCMModel(fx, fy, cx, cy, alpha)
    if cm == "ds_plus":
        alpha, fx, fy, cx, cy = intr
        lambda1, lambda2, tau_x, tau_y = D
        return DSPlusModel(fx, fy, cx, cy, alpha, lambda1, lambda2, tau_x, tau_y)
    if cm == "pinhole":
        fx, fy, cx, cy = intr
        if dm == "equidistant":
            k1, k2, k3, k4 = D
            return KannalaBrandtModel(fx, fy, cx, cy, k1, k2, k3, k4)
        if dm == "radtan":
            k1, k2, p1, p2 = D
            return RadTanModel(fx, fy, cx, cy, k1, k2, p1, p2, 0.0)
        if dm in ("none", None, ""):
            return RadTanModel(fx, fy, cx, cy)  # plain pinhole (zero distortion)
        raise NotImplementedError(f"Unsupported pinhole distortion_model '{dm}'")
    raise NotImplementedError(f"Unsupported camera_model '{cm}'")

load_kalibr

load_kalibr(path: str, cam: str = 'cam0')

Read a model from a Kalibr camchain YAML file.

Returns the model. Also accessible as (model, (width, height)) via :func:load_kalibr_with_resolution.

Source code in ds_msp/io/kalibr.py

def load_kalibr(path: str, cam: str = "cam0"):
    """Read a model from a Kalibr camchain YAML file.

    Returns the model. Also accessible as ``(model, (width, height))`` via
    :func:`load_kalibr_with_resolution`.
    """
    model, _ = load_kalibr_with_resolution(path, cam)
    return model

load_kalibr_extrinsics

load_kalibr_extrinsics(path: str, cam: str = 'cam1') -> np.ndarray

Load the stereo extrinsic T_cn_cnm1 (4x4) for cam from a camchain.

Kalibr stores, per camera, the transform from the previous camera into this one (so cam1's T_cn_cnm1 is T_cam1_cam0). Raises KeyError if the camera has no extrinsic block (e.g. the first camera in the chain).

Source code in ds_msp/io/kalibr.py

def load_kalibr_extrinsics(path: str, cam: str = "cam1") -> np.ndarray:
    """Load the stereo extrinsic ``T_cn_cnm1`` (4x4) for ``cam`` from a camchain.

    Kalibr stores, per camera, the transform from the *previous* camera into this
    one (so ``cam1``'s ``T_cn_cnm1`` is ``T_cam1_cam0``). Raises ``KeyError`` if the
    camera has no extrinsic block (e.g. the first camera in the chain).
    """
    with open(path, "r") as f:
        data = yaml.safe_load(f)
    T = data.get(cam, {}).get("T_cn_cnm1")
    if T is None:
        raise KeyError(f"{cam!r} has no T_cn_cnm1 in {path}")
    return np.asarray(T, dtype=np.float64)

load_kalibr_with_resolution

load_kalibr_with_resolution(path: str, cam: str = 'cam0') -> Tuple[object, Tuple[int, int]]

Read a model and its image resolution from a Kalibr camchain YAML file.

Parameters:

Name	Type	Description	Default
path	str	Path to a Kalibr camchain YAML file.	required
cam	str	Stanza key to read (e.g. "cam0", "cam1"). If cam is not present in the file, falls back to the alphabetically first camN key found.	"cam0"

Returns:

Name	Type	Description
model	CameraModel	The reconstructed DS-MSP camera model (see the module docstring for the camera-model mapping table).
resolution	tuple of int	(width, height) in pixels, read from the stanza's resolution field; (0, 0) if absent.

Source code in ds_msp/io/kalibr.py

def load_kalibr_with_resolution(path: str, cam: str = "cam0") -> Tuple[object, Tuple[int, int]]:
    """Read a model and its image resolution from a Kalibr camchain YAML file.

    Parameters
    ----------
    path : str
        Path to a Kalibr camchain YAML file.
    cam : str, default "cam0"
        Stanza key to read (e.g. ``"cam0"``, ``"cam1"``). If ``cam`` is not present
        in the file, falls back to the alphabetically first ``camN`` key found.

    Returns
    -------
    model : CameraModel
        The reconstructed DS-MSP camera model (see the module docstring for the
        camera-model mapping table).
    resolution : tuple of int
        ``(width, height)`` in pixels, read from the stanza's ``resolution`` field;
        ``(0, 0)`` if absent.
    """
    with open(path, "r") as f:
        data = yaml.safe_load(f)
    if cam not in data:
        cam = sorted(k for k in data if k.startswith("cam"))[0]
    block = data[cam]
    model = from_kalibr_cam(block)
    res = block.get("resolution", [0, 0])
    return model, (int(res[0]), int(res[1]))

model_to_colmap

model_to_colmap(model) -> tuple[str, List[float]]

Map a DS-MSP camera model to a (colmap_model_name, params) pair.

Source code in ds_msp/io/colmap.py

def model_to_colmap(model) -> tuple[str, List[float]]:
    """Map a DS-MSP camera model to a ``(colmap_model_name, params)`` pair."""
    name = model.name
    if name == "kb":
        return "OPENCV_FISHEYE", [model.fx, model.fy, model.cx, model.cy,
                                  model.k1, model.k2, model.k3, model.k4]
    if name == "radtan":
        if abs(model.k3) > 1e-12:
            warnings.warn(
                "COLMAP OPENCV has no k3; dropping k3=%.3g on export." % model.k3
            )
        return "OPENCV", [model.fx, model.fy, model.cx, model.cy,
                          model.k1, model.k2, model.p1, model.p2]
    raise NotImplementedError(
        f"COLMAP has no native model for '{name}'. Convert to Kannala-Brandt first: "
        f"ds_msp.adapt.convert(model, ds_msp.models.kb.KannalaBrandtModel) "
        f"→ exports as OPENCV_FISHEYE."
    )

read_colmap

read_colmap(in_dir: str) -> dict

Read a COLMAP text sparse model.

Returns a dict with keys: cameras (list[ColmapCamera]), model (the DS-MSP model for the first camera), images (list[ColmapImage], sorted by id), poses ((N,4,4) T_cam_world), image_names (list[str]), points3d ((M,3) or None) and point_colors ((M,3) uint8 or None).

Source code in ds_msp/io/colmap.py

def read_colmap(in_dir: str) -> dict:
    """Read a COLMAP text sparse model.

    Returns a dict with keys: ``cameras`` (list[ColmapCamera]), ``model`` (the
    DS-MSP model for the first camera), ``images`` (list[ColmapImage], sorted by id),
    ``poses`` ((N,4,4) ``T_cam_world``), ``image_names`` (list[str]), ``points3d``
    ((M,3) or None) and ``point_colors`` ((M,3) uint8 or None).
    """
    cameras: List[ColmapCamera] = []
    for line in _data_lines(os.path.join(in_dir, "cameras.txt")):
        parts = line.split()
        cameras.append(ColmapCamera(
            id=int(parts[0]), model=parts[1], width=int(parts[2]), height=int(parts[3]),
            params=[float(v) for v in parts[4:]],
        ))

    images: List[ColmapImage] = []
    img_lines = list(_data_lines(os.path.join(in_dir, "images.txt")))
    i = 0
    while i < len(img_lines):
        h = img_lines[i].split()
        qvec = np.array([float(v) for v in h[1:5]])
        tvec = np.array([float(v) for v in h[5:8]])
        img = ColmapImage(id=int(h[0]), qvec=qvec, tvec=tvec,
                          camera_id=int(h[8]), name=h[9])
        # The points2D line is only present when correspondences were written; a
        # pose-only export omits the (otherwise empty) second line entirely.
        if i + 1 < len(img_lines) and not _looks_like_image_header(img_lines[i + 1]):
            i += 1  # skip the observations line
        images.append(img)
        i += 1
    images.sort(key=lambda im: im.id)

    points_path = os.path.join(in_dir, "points3D.txt")
    xyz_list, rgb_list = [], []
    if os.path.exists(points_path):
        for line in _data_lines(points_path):
            parts = line.split()
            xyz_list.append([float(v) for v in parts[1:4]])
            rgb_list.append([int(v) for v in parts[4:7]])
    points3d = np.array(xyz_list) if xyz_list else None
    point_colors = np.array(rgb_list, dtype=np.uint8) if rgb_list else None

    poses = np.stack([im.T_cam_world for im in images]) if images else np.empty((0, 4, 4))
    model = colmap_to_model(cameras[0].model, cameras[0].params) if cameras else None
    return {
        "cameras": cameras, "model": model, "images": images, "poses": poses,
        "image_names": [im.name for im in images],
        "points3d": points3d, "point_colors": point_colors,
    }

read_nerfstudio

read_nerfstudio(path: str) -> dict

Read a nerfstudio transforms.json.

Returns a dict with model (DS-MSP model), width, height, poses ((N,4,4) T_cam_world) and image_names.

Source code in ds_msp/io/nerfstudio.py

def read_nerfstudio(path: str) -> dict:
    """Read a nerfstudio ``transforms.json``.

    Returns a dict with ``model`` (DS-MSP model), ``width``, ``height``,
    ``poses`` ((N,4,4) ``T_cam_world``) and ``image_names``.
    """
    with open(path, "r") as f:
        data = json.load(f)

    colmap_model = data["camera_model"]
    params: List[float] = [data["fl_x"], data["fl_y"], data["cx"], data["cy"]]
    params += [data[k] for k in _DISTORTION_KEYS[colmap_model] if k in data]
    model = colmap_to_model(colmap_model, params)

    frames = data["frames"]
    poses = np.stack([_Tcw_from_c2w_gl(np.array(fr["transform_matrix"])) for fr in frames]) \
        if frames else np.empty((0, 4, 4))
    return {
        "model": model, "width": int(data["w"]), "height": int(data["h"]),
        "poses": poses, "image_names": [fr["file_path"] for fr in frames],
    }

save_kalibr

save_kalibr(model, path: str, width: int, height: int, cam: str = 'cam0') -> None

Write a single-camera Kalibr camchain YAML file.

Source code in ds_msp/io/kalibr.py

def save_kalibr(model, path: str, width: int, height: int, cam: str = "cam0") -> None:
    """Write a single-camera Kalibr camchain YAML file."""
    data = {cam: to_kalibr_cam(model, width, height)}
    with open(path, "w") as f:
        yaml.safe_dump(data, f, default_flow_style=None, sort_keys=False)

to_kalibr_cam

to_kalibr_cam(model, width: int, height: int) -> dict

Serialize a model into a Kalibr camN stanza (dict).

Source code in ds_msp/io/kalibr.py

def to_kalibr_cam(model, width: int, height: int) -> dict:
    """Serialize a model into a Kalibr ``camN`` stanza (dict)."""
    name = model.name
    if name == "ds":
        block = dict(camera_model="ds",
                     intrinsics=[model.xi, model.alpha, model.fx, model.fy, model.cx, model.cy],
                     distortion_model="none", distortion_coeffs=[])
    elif name == "eucm":
        block = dict(camera_model="eucm",
                     intrinsics=[model.alpha, model.beta, model.fx, model.fy, model.cx, model.cy],
                     distortion_model="none", distortion_coeffs=[])
    elif name == "kb":
        block = dict(camera_model="pinhole",
                     intrinsics=[model.fx, model.fy, model.cx, model.cy],
                     distortion_model="equidistant",
                     distortion_coeffs=[model.k1, model.k2, model.k3, model.k4])
    elif name == "radtan":
        if abs(model.k3) > 1e-12:
            warnings.warn("Kalibr radtan has no k3; dropping k3=%.3g on export." % model.k3)
        block = dict(camera_model="pinhole",
                     intrinsics=[model.fx, model.fy, model.cx, model.cy],
                     distortion_model="radtan",
                     distortion_coeffs=[model.k1, model.k2, model.p1, model.p2])
    elif name == "ucm":
        xi_mei = model.alpha / (1.0 - model.alpha)
        block = dict(camera_model="omni",
                     intrinsics=[xi_mei, model.fx, model.fy, model.cx, model.cy],
                     distortion_model="none", distortion_coeffs=[])
    elif name == "dsplus":
        block = dict(camera_model="ds_plus",
                     intrinsics=[model.alpha, model.fx, model.fy, model.cx, model.cy],
                     distortion_model="ds_plus_div_tilt",
                     distortion_coeffs=[model.lambda1, model.lambda2, model.tau_x, model.tau_y])
    else:
        raise ValueError(f"No Kalibr mapping for model '{name}'")
    block["resolution"] = [int(width), int(height)]
    return block