ouster.sdk.pcap

Module contents

Copyright (c) 2021, Ouster, Inc. All rights reserved.

Pcap tools to record/read/write Ouster sensor data.

ouster.sdk.pcap.pcap

Copyright (c) 2021, Ouster, Inc. All rights reserved.

record(packets, pcap_path, *, src_ip='127.0.0.1', dst_ip='127.0.0.1', lidar_port=7502, imu_port=7503, zm_port=7504, use_sll_encapsulation=False)[source]

Record a sequence of sensor packets to a pcap file.

Parameters:

  • packets (Iterable[Packet]) – A (finite!) sequence of packets

  • pcap_path (str) – Path of the output pcap file

  • src_ip (str) – Source IP to use for all packets

  • dst_ip (str) – Destination IP to use for all packets

  • lidar_port (int) – Src/dst port to use for lidar packets

  • imu_port (int) – Src/dst port to use for imu packets

  • use_sll_encapsulation (bool) – Use sll encapsulation

  • zm_port (int)

Return type:

int

Returns:

Number of packets captured

Classes

Functions

PacketInfo

class PacketInfo(self)
__annotations__ = {'dst_ip': 'None', 'dst_port': 'None', 'encapsulation_protocol': 'int', 'file_offset': 'int', 'fragments_in_packet': 'int', 'ip_version': 'int', 'network_protocol': 'int', 'payload_size': 'int', 'src_ip': 'None', 'src_port': 'None', 'timestamp': 'None'}
__init__(self) → None
__module__ = 'ouster.sdk.pcap'
__new__(**kwargs)
__repr__(self) → str
property dst_ip: None

(self) -> str

property dst_port: None

(self) -> int

property encapsulation_protocol: int

(self) -> int

property file_offset: int

(self) -> int

property fragments_in_packet: int

(self) -> int

property ip_version: int

(self) -> int

property network_protocol: int

(self) -> int

property payload_size: int

(self) -> int

property src_ip: None

(self) -> str

property src_port: None

(self) -> int

property timestamp: None

(self) -> float

PcapDuplicatePortException

class PcapDuplicatePortException
__module__ = 'ouster.sdk.pcap'
__weakref__

list of weak references to the object

PcapFrameSetSource

class PcapFrameSetSource(self, file: str, **kwargs)
__annotations__ = {'id_error_count': 'int', 'size_error_count': 'int'}
__init__(self, file: str, **kwargs) → None
__iter__(self) → collections.abc.Iterator[ouster.sdk.core.FrameSet]
__module__ = 'ouster.sdk.pcap'
__new__(**kwargs)
property id_error_count: int

(self) -> int

property size_error_count: int

(self) -> int

PcapPacketSource

class PcapPacketSource(self, file: str, **kwargs)
__annotations__ = {'id_error_count': 'int', 'size_error_count': 'int'}
__init__(self, file: str, **kwargs) → None
__module__ = 'ouster.sdk.pcap'
__new__(**kwargs)
property id_error_count: int

(self) -> int

property size_error_count: int

(self) -> int

ouster.sdk.pcap.packet_iter

class RecordingPacketSource(source, prefix_path, *, sensor_idx=-1, n_seconds=0.0, n_frames, chunk_size=0, src_ip='127.0.0.1', dst_ip='127.0.0.1', lidar_port=-1, imu_port=-1, use_sll_encapsulation=False, overwrite=True)[source]

Bases: object

Parameters:

  • source (PacketSource)

  • prefix_path (str)

  • sensor_idx (int)

  • n_seconds (float)

  • n_frames (int | None)

  • chunk_size (int)

  • src_ip (str)

  • dst_ip (str)

  • lidar_port (int)

  • imu_port (int)

  • use_sll_encapsulation (bool)

  • overwrite (bool)

ichunked_before(it, pred)[source]

Return the given stream chunked by the predicate.

Each sub-iterator will be fully consumed when the next chunk is requested. No caching of unused items is performed, so client code should evaluate sub-iterators (e.g. into lists) to avoid dropping items.

This should behave same as more_itertools.split_before, except that chunks aren’t eagerly evaluated into lists. This makes it safe to use on streams where it’s possible that pred never evaluates to true. :rtype: Iterator[Iterator[TypeVar(T)]]

Parameters:

  • it (Iterable[T])

  • pred (Callable[[T], bool])

Return type:

Iterator[Iterator[T]]

Submodules