Skip to content

Data Transforms

Data transformations and compatibility layers.

roverd.transforms.ouster

Ouster Lidar transforms.

Warning

This module is not automatically imported; you will need to explicitly import it:

from roverd.transforms import ouster

This is a pure-numpy reimplementation of the subset of the ouster-sdk lidar projection API (SensorInfo, destagger, XYZLut) required by Destagger and PointCloud, since ouster-sdk is an unstable dependency which we would like to avoid requiring.

The projection math follows the "Lidar Range Data to Cartesian Coordinates" algorithm described in the Ouster Software User Guide, and has been validated to reproduce ouster.sdk.client.XYZLut bit-for-bit (up to floating point error) on the sensor metadata this codebase generates.

roverd.transforms.ouster.ConfigCache

Ouster sensor intrinsics cache.

Source code in format/src/roverd/transforms/ouster.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
class ConfigCache:
    """Ouster sensor intrinsics cache."""

    def __init__(self) -> None:
        self._by_path: dict[str, LidarIntrinsics] = {}
        self._by_cfg: dict[str, LidarIntrinsics] = {}

    def __getitem__(self, path: str) -> LidarIntrinsics:
        """Get sensor intrinsics by path.

        - First checks if we have `LidarIntrinsics` loaded already for the
            given path.
        - If not, we then check if we have `LidarIntrinsics` loaded with an
            identical configuration.
        - If neither is found, we read the file and parse a new
            `LidarIntrinsics`.

        Args:
            path: Path to the sensor intrinsics file.

        Returns:
            Parsed lidar intrinsics.
        """
        # We've never seen this path before. Need to inspect it.
        if path not in self._by_path:
            try:
                with open(path) as f:
                    cfg = f.read()
            except FileNotFoundError as e:
                raise ValueError(
                    f"Lidar sensor intrinsics {path} does not exist.") from e

            # Truly new -> load
            if cfg not in self._by_cfg:
                self._by_cfg[cfg] = LidarIntrinsics.from_json(cfg)
            # Make a link
            self._by_path[path] = self._by_cfg[cfg]

        return self._by_path[path]

__getitem__ 

__getitem__(path: str) -> LidarIntrinsics

Get sensor intrinsics by path.

  • First checks if we have LidarIntrinsics loaded already for the given path.
  • If not, we then check if we have LidarIntrinsics loaded with an identical configuration.
  • If neither is found, we read the file and parse a new LidarIntrinsics.

Parameters:

Name Type Description Default
path str

Path to the sensor intrinsics file.

 required

Returns:

Type Description
LidarIntrinsics

Parsed lidar intrinsics.
Source code in format/src/roverd/transforms/ouster.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def __getitem__(self, path: str) -> LidarIntrinsics:
    """Get sensor intrinsics by path.

    - First checks if we have `LidarIntrinsics` loaded already for the
        given path.
    - If not, we then check if we have `LidarIntrinsics` loaded with an
        identical configuration.
    - If neither is found, we read the file and parse a new
        `LidarIntrinsics`.

    Args:
        path: Path to the sensor intrinsics file.

    Returns:
        Parsed lidar intrinsics.
    """
    # We've never seen this path before. Need to inspect it.
    if path not in self._by_path:
        try:
            with open(path) as f:
                cfg = f.read()
        except FileNotFoundError as e:
            raise ValueError(
                f"Lidar sensor intrinsics {path} does not exist.") from e

        # Truly new -> load
        if cfg not in self._by_cfg:
            self._by_cfg[cfg] = LidarIntrinsics.from_json(cfg)
        # Make a link
        self._by_path[path] = self._by_cfg[cfg]

    return self._by_path[path]

roverd.transforms.ouster.Destagger

Bases: Transform[OSDepth, Depth]

Destagger Ouster Lidar depth data.

Parameters:

Name Type Description Default
config ConfigCache | None

if provided, use this configuration cache (to share with other transforms).

 None
Source code in format/src/roverd/transforms/ouster.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
class Destagger(spec.Transform[types.OSDepth, types.Depth]):
    """Destagger Ouster Lidar depth data.

    Args:
        config: if provided, use this configuration cache (to share with other
            transforms).
    """

    def __init__(self, config: ConfigCache | None = None) -> None:
        if config is None:
            config = ConfigCache()
        self._config = config

    def __call__(self, data: types.OSDepth) -> types.Depth:
        """Destagger Ouster Lidar depth data.

        Args:
            data: Ouster Lidar depth data with staggered measurements.

        Returns:
            Depth data with destaggered measurements.
        """
        info = self._config[data.intrinsics]
        shifts = info.pixel_shift_by_row
        width = data.rng.shape[-1]

        # Each beam's column is circularly shifted by its own azimuth
        # firing offset, so that all beams end up aligned to a common
        # azimuth per column: destaggered[..., h, w] = rng[..., h, (w -
        # shifts[h]) % width].
        columns = (np.arange(width)[None, :] - shifts[:, None]) % width
        destaggered = np.take_along_axis(
            data.rng, columns[None, None], axis=-1)

        return types.Depth(rng=destaggered, timestamps=data.timestamps)

