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
PS > ouster-cli.exe 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 = "") -> Iterable[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 saved = False
82
83 def save_iter():
84 nonlocal field_names, field_fmts, saved
85 try:
86 if saved:
87 for scan in scan_iter():
88 yield scan
89 return
90 for idx, scans in enumerate(scan_iter()):
91 for lidar_idx, scan in enumerate(scans):
92 if scan is None:
93 continue
94
95 # Initialize the field names for csv header
96 if lidar_idx not in field_names or lidar_idx not in field_fmts:
97 field_names[lidar_idx], field_fmts[lidar_idx] = get_fields_info(scan)
98
99 # Copy per-column timestamps and measurement_ids for each beam
100 timestamps = np.tile(scan.timestamp, (scan.h, 1))
101 measurement_ids = np.tile(scan.measurement_id, (scan.h, 1))
102
103 # Grab channel data
104 fields_values = [scan.field(ch) for ch in scan.fields]
105
106 frame = np.dstack((timestamps, row_layer[lidar_idx], column_layer_staggered[lidar_idx],
107 measurement_ids, *fields_values))
108
109 # Output points in "image" vs. staggered order
110 frame = destagger(info, frame)
111
112 # Destagger XYZ separately since it has a different type
113 xyz = xyzlut[lidar_idx](scan.field(ChanField.RANGE))
114 xyz_destaggered = destagger(info, xyz)
115
116 if ChanField.RANGE2 in scan.fields:
117 xyz2 = xyzlut[lidar_idx](scan.field(ChanField.RANGE2))
118 xyz2_destaggered = destagger(info, xyz2)
119
120 # Get all data as one H x W x num fields int64 array for savetxt()
121 frame = np.dstack(tuple(map(lambda x: x.astype(object),
122 (frame, xyz_destaggered, xyz2_destaggered))))
123 else:
124 # Get all data as one H x W x num fields int64 array for savetxt()
125 frame = np.dstack(tuple(map(lambda x: x.astype(object),
126 (frame, xyz_destaggered))))
127
128 frame_colmajor = np.swapaxes(frame, 0, 1)
129
130 # Write csv out to file
131 csv_path = f"{filenames[lidar_idx]}_{idx}.csv"
132 print(f'write frame index #{idx}, to file: {csv_path}')
133
134 if os.path.isfile(csv_path) and not overwrite:
135 print(_file_exists_error(csv_path))
136 exit(1)
137
138 header = '\n'.join([f'frame num: {idx}', field_names[lidar_idx]])
139
140 np.savetxt(csv_path,
141 frame_colmajor.reshape(-1, frame.shape[2]),
142 fmt=field_fmts[lidar_idx],
143 delimiter=',',
144 header=header)
145
146 yield scan
147 except (KeyboardInterrupt, StopIteration):
148 pass
149 finally:
150 saved = True
151
152 # type ignored because generators are tricky to mypy
153 return save_iter # type: ignore
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 pcap-to-las --scan-num 5
PS > py -3 -m ouster.sdk.examples.pcap $SAMPLE_DATA_PCAP_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 pcap-to-pcd --scan-num 5
PS > py -3 -m ouster.sdk.examples.pcap $SAMPLE_DATA_PCAP_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 pcap-to-ply --scan-num 5
PS > py -3 -m ouster.sdk.examples.pcap $SAMPLE_DATA_PCAP_PATH pcap-to-ply --scan-num 5
Checkout the examples.pcap.pcap_to_ply()
documentation for the example source code.