tools.lib package

Submodules

tools.lib.api module

exception tools.lib.api.APIError[source]

Bases: Exception

class tools.lib.api.CommaApi(token=None)[source]

Bases: object

get(endpoint, **kwargs)[source]
post(endpoint, **kwargs)[source]
request(method, endpoint, **kwargs)[source]
exception tools.lib.api.UnauthorizedError[source]

Bases: Exception

tools.lib.auth module

Usage:

usage: auth.py [-h] [{google,apple,github,jwt}] [jwt]

Login to your comma account

positional arguments:
  {google,apple,github,jwt}
  jwt

optional arguments:
  -h, --help            show this help message and exit

Examples:

./auth.py  # Log in with google account
./auth.py github  # Log in with GitHub Account
./auth.py jwt ey......hw  # Log in with a JWT from https://jwt.comma.ai, for use in CI
class tools.lib.auth.ClientRedirectHandler(request, client_address, server)[source]

Bases: BaseHTTPRequestHandler

do_GET()[source]
log_message(format, *args)[source]

Log an arbitrary message.

This is used by all other logging functions. Override it if you have specific logging wishes.

The first argument, FORMAT, is a format string for the message to be logged. If the format string contains any % escapes requiring parameters, they should be specified as subsequent arguments (it’s just like printf!).

The client ip and current date/time are prefixed to every message.

class tools.lib.auth.ClientRedirectServer(server_address, RequestHandlerClass, bind_and_activate=True)[source]

Bases: HTTPServer

query_params: Dict[str, Any] = {}
tools.lib.auth.login(method)[source]

tools.lib.auth_config module

exception tools.lib.auth_config.MissingAuthConfigError[source]

Bases: Exception

tools.lib.auth_config.clear_token()[source]
tools.lib.auth_config.get_token()[source]
tools.lib.auth_config.set_token(token)[source]

tools.lib.bootlog module

class tools.lib.bootlog.Bootlog(url: str)[source]

Bases: object

property datetime: datetime
property dongle_id: str
property timestamp: str
property url: str
tools.lib.bootlog.get_bootlog_from_id(bootlog_id: str) Optional[Bootlog][source]
tools.lib.bootlog.get_bootlogs(dongle_id: str) List[Bootlog][source]

tools.lib.cache module

tools.lib.cache.cache_path_for_file_path(fn, cache_prefix=None)[source]

tools.lib.exceptions module

exception tools.lib.exceptions.DataUnreadableError[source]

Bases: Exception

tools.lib.filereader module

tools.lib.filereader.FileReader(fn, debug=False)[source]

tools.lib.framereader module

class tools.lib.framereader.BaseFrameReader[source]

Bases: object

close()[source]
get(num, count=1, pix_fmt='yuv420p')[source]
class tools.lib.framereader.DoNothingContextManager[source]

Bases: object

tools.lib.framereader.FrameIterator(fn, pix_fmt, **kwargs)[source]
tools.lib.framereader.FrameReader(fn, cache_prefix=None, readahead=False, readbehind=False, index_data=None)[source]
class tools.lib.framereader.FrameType(value)[source]

Bases: IntEnum

An enumeration.

h265_stream = 2
raw = 1
tools.lib.framereader.GOPFrameIterator(gop_reader, pix_fmt)[source]
class tools.lib.framereader.GOPFrameReader(readahead=False, readbehind=False)[source]

Bases: BaseFrameReader

close()[source]
get(num, count=1, pix_fmt='yuv420p')[source]
class tools.lib.framereader.GOPReader[source]

Bases: object

get_gop(num)[source]
class tools.lib.framereader.RawData(f)[source]

Bases: object

read(i)[source]
class tools.lib.framereader.RawFrameReader(fn)[source]

Bases: BaseFrameReader

get(num, count=1, pix_fmt='yuv420p')[source]
load_and_debayer(img)[source]
class tools.lib.framereader.StreamFrameReader(fn, frame_type, index_data, readahead=False, readbehind=False)[source]

Bases: StreamGOPReader, GOPFrameReader

class tools.lib.framereader.StreamGOPReader(fn, frame_type, index_data)[source]

Bases: GOPReader

get_gop(num)[source]
class tools.lib.framereader.VideoStreamDecompressor(fn, vid_fmt, w, h, pix_fmt)[source]

Bases: object

read()[source]
write_thread()[source]
tools.lib.framereader.cache_fn(func)[source]
tools.lib.framereader.decompress_video_data(rawdat, vid_fmt, w, h, pix_fmt)[source]
tools.lib.framereader.ffprobe(fn, fmt=None)[source]
tools.lib.framereader.fingerprint_video(fn)[source]
tools.lib.framereader.get_video_index(fn, frame_type, cache_prefix=None)[source]
tools.lib.framereader.index_stream(fn, typ)[source]
tools.lib.framereader.index_video(fn, frame_type=None, cache_prefix=None)[source]
tools.lib.framereader.index_videos(camera_paths, cache_prefix=None)[source]