__call__ 

__call__(data: OSDepth) -> Depth

Destagger Ouster Lidar depth data.

Parameters:

Name Type Description Default
data OSDepth

Ouster Lidar depth data with staggered measurements.

 required

Returns:

Type Description
Depth

Depth data with destaggered measurements.
Source code in format/src/roverd/transforms/ouster.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
def __call__(self, data: types.OSDepth) -> types.Depth:
    """Destagger Ouster Lidar depth data.

    Args:
        data: Ouster Lidar depth data with staggered measurements.

    Returns:
        Depth data with destaggered measurements.
    """
    info = self._config[data.intrinsics]
    shifts = info.pixel_shift_by_row
    width = data.rng.shape[-1]

    # Each beam's column is circularly shifted by its own azimuth
    # firing offset, so that all beams end up aligned to a common
    # azimuth per column: destaggered[..., h, w] = rng[..., h, (w -
    # shifts[h]) % width].
    columns = (np.arange(width)[None, :] - shifts[:, None]) % width
    destaggered = np.take_along_axis(
        data.rng, columns[None, None], axis=-1)

    return types.Depth(rng=destaggered, timestamps=data.timestamps)

roverd.transforms.ouster.LidarIntrinsics dataclass

Parsed Ouster lidar sensor metadata (lidar.json).

Only the fields required for destaggering and XYZ projection are kept; see the Ouster sensor metadata reference for the full schema.

Attributes:

Name Type Description
pixels_per_column int

number of beams.
columns_per_frame int

number of measurements ("ticks") per rotation.
pixel_shift_by_row Int64[ndarray, beam]

per-beam staggering offset, in columns.
beam_altitude_angles Float64[ndarray, beam]

per-beam altitude angle, in degrees.
beam_azimuth_angles Float64[ndarray, beam]

per-beam azimuth firing offset, in degrees.
beam_to_lidar_transform Float64[ndarray, '4 4']

4x4 transform from the per-beam origin to the lidar frame (at zero encoder angle).
lidar_to_sensor_transform Float64[ndarray, '4 4']

4x4 transform from the lidar frame to the sensor frame.
Source code in format/src/roverd/transforms/ouster.py 
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@dataclass
class LidarIntrinsics:
    """Parsed Ouster lidar sensor metadata (`lidar.json`).

    Only the fields required for destaggering and XYZ projection are kept;
    see the [Ouster sensor metadata reference](
    https://static.ouster.dev/sdk-docs/reference/sensor-data.html#sensor-metadata)
    for the full schema.

    Attributes:
        pixels_per_column: number of beams.
        columns_per_frame: number of measurements ("ticks") per rotation.
        pixel_shift_by_row: per-beam staggering offset, in columns.
        beam_altitude_angles: per-beam altitude angle, in degrees.
        beam_azimuth_angles: per-beam azimuth firing offset, in degrees.
        beam_to_lidar_transform: `4x4` transform from the per-beam origin to
            the lidar frame (at zero encoder angle).
        lidar_to_sensor_transform: `4x4` transform from the lidar frame to
            the sensor frame.
    """

    pixels_per_column: int
    columns_per_frame: int
    pixel_shift_by_row: Int64[np.ndarray, "beam"]
    beam_altitude_angles: Float64[np.ndarray, "beam"]
    beam_azimuth_angles: Float64[np.ndarray, "beam"]
    beam_to_lidar_transform: Float64[np.ndarray, "4 4"]
    lidar_to_sensor_transform: Float64[np.ndarray, "4 4"]

    @classmethod
    def from_json(cls, cfg: str) -> "LidarIntrinsics":
        """Parse lidar intrinsics from a `lidar.json` metadata string.

        Args:
            cfg: `lidar.json` file contents, as written by
                `ouster.sdk.client.SensorInfo.to_json_string`.

        Returns:
            Parsed lidar intrinsics.
        """
        meta = json.loads(cfg)
        fmt = meta["lidar_data_format"]
        beam = meta["beam_intrinsics"]
        lidar = meta["lidar_intrinsics"]

        return cls(
            pixels_per_column=fmt["pixels_per_column"],
            columns_per_frame=fmt["columns_per_frame"],
            pixel_shift_by_row=np.array(
                fmt["pixel_shift_by_row"], dtype=np.int64),
            beam_altitude_angles=np.array(
                beam["beam_altitude_angles"], dtype=np.float64),
            beam_azimuth_angles=np.array(
                beam["beam_azimuth_angles"], dtype=np.float64),
            beam_to_lidar_transform=np.array(
                beam["beam_to_lidar_transform"], dtype=np.float64
            ).reshape(4, 4),
            lidar_to_sensor_transform=np.array(
                lidar["lidar_to_sensor_transform"], dtype=np.float64
            ).reshape(4, 4))

