Converting PCAPs to Other Formats

PCAPs to CSV

Sometimes we want to get a point cloud (XYZ + other fields) as a CSV file for further analysis with other tools.

To convert the first 5 scans of our sample data from a pcap file, you can try:

$ ouster-cli source --meta $SAMPLE_DATA_JSON_PATH $SAMPLE_DATA_PCAP_PATH slice 0:5 save output.csv

The following function implements the pcap to csv conversion above.

  1def source_to_csv_iter(scan_iter: Iterator[List[Optional[LidarScan]]], infos: List[SensorInfo],
  2                       prefix: str = "", dir: str = "", overwrite: bool = True,
  3                       filename: str = "") -> Iterator[List[Optional[LidarScan]]]:
  4    """Create a CSV saving iterator from a LidarScan iterator
  5
  6    The number of saved lines per csv file is always H x W, which corresponds to
  7    a full 2D image representation of a lidar scan.
  8
  9    Each line in a csv file is (for DUAL profile):
 10
 11        TIMESTAMP (ns), RANGE (mm), RANGE2 (mm), SIGNAL (photons),
 12            SIGNAL2 (photons), REFLECTIVITY (%), REFLECTIVITY2 (%),
 13            NEAR_IR (photons), X (m), Y (m), Z (m), X2 (m), Y2 (m), Z2(m),
 14            MEASUREMENT_ID, ROW, COLUMN
 15    """
 16
 17    dual_formats = [UDPProfileLidar.PROFILE_LIDAR_RNG19_RFL8_SIG16_NIR16_DUAL,
 18                    UDPProfileLidar.PROFILE_LIDAR_FUSA_RNG15_RFL8_NIR8_DUAL]
 19    for info in infos:
 20        if info.format.udp_profile_lidar in dual_formats:
 21            print("Note: You've selected to convert a dual returns pcap to CSV. Each row "
 22                  "will represent a single pixel, so that both returns for that pixel will "
 23                  "be on a single row. As this is an example we provide for getting "
 24                  "started, we realize that you may have conversion needs which are not met "
 25                  "by this function. You can find the source code on the Python SDK "
 26                  "documentation website to modify it for your own needs.")
 27            break
 28
 29    # Build filenames
 30    filenames = []
 31    for info in infos:
 32        name = determine_filename(filename=filename, info=info, extension=".csv", prefix=prefix, dir=dir)
 33
 34        name = name[0:-4]  # remove extension
 35        filenames.append(name)
 36        click.echo(f"Saving CSV files at {name}_XXX.csv")
 37
 38    create_directories_if_missing(filenames[0])
 39
 40    # Construct csv header and data format
 41    def get_fields_info(scan: LidarScan) -> Tuple[str, List[str]]:
 42        field_names = 'TIMESTAMP (ns), ROW, DESTAGGERED IMAGE COLUMN, MEASUREMENT_ID'
 43        field_fmts = ['%d'] * 4
 44        dual = ChanField.RANGE2 in scan.fields
 45        for chan_field in scan.fields:
 46            field_names += f', {chan_field}'
 47            if chan_field in [ChanField.RANGE, ChanField.RANGE2]:
 48                field_names += ' (mm)'
 49            if chan_field in [ChanField.REFLECTIVITY, ChanField.REFLECTIVITY2]:
 50                field_names += ' (%)'
 51            if chan_field in [ChanField.SIGNAL, ChanField.SIGNAL2,
 52                    ChanField.NEAR_IR]:
 53                field_names += ' (photons)'
 54            field_fmts.append('%d')
 55        field_names += ', X1 (m), Y1 (m), Z1 (m)'
 56        field_fmts.extend(3 * ['%.4f'])
 57        if dual:
 58            field_names += ', X2 (m), Y2 (m), Z2 (m)'
 59            field_fmts.extend(3 * ['%.4f'])
 60        return field_names, field_fmts
 61
 62    field_names: Dict[int, str] = {}
 63    field_fmts: Dict[int, List[str]] = {}
 64
 65    # {recompute xyzlut to save computation in a loop
 66    xyzlut = []
 67    row_layer = []
 68    column_layer_staggered = []
 69    for info in infos:
 70        xyzlut.append(XYZLut(info))
 71
 72        row_layer.append(np.fromfunction(lambda i, j: i,
 73                (info.format.pixels_per_column,
 74                    info.format.columns_per_frame), dtype=int))
 75        column_layer = np.fromfunction(lambda i, j: j,
 76                (info.format.pixels_per_column,
 77                    info.format.columns_per_frame), dtype=int)
 78        column_layer_staggered.append(destagger(info, column_layer,
 79                inverse=True))
 80
 81    def save_iter():
 82        nonlocal field_names, field_fmts
 83        try:
 84            for idx, scans in enumerate(scan_iter):
 85                for lidar_idx, scan in enumerate(scans):
 86                    if scan is None:
 87                        continue
 88
 89                    # Initialize the field names for csv header
 90                    if lidar_idx not in field_names or lidar_idx not in field_fmts:
 91                        field_names[lidar_idx], field_fmts[lidar_idx] = get_fields_info(scan)
 92
 93                    # Copy per-column timestamps and measurement_ids for each beam
 94                    timestamps = np.tile(scan.timestamp, (scan.h, 1))
 95                    measurement_ids = np.tile(scan.measurement_id, (scan.h, 1))
 96
 97                    # Grab channel data
 98                    fields_values = [scan.field(ch) for ch in scan.fields]
 99