Requires that paths in camera_paths are contiguous and of the same type.

tools.lib.framereader.read_file_check_size(f, sz, cookie)[source]
tools.lib.framereader.rgb24tonv12(rgb)[source]
tools.lib.framereader.rgb24toyuv(rgb)[source]
tools.lib.framereader.rgb24toyuv420(rgb)[source]
tools.lib.framereader.vidindex(fn, typ)[source]

tools.lib.helpers module

class tools.lib.helpers.RE[source]

Bases: object

BOOTLOG_NAME = '(?P<dongle_id>[a-z0-9]{16})[|_/](?P<timestamp>[0-9]{4}-[0-9]{2}-[0-9]{2}--[0-9]{2}-[0-9]{2}-[0-9]{2})'
DONGLE_ID = '(?P<dongle_id>[a-z0-9]{16})'
EXPLORER_FILE = '^(?P<segment_name>(?P<dongle_id>[a-z0-9]{16})[|_/](?P<timestamp>[0-9]{4}-[0-9]{2}-[0-9]{2}--[0-9]{2}-[0-9]{2}-[0-9]{2})(?:--|/)(?P<segment_num>[0-9]+))--(?P<file_name>[a-z]+\\.[a-z0-9]+)$'
OP_SEGMENT_DIR = '^(?P<segment_name>(?P<dongle_id>[a-z0-9]{16})[|_/](?P<timestamp>[0-9]{4}-[0-9]{2}-[0-9]{2}--[0-9]{2}-[0-9]{2}-[0-9]{2})(?:--|/)(?P<segment_num>[0-9]+))$'
ROUTE_NAME = '(?P<dongle_id>[a-z0-9]{16})[|_/](?P<timestamp>[0-9]{4}-[0-9]{2}-[0-9]{2}--[0-9]{2}-[0-9]{2}-[0-9]{2})'
SEGMENT_NAME = '(?P<dongle_id>[a-z0-9]{16})[|_/](?P<timestamp>[0-9]{4}-[0-9]{2}-[0-9]{2}--[0-9]{2}-[0-9]{2}-[0-9]{2})(?:--|/)(?P<segment_num>[0-9]+)'
TIMESTAMP = '(?P<timestamp>[0-9]{4}-[0-9]{2}-[0-9]{2}--[0-9]{2}-[0-9]{2}-[0-9]{2})'
tools.lib.helpers.timestamp_to_datetime(t: str) datetime[source]

Convert an openpilot route timestamp to a python datetime

tools.lib.kbhit module

class tools.lib.kbhit.KBHit[source]

Bases: object

static getarrow() int[source]

Returns an arrow-key code after kbhit() has been called. Codes are 0 : up 1 : right 2 : down 3 : left Should not be called in the same program as getch().

static getch() str[source]

Returns a keyboard character after kbhit() has been called. Should not be called in the same program as getarrow().

static kbhit()[source]

Returns True if keyboard character was hit, False otherwise.

set_kbhit_terminal() None[source]

Save old terminal settings for closure, remove ICANON & ECHO flags.

set_normal_term() None[source]

Resets to normal terminal. On Windows this is a no-op.

tools.lib.logreader module

class tools.lib.logreader.LogReader(fn, canonicalize=True, only_union_types=False, sort_by_time=False, dat=None)[source]

Bases: object

classmethod from_bytes(dat)[source]
class tools.lib.logreader.MultiLogIterator(log_paths, sort_by_time=False)[source]

Bases: object

reset()[source]
seek(ts)[source]
tell()[source]
tools.lib.logreader.logreader_from_route_or_segment(r, sort_by_time=False)[source]

tools.lib.route module

class tools.lib.route.Route(name, data_dir=None)[source]

Bases: object

camera_paths()[source]
dcamera_paths()[source]
ecamera_paths()[source]
log_paths()[source]
property name
qcamera_paths()[source]
qlog_paths()[source]
property segments
class tools.lib.route.RouteName(name_str: str)[source]

Bases: object

property canonical_name: str
property dongle_id: str
property time_str: str
class tools.lib.route.Segment(name, log_path, qlog_path, camera_path, dcamera_path, ecamera_path, qcamera_path)[source]

Bases: object

property name
class tools.lib.route.SegmentName(name_str: str, allow_route_name=False)[source]

Bases: object

property canonical_name: str
property data_dir: Optional[str]
property dongle_id: str
property route_name: RouteName
property segment_num: int
property time_str: str

tools.lib.url_file module

class tools.lib.url_file.URLFile(url, debug=False, cache=None)[source]

Bases: object

get_length()[source]
get_length_online()[source]
property name

Returns a local path to file with the URLFile’s contents.

This can be used to interface with modules that require local files.

read(ll=None)[source]
read_aux(ll=None)[source]
seek(pos)[source]
tools.lib.url_file.hash_256(link)[source]

Module contents