from_json classmethod 

from_json(cfg: str) -> LidarIntrinsics

Parse lidar intrinsics from a lidar.json metadata string.

Parameters:

Name Type Description Default
cfg str

lidar.json file contents, as written by ouster.sdk.client.SensorInfo.to_json_string.

 required

Returns:

Type Description
LidarIntrinsics

Parsed lidar intrinsics.
Source code in format/src/roverd/transforms/ouster.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@classmethod
def from_json(cls, cfg: str) -> "LidarIntrinsics":
    """Parse lidar intrinsics from a `lidar.json` metadata string.

    Args:
        cfg: `lidar.json` file contents, as written by
            `ouster.sdk.client.SensorInfo.to_json_string`.

    Returns:
        Parsed lidar intrinsics.
    """
    meta = json.loads(cfg)
    fmt = meta["lidar_data_format"]
    beam = meta["beam_intrinsics"]
    lidar = meta["lidar_intrinsics"]

    return cls(
        pixels_per_column=fmt["pixels_per_column"],
        columns_per_frame=fmt["columns_per_frame"],
        pixel_shift_by_row=np.array(
            fmt["pixel_shift_by_row"], dtype=np.int64),
        beam_altitude_angles=np.array(
            beam["beam_altitude_angles"], dtype=np.float64),
        beam_azimuth_angles=np.array(
            beam["beam_azimuth_angles"], dtype=np.float64),
        beam_to_lidar_transform=np.array(
            beam["beam_to_lidar_transform"], dtype=np.float64
        ).reshape(4, 4),
        lidar_to_sensor_transform=np.array(
            lidar["lidar_to_sensor_transform"], dtype=np.float64
        ).reshape(4, 4))

roverd.transforms.ouster.PointCloud

Bases: Transform[OSDepth, PointCloud]

Calculate point cloud from Ouster Lidar depth data.

Parameters:

Name Type Description Default
min_range float | None

minimum range to include in the point cloud; if None, all points are included.

 None
Source code in format/src/roverd/transforms/ouster.py 
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
class PointCloud(spec.Transform[types.OSDepth, types.PointCloud]):
    """Calculate point cloud from Ouster Lidar depth data.

    Args:
        min_range: minimum range to include in the point cloud; if `None`,
            all points are included.
    """

    def __init__(self, min_range: float | None = None) -> None:
        self.cache = ConfigCache()
        self.lut_cache: dict[int, _LidarXYZLut] = {}
        self.min_range = min_range

    def __call__(self, data: types.OSDepth) -> types.PointCloud:
        """Convert Ouster Lidar depth data to point cloud.

        Args:
            data: Ouster Lidar depth data with staggered measurements.

        Returns:
            Point cloud data.
        """
        _batch = data.rng.shape[:2]

        info = self.cache[data.intrinsics]
        mid = id(info)

        if mid not in self.lut_cache:
            self.lut_cache[mid] = _LidarXYZLut.from_intrinsics(info)
        lut = self.lut_cache[mid]

        xyz_all = lut.compute_xyz(
            data.rng.reshape(-1, *data.rng.shape[-2:])
        ).astype(np.float32)

        xyz = []
        for pc in xyz_all:
            if self.min_range is not None:
                valid = (np.linalg.norm(pc, axis=-1) >= self.min_range)
            else:
                valid = np.any(pc != 0, axis=-1)
            valid = valid & np.all(~np.isnan(pc), axis=-1)

            xyz.append(pc.reshape(-1, 3)[valid.reshape(-1)])

        padded = np.zeros(
            (len(xyz), max(x.shape[0] for x in xyz), 3), dtype=np.float32)
        length = np.array([x.shape[0] for x in xyz], dtype=np.int32)
        for i, x in enumerate(xyz):
            padded[i, :x.shape[0]] = x

        return types.PointCloud(
            xyz=padded.reshape(*_batch, -1, 3),
            length=length.reshape(_batch),
            timestamps=data.timestamps)

__call__ 

__call__(data: OSDepth) -> PointCloud

Convert Ouster Lidar depth data to point cloud.

Parameters:

Name Type Description Default
data OSDepth

Ouster Lidar depth data with staggered measurements.

 required

Returns:

Type Description
PointCloud