100                    frame = np.dstack((timestamps, row_layer[lidar_idx], column_layer_staggered[lidar_idx],
101                        measurement_ids, *fields_values))
102
103                    # Output points in "image" vs. staggered order
104                    frame = destagger(info, frame)
105
106                    # Destagger XYZ separately since it has a different type
107                    xyz = xyzlut[lidar_idx](scan.field(ChanField.RANGE))
108                    xyz_destaggered = destagger(info, xyz)
109
110                    if ChanField.RANGE2 in scan.fields:
111                        xyz2 = xyzlut[lidar_idx](scan.field(ChanField.RANGE2))
112                        xyz2_destaggered = destagger(info, xyz2)
113
114                        # Get all data as one H x W x num fields int64 array for savetxt()
115                        frame = np.dstack(tuple(map(lambda x: x.astype(object),
116                            (frame, xyz_destaggered, xyz2_destaggered))))
117                    else:
118                        # Get all data as one H x W x num fields int64 array for savetxt()
119                        frame = np.dstack(tuple(map(lambda x: x.astype(object),
120                            (frame, xyz_destaggered))))
121
122                    frame_colmajor = np.swapaxes(frame, 0, 1)
123
124                    # Write csv out to file
125                    csv_path = f"{filenames[lidar_idx]}_{idx}.csv"
126                    print(f'write frame index #{idx}, to file: {csv_path}')
127
128                    if os.path.isfile(csv_path) and not overwrite:
129                        print(_file_exists_error(csv_path))
130                        exit(1)
131
132                    header = '\n'.join([f'frame num: {idx}', field_names[lidar_idx]])
133
134                    np.savetxt(csv_path,
135                            frame_colmajor.reshape(-1, frame.shape[2]),
136                            fmt=field_fmts[lidar_idx],
137                            delimiter=',',
138                            header=header)
139
140                yield scan
141        except (KeyboardInterrupt, StopIteration):
142            pass
143
144    return save_iter()

Because we stored the scan as structured 2D images, we can easily recover it by loading it back into a numpy.ndarray and continuing to use it as a 2D image.

import numpy as np

# read array from CSV
frame = np.loadtxt('my_frame_00000.csv', delimiter=',')

# convert back to "fat" 2D image [H x W x num_fields] shape
frame = frame.reshape((128, -1, frame.shape[1]))

We used 128 while restoring 2D image from a CSV file because it’s the number of channels of our OS-1-128.pcap sample data recording.

PCAPs to LAS

To convert to the first 5 scans of our sample data from a pcap file to LAS, you can try:

$ python3 -m ouster.sdk.examples.pcap $SAMPLE_DATA_PCAP_PATH $SAMPLE_DATA_JSON_PATH pcap-to-las --scan-num 5

Checkout the examples.pcap.pcap_to_las() documentation for the example source code.

PCAPs to PCD

To convert to the first 5 scans of our sample data from a pcap file to PCD, you can try:

$ python3 -m ouster.sdk.examples.pcap $SAMPLE_DATA_PCAP_PATH $SAMPLE_DATA_JSON_PATH pcap-to-pcd --scan-num 5

Checkout the examples.pcap.pcap_to_pcd() documentation for the example source code.

PCAPs to PLY

Here we will reuse the PCAP to PCD function that uses Open3d and will exploit the extensive Open3d File IO that gives us an easy way to save the loaded point cloud to PLY. Alternative ways are available via plyfile library.

To convert to the first 5 scans of our sample data from a pcap file to PLY, you can try:

$ python3 -m ouster.sdk.examples.pcap $SAMPLE_DATA_PCAP_PATH $SAMPLE_DATA_JSON_PATH pcap-to-ply --scan-num 5

Checkout the examples.pcap.pcap_to_ply() documentation for the example source code.