Point cloud data.
Source code in format/src/roverd/transforms/ouster.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
def __call__(self, data: types.OSDepth) -> types.PointCloud:
    """Convert Ouster Lidar depth data to point cloud.

    Args:
        data: Ouster Lidar depth data with staggered measurements.

    Returns:
        Point cloud data.
    """
    _batch = data.rng.shape[:2]

    info = self.cache[data.intrinsics]
    mid = id(info)

    if mid not in self.lut_cache:
        self.lut_cache[mid] = _LidarXYZLut.from_intrinsics(info)
    lut = self.lut_cache[mid]

    xyz_all = lut.compute_xyz(
        data.rng.reshape(-1, *data.rng.shape[-2:])
    ).astype(np.float32)

    xyz = []
    for pc in xyz_all:
        if self.min_range is not None:
            valid = (np.linalg.norm(pc, axis=-1) >= self.min_range)
        else:
            valid = np.any(pc != 0, axis=-1)
        valid = valid & np.all(~np.isnan(pc), axis=-1)

        xyz.append(pc.reshape(-1, 3)[valid.reshape(-1)])

    padded = np.zeros(
        (len(xyz), max(x.shape[0] for x in xyz), 3), dtype=np.float32)
    length = np.array([x.shape[0] for x in xyz], dtype=np.int32)
    for i, x in enumerate(xyz):
        padded[i, :x.shape[0]] = x

    return types.PointCloud(
        xyz=padded.reshape(*_batch, -1, 3),
        length=length.reshape(_batch),
        timestamps=data.timestamps)

roverd.transforms.ros

Rover to rosbag conversion.

Warning

This module is not automatically imported; you will need to explicitly import it:

from roverd.transforms import ros

You will also need to have the ros extra installed.

roverd.transforms.ros.rover_to_rosbag 

rover_to_rosbag(
    out: str, imu: IMU, lidar: OSLidarDepth, min_range: float | None = None
) -> None

Write lidar and IMU data to a ros bag (to run SLAM, e.g., cartographer).

Parameters:

Name Type Description Default
out str

output rosbag file path.

 required
imu IMU

IMU sensor.

 required
lidar OSLidarDepth

Lidar sensor.

 required
min_range float | None

minimum range for lidar points.

 None
Source code in format/src/roverd/transforms/ros.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
109
110
111
112
113
114
def rover_to_rosbag(
    out: str, imu: sensors.IMU, lidar: sensors.OSLidarDepth,
    min_range: float | None = None
) -> None:
    """Write lidar and IMU data to a ros bag (to run SLAM, e.g., cartographer).

    Args:
        out: output rosbag file path.
        imu: IMU sensor.
        lidar: Lidar sensor.
        min_range: minimum range for lidar points.
    """
    os.makedirs(os.path.dirname(out), exist_ok=True)

    with Writer(out) as writer:

        msgtype = types.sensor_msgs__msg__Imu.__msgtype__
        connection = writer.add_connection("/imu", msgtype, latching=1)
        data = zip(
            Rotation.from_euler('XYZ', imu["rot"].read()).as_quat(False),
            imu["acc"].read(), imu["avel"].read(),
            *ts_sec_ns(imu.metadata.timestamps))
        for rot, acc, avel, sec, nsec, ts in tqdm(data, total=len(imu), desc="imu"):
            msg = __sensor_msgs_imu(rot, acc, avel, sec, nsec)
            writer.write(
                connection, ts,
                cdr_to_ros1(serialize_cdr(msg, msgtype), msgtype))  # type: ignore

        to_pointcloud = ouster.PointCloud(min_range=min_range)
        msgtype = types.sensor_msgs__msg__PointCloud2.__msgtype__
        connection = writer.add_connection("/points2", msgtype, latching=1)
        data = utils.Prefetch(
            zip(lidar.stream(), *ts_sec_ns(lidar.metadata.timestamps)))
        for raw, sec, nsec, ts in tqdm(data, total=len(lidar), desc="lidar"):
            points = to_pointcloud(raw).xyz
            msg = __sensor_msgs_pointcloud2(points, sec, nsec)
            writer.write(
                connection, ts,
                cdr_to_ros1(serialize_cdr(msg, msgtype), msgtype))  # type: ignore

roverd.transforms.ros.ts_sec_ns 

ts_sec_ns(t)

Convert float64 timestamp to seconds, ns remainder, and ns timestamp.

Source code in format/src/roverd/transforms/ros.py 
29
30
31
32
33
34
def ts_sec_ns(t):
    """Convert float64 timestamp to seconds, ns remainder, and ns timestamp."""
    ts = (t * 1e9).astype(np.int64)
    sec = t.astype(np.int64)
    nsec = ts % (1000 * 1000 * 1000)
    return sec, nsec, ts