dragonpilot beta3

date: 2024-02-12T12:52:55
commit: 0f5fdc8c6be06fb8901e8be8f3fa7d00596eb1f4

tools/bodyteleop/web.py

@@ -6,9 +6,10 @@ import os
 import ssl
 import subprocess
 
-from aiohttp import web, ClientSession
 import pyaudio
 import wave
+from aiohttp import web
+from aiohttp import ClientSession
 
 from openpilot.common.basedir import BASEDIR
 from openpilot.system.webrtc.webrtcd import StreamRequestBody
@@ -22,7 +23,7 @@ WEBRTCD_HOST, WEBRTCD_PORT = "localhost", 5001
 
 
 ## UTILS
-async def play_sound(sound):
+async def play_sound(sound: str):
   SOUNDS = {
     "engage": "selfdrive/assets/sounds/engage.wav",
     "disengage": "selfdrive/assets/sounds/disengage.wav",
@@ -51,7 +52,7 @@ async def play_sound(sound):
     p.terminate()
 
 ## SSL
-def create_ssl_cert(cert_path, key_path):
+def create_ssl_cert(cert_path: str, key_path: str):
   try:
     proc = subprocess.run(f'openssl req -x509 -newkey rsa:4096 -nodes -out {cert_path} -keyout {key_path} \
                           -days 365 -subj "/C=US/ST=California/O=commaai/OU=comma body"',
@@ -75,17 +76,17 @@ def create_ssl_context():
   return ssl_context
 
 ## ENDPOINTS
-async def index(request):
+async def index(request: 'web.Request'):
   with open(os.path.join(TELEOPDIR, "static", "index.html"), "r") as f:
     content = f.read()
     return web.Response(content_type="text/html", text=content)
 
 
-async def ping(request):
+async def ping(request: 'web.Request'):
   return web.Response(text="pong")
 
 
-async def sound(request):
+async def sound(request: 'web.Request'):
   params = await request.json()
   sound_to_play = params["sound"]
 
@@ -93,7 +94,7 @@ async def sound(request):
   return web.json_response({"status": "ok"})
 
 
-async def offer(request):
+async def offer(request: 'web.Request'):
   params = await request.json()
   body = StreamRequestBody(params["sdp"], ["driver"], ["testJoystick"], ["carState"])
   body_json = json.dumps(dataclasses.asdict(body))

tools/joystick/joystickd.py

@@ -47,7 +47,7 @@ class Joystick:
     else:
       self.cancel_button = 'BTN_TRIGGER'
       accel_axis = 'ABS_Y'
-      steer_axis = 'ABS_RZ'
+      steer_axis = 'ABS_RX'
     self.min_axis_value = {accel_axis: 0., steer_axis: 0.}
     self.max_axis_value = {accel_axis: 255., steer_axis: 255.}
     self.axes_values = {accel_axis: 0., steer_axis: 0.}

tools/lib/README.md

@@ -32,20 +32,22 @@ for msg in lr:
     print(msg.carState.steeringAngleDeg)
 ```
 
-### MultiLogIterator
+### Segment Ranges
 
-`MultiLogIterator` is similar to `LogReader`, but reads multiple logs. 
+We also support a new format called a "segment range", where you can specify which segments from a route to load.
 
 ```python
-from openpilot.tools.lib.route import Route
-from openpilot.tools.lib.logreader import MultiLogIterator
 
-# setup a MultiLogIterator to read all the logs in the route
-r = Route("a2a0ccea32023010|2023-07-27--13-01-19")
-lr = MultiLogIterator(r.log_paths())
+lr = LogReader("a2a0ccea32023010|2023-07-27--13-01-19/4")   # 4th segment
+lr = LogReader("a2a0ccea32023010|2023-07-27--13-01-19/4:6") # 4th and 5th segment
+lr = LogReader("a2a0ccea32023010|2023-07-27--13-01-19/-1")  # last segment
+lr = LogReader("a2a0ccea32023010|2023-07-27--13-01-19/:5")  # first 5 segments
+lr = LogReader("a2a0ccea32023010|2023-07-27--13-01-19/1:")  # all except first segment
+```
 
-# print all the steering angles values from all the logs in the route
-for msg in lr:
-  if msg.which() == "carState":
-    print(msg.carState.steeringAngleDeg)
+and can select which type of logs to grab
+
+```python
+lr = LogReader("a2a0ccea32023010|2023-07-27--13-01-19/4/q") # get qlogs
+lr = LogReader("a2a0ccea32023010|2023-07-27--13-01-19/4/r") # get rlogs (default)
 ```

tools/lib/azure_container.py

@@ -0,0 +1,74 @@
+import os
+from datetime import datetime, timedelta
+from functools import lru_cache
+from pathlib import Path
+from typing import IO, Union
+
+
+TOKEN_PATH = Path("/data/azure_token")
+
+@lru_cache
+def get_azure_credential():
+  if "AZURE_TOKEN" in os.environ:
+    return os.environ["AZURE_TOKEN"]
+  elif TOKEN_PATH.is_file():
+    return TOKEN_PATH.read_text().strip()
+  else:
+    from azure.identity import AzureCliCredential
+    return AzureCliCredential()
+
+@lru_cache
+def get_container_sas(account_name: str, container_name: str):
+  from azure.storage.blob import BlobServiceClient, ContainerSasPermissions, generate_container_sas
+  start_time = datetime.utcnow()
+  expiry_time = start_time + timedelta(hours=1)
+  blob_service = BlobServiceClient(
+    account_url=f"https://{account_name}.blob.core.windows.net",
+    credential=get_azure_credential(),
+  )
+  return generate_container_sas(
+    account_name,
+    container_name,
+    user_delegation_key=blob_service.get_user_delegation_key(start_time, expiry_time),
+    permission=ContainerSasPermissions(read=True, write=True, list=True),
+    expiry=expiry_time,
+  )
+
+class AzureContainer:
+  def __init__(self, account, container):
+    self.ACCOUNT = account
+    self.CONTAINER = container
+
+  @property
+  def ACCOUNT_URL(self) -> str:
+    return f"https://{self.ACCOUNT}.blob.core.windows.net"
+
+  @property
+  def BASE_URL(self) -> str:
+    return f"{self.ACCOUNT_URL}/{self.CONTAINER}/"
+
+  def get_client_and_key(self):
+    from azure.storage.blob import ContainerClient
+    client = ContainerClient(self.ACCOUNT_URL, self.CONTAINER, credential=get_azure_credential())
+    key = get_container_sas(self.ACCOUNT, self.CONTAINER)
+    return client, key
+
+  def get_url(self, route_name: str, segment_num, log_type="rlog") -> str:
+    ext = "hevc" if log_type.endswith('camera') else "bz2"
+    return self.BASE_URL + f"{route_name.replace('|', '/')}/{segment_num}/{log_type}.{ext}"
+
+  def upload_bytes(self, data: Union[bytes, IO], blob_name: str) -> str:
+    from azure.storage.blob import BlobClient
+    blob = BlobClient(
+      account_url=self.ACCOUNT_URL,
+      container_name=self.CONTAINER,
+      blob_name=blob_name,
+      credential=get_azure_credential(),
+      overwrite=False,
+    )
+    blob.upload_blob(data)
+    return self.BASE_URL + blob_name
+
+  def upload_file(self, path: Union[str, os.PathLike], blob_name: str) -> str:
+    with open(path, "rb") as f:
+      return self.upload_bytes(f, blob_name)

tools/lib/bootlog.py

@@ -1,11 +1,10 @@
-import datetime
 import functools
 import re
 from typing import List, Optional
 
 from openpilot.tools.lib.auth_config import get_token
 from openpilot.tools.lib.api import CommaApi
-from openpilot.tools.lib.helpers import RE, timestamp_to_datetime
+from openpilot.tools.lib.helpers import RE
 
 
 @functools.total_ordering
@@ -17,8 +16,8 @@ class Bootlog:
     if not r:
       raise Exception(f"Unable to parse: {url}")
 
+    self._id = r.group('log_id')
     self._dongle_id = r.group('dongle_id')
-    self._timestamp = r.group('timestamp')
 
   @property
   def url(self) -> str:
@@ -29,25 +28,21 @@ class Bootlog:
     return self._dongle_id
 
   @property
-  def timestamp(self) -> str:
-    return self._timestamp
-
-  @property
-  def datetime(self) -> datetime.datetime:
-    return timestamp_to_datetime(self._timestamp)
+  def id(self) -> str:
+    return self._id
 
   def __str__(self):
-    return f"{self._dongle_id}|{self._timestamp}"
+    return f"{self._dongle_id}/{self._id}"
 
   def __eq__(self, b) -> bool:
     if not isinstance(b, Bootlog):
       return False
-    return self.datetime == b.datetime
+    return self.id == b.id
 
   def __lt__(self, b) -> bool:
     if not isinstance(b, Bootlog):
       return False
-    return self.datetime < b.datetime
+    return self.id < b.id
 
 def get_bootlog_from_id(bootlog_id: str) -> Optional[Bootlog]:
   # TODO: implement an API endpoint for this

tools/lib/comma_car_segments.py

@@ -0,0 +1,81 @@
+import os
+import requests
+
+# Forks with additional car support can fork the commaCarSegments repo on huggingface or host the LFS files themselves
+COMMA_CAR_SEGMENTS_REPO = os.environ.get("COMMA_CAR_SEGMENTS_REPO", "https://huggingface.co/datasets/commaai/commaCarSegments")
+COMMA_CAR_SEGMENTS_BRANCH = os.environ.get("COMMA_CAR_SEGMENTS_BRANCH", "main")
+COMMA_CAR_SEGMENTS_LFS_INSTANCE = os.environ.get("COMMA_CAR_SEGMENTS_LFS_INSTANCE", COMMA_CAR_SEGMENTS_REPO)
+
+def get_comma_car_segments_database():
+  return requests.get(get_repo_raw_url("database.json")).json()
+
+
+# Helpers related to interfacing with the commaCarSegments repository, which contains a collection of public segments for users to perform validation on.
+
+def parse_lfs_pointer(text):
+  header, lfs_version = text.splitlines()[0].split(" ")
+  assert header == "version"
+  assert lfs_version == "https://git-lfs.github.com/spec/v1"
+
+  header, oid_raw = text.splitlines()[1].split(" ")
+  assert header == "oid"
+  header, oid = oid_raw.split(":")
+  assert header == "sha256"
+
+  header, size = text.splitlines()[2].split(" ")
+  assert header == "size"
+
+  return oid, size
+
+def get_lfs_file_url(oid, size):
+  data = {
+    "operation": "download",
+    "transfers": [ "basic" ],
+    "objects": [
+      {
+        "oid": oid,
+        "size": int(size)
+      }
+    ],
+    "hash_algo": "sha256"
+  }
+
+  headers = {
+    "Accept": "application/vnd.git-lfs+json",
+    "Content-Type": "application/vnd.git-lfs+json"
+  }
+
+  response = requests.post(f"{COMMA_CAR_SEGMENTS_LFS_INSTANCE}.git/info/lfs/objects/batch", json=data, headers=headers)
+
+  assert response.ok
+
+  obj = response.json()["objects"][0]
+
+  assert "error" not in obj, obj
+
+  return obj["actions"]["download"]["href"]
+
+def get_repo_raw_url(path):
+  if "huggingface" in COMMA_CAR_SEGMENTS_REPO:
+    return f"{COMMA_CAR_SEGMENTS_REPO}/raw/{COMMA_CAR_SEGMENTS_BRANCH}/{path}"
+
+def get_repo_url(path):
+  # Automatically switch to LFS if we are requesting a file that is stored in LFS
+
+  response = requests.head(get_repo_raw_url(path))
+
+  if "text/plain" in response.headers.get("content-type"):
+    # This is an LFS pointer, so download the raw data from lfs
+    response = requests.get(get_repo_raw_url(path))
+    assert response.status_code == 200
+    oid, size = parse_lfs_pointer(response.text)
+
+    return get_lfs_file_url(oid, size)
+  else:
+    # File has not been uploaded to LFS yet
+    # (either we are on a fork where the data hasn't been pushed to LFS yet, or the CI job to push hasn't finished)
+    return get_repo_raw_url(path)
+
+
+def get_url(route, segment, file="rlog.bz2"):
+  return get_repo_url(f"segments/{route.replace('|', '/')}/{segment}/{file}")

tools/lib/filereader.py

@@ -1,4 +1,5 @@
 import os
+
 from openpilot.tools.lib.url_file import URLFile
 
 DATA_ENDPOINT = os.getenv("DATA_ENDPOINT", "http://data-raw.comma.internal/")
@@ -8,6 +9,12 @@ def resolve_name(fn):
     return fn.replace("cd:/", DATA_ENDPOINT)
   return fn
 
+def file_exists(fn):
+  fn = resolve_name(fn)
+  if fn.startswith(("http://", "https://")):
+    return URLFile(fn).get_length_online() != -1
+  return os.path.exists(fn)
+
 def FileReader(fn, debug=False):
   fn = resolve_name(fn)
   if fn.startswith(("http://", "https://")):

tools/lib/helpers.py

@@ -7,8 +7,15 @@ TIME_FMT = "%Y-%m-%d--%H-%M-%S"
 class RE:
   DONGLE_ID =  r'(?P<dongle_id>[a-z0-9]{16})'
   TIMESTAMP = r'(?P<timestamp>[0-9]{4}-[0-9]{2}-[0-9]{2}--[0-9]{2}-[0-9]{2}-[0-9]{2})'
-  ROUTE_NAME = r'{}[|_/]{}'.format(DONGLE_ID, TIMESTAMP)
+  LOG_ID_V2 = r'(?P<count>[a-z0-9]{8})--(?P<uid>[a-z0-9]{10})'
+  LOG_ID = r'(?P<log_id>(?:{}|{}))'.format(TIMESTAMP, LOG_ID_V2)
+  ROUTE_NAME = r'(?P<route_name>{}[|_/]{})'.format(DONGLE_ID, LOG_ID)
   SEGMENT_NAME = r'{}(?:--|/)(?P<segment_num>[0-9]+)'.format(ROUTE_NAME)
+
+  INDEX = r'-?[0-9]+'
+  SLICE = r'(?P<start>{})?:?(?P<end>{})?:?(?P<step>{})?'.format(INDEX, INDEX, INDEX)
+  SEGMENT_RANGE = r'{}(?:--|/)?(?P<slice>({}))?/?(?P<selector>([qras]))?'.format(ROUTE_NAME, SLICE)
+
   BOOTLOG_NAME = ROUTE_NAME
 
   EXPLORER_FILE = r'^(?P<segment_name>{})--(?P<file_name>[a-z]+\.[a-z0-9]+)$'.format(SEGMENT_NAME)

tools/lib/logreader.py

@@ -1,82 +1,34 @@
 #!/usr/bin/env python3
-import os
-import sys
 import bz2
-import urllib.parse
+from functools import partial
+import multiprocessing
 import capnp
+import enum
+import numpy as np
+import os
+import pathlib
+import re
+import sys
+import tqdm
+import urllib.parse
 import warnings
 
-from typing import Iterable, Iterator
+from typing import Dict, Iterable, Iterator, List, Type
+from urllib.parse import parse_qs, urlparse
 
 from cereal import log as capnp_log
-from openpilot.tools.lib.filereader import FileReader
-from openpilot.tools.lib.route import Route, SegmentName
+from openpilot.common.swaglog import cloudlog
+from openpilot.tools.lib.comma_car_segments import get_url as get_comma_segments_url
+from openpilot.tools.lib.openpilotci import get_url
+from openpilot.tools.lib.filereader import FileReader, file_exists
+from openpilot.tools.lib.helpers import RE
+from openpilot.tools.lib.route import Route, SegmentRange
 
-LogIterable = Iterable[capnp._DynamicStructReader]
-
-# this is an iterator itself, and uses private variables from LogReader
-class MultiLogIterator:
-  def __init__(self, log_paths, sort_by_time=False):
-    self._log_paths = log_paths
-    self.sort_by_time = sort_by_time
-
-    self._first_log_idx = next(i for i in range(len(log_paths)) if log_paths[i] is not None)
-    self._current_log = self._first_log_idx
-    self._idx = 0
-    self._log_readers = [None]*len(log_paths)
-    self.start_time = self._log_reader(self._first_log_idx)._ts[0]
-
-  def _log_reader(self, i):
-    if self._log_readers[i] is None and self._log_paths[i] is not None:
-      log_path = self._log_paths[i]
-      self._log_readers[i] = LogReader(log_path, sort_by_time=self.sort_by_time)
-
-    return self._log_readers[i]
-
-  def __iter__(self) -> Iterator[capnp._DynamicStructReader]:
-    return self
-
-  def _inc(self):
-    lr = self._log_reader(self._current_log)
-    if self._idx < len(lr._ents)-1:
-      self._idx += 1
-    else:
-      self._idx = 0
-      self._current_log = next(i for i in range(self._current_log + 1, len(self._log_readers) + 1)
-                               if i == len(self._log_readers) or self._log_paths[i] is not None)
-      if self._current_log == len(self._log_readers):
-        raise StopIteration
-
-  def __next__(self):
-    while 1:
-      lr = self._log_reader(self._current_log)
-      ret = lr._ents[self._idx]
-      self._inc()
-      return ret
-
-  def tell(self):
-    # returns seconds from start of log
-    return (self._log_reader(self._current_log)._ts[self._idx] - self.start_time) * 1e-9
-
-  def seek(self, ts):
-    # seek to nearest minute
-    minute = int(ts/60)
-    if minute >= len(self._log_paths) or self._log_paths[minute] is None:
-      return False
-
-    self._current_log = minute
-
-    # HACK: O(n) seek afterward
-    self._idx = 0
-    while self.tell() < ts:
-      self._inc()
-    return True
-
-  def reset(self):
-    self.__init__(self._log_paths, sort_by_time=self.sort_by_time)
+LogMessage = Type[capnp._DynamicStructReader]
+LogIterable = Iterable[LogMessage]
 
 
-class LogReader:
+class _LogFileReader:
   def __init__(self, fn, canonicalize=True, only_union_types=False, sort_by_time=False, dat=None):
     self.data_version = None
     self._only_union_types = only_union_types
@@ -106,10 +58,6 @@ class LogReader:
     self._ents = list(sorted(_ents, key=lambda x: x.logMonoTime) if sort_by_time else _ents)
     self._ts = [x.logMonoTime for x in self._ents]
 
-  @classmethod
-  def from_bytes(cls, dat):
-    return cls("", dat=dat)
-
   def __iter__(self) -> Iterator[capnp._DynamicStructReader]:
     for ent in self._ents:
       if self._only_union_types:
@@ -121,13 +69,224 @@ class LogReader:
       else:
         yield ent
 
-def logreader_from_route_or_segment(r, sort_by_time=False):
-  sn = SegmentName(r, allow_route_name=True)
-  route = Route(sn.route_name.canonical_name)
-  if sn.segment_num < 0:
-    return MultiLogIterator(route.log_paths(), sort_by_time=sort_by_time)
+
+class ReadMode(enum.StrEnum):
+  RLOG = "r" # only read rlogs
+  QLOG = "q" # only read qlogs
+  SANITIZED = "s" # read from the commaCarSegments database
+  AUTO = "a" # default to rlogs, fallback to qlogs
+  AUTO_INTERACIVE = "i" # default to rlogs, fallback to qlogs with a prompt from the user
+
+def create_slice_from_string(s: str):
+  m = re.fullmatch(RE.SLICE, s)
+  assert m is not None, f"Invalid slice: {s}"
+  start, end, step = m.groups()
+  start = int(start) if start is not None else None
+  end = int(end) if end is not None else None
+  step = int(step) if step is not None else None
+
+  if start is not None and ":" not in s and end is None and step is None:
+    return start
+  return slice(start, end, step)
+
+def default_valid_file(fn):
+  return fn is not None and file_exists(fn)
+
+def auto_strategy(rlog_paths, qlog_paths, interactive, valid_file):
+  # auto select logs based on availability
+  if any(rlog is None or not valid_file(rlog) for rlog in rlog_paths):
+    if interactive:
+      if input("Some rlogs were not found, would you like to fallback to qlogs for those segments? (y/n) ").lower() != "y":
+        return rlog_paths
+    else:
+      cloudlog.warning("Some rlogs were not found, falling back to qlogs for those segments...")
+
+    return [rlog if (valid_file(rlog)) else (qlog if (valid_file(qlog)) else None)
+                        for (rlog, qlog) in zip(rlog_paths, qlog_paths, strict=True)]
+  return rlog_paths
+
+def apply_strategy(mode: ReadMode, rlog_paths, qlog_paths, valid_file=default_valid_file):
+  if mode == ReadMode.RLOG:
+    return rlog_paths
+  elif mode == ReadMode.QLOG:
+    return qlog_paths
+  elif mode == ReadMode.AUTO:
+    return auto_strategy(rlog_paths, qlog_paths, False, valid_file)
+  elif mode == ReadMode.AUTO_INTERACIVE:
+    return auto_strategy(rlog_paths, qlog_paths, True, valid_file)
+
+def parse_slice(sr: SegmentRange):
+  s = create_slice_from_string(sr._slice)
+  if isinstance(s, slice):
+    if s.stop is None or s.stop < 0 or (s.start is not None and s.start < 0): # we need the number of segments in order to parse this slice
+      segs = np.arange(sr.get_max_seg_number()+1)
+    else:
+      segs = np.arange(s.stop + 1)
+    return segs[s]
   else:
-    return LogReader(route.log_paths()[sn.segment_num], sort_by_time=sort_by_time)
+    if s < 0:
+      s = sr.get_max_seg_number() + s + 1
+    return [s]
+
+def comma_api_source(sr: SegmentRange, mode: ReadMode):
+  segs = parse_slice(sr)
+
+  route = Route(sr.route_name)
+
+  rlog_paths = [route.log_paths()[seg] for seg in segs]
+  qlog_paths = [route.qlog_paths()[seg] for seg in segs]
+
+  # comma api will have already checked if the file exists
+  def valid_file(fn):
+    return fn is not None
+
+  return apply_strategy(mode, rlog_paths, qlog_paths, valid_file=valid_file)
+
+def internal_source(sr: SegmentRange, mode: ReadMode):
+  segs = parse_slice(sr)
+
+  def get_internal_url(sr: SegmentRange, seg, file):
+    return f"cd:/{sr.dongle_id}/{sr.timestamp}/{seg}/{file}.bz2"
+
+  rlog_paths = [get_internal_url(sr, seg, "rlog") for seg in segs]
+  qlog_paths = [get_internal_url(sr, seg, "qlog")  for seg in segs]
+
+  return apply_strategy(mode, rlog_paths, qlog_paths)
+
+def openpilotci_source(sr: SegmentRange, mode: ReadMode):
+  segs = parse_slice(sr)
+
+  rlog_paths = [get_url(sr.route_name, seg, "rlog") for seg in segs]
+  qlog_paths = [get_url(sr.route_name, seg, "qlog")  for seg in segs]
+
+  return apply_strategy(mode, rlog_paths, qlog_paths)
+
+def comma_car_segments_source(sr: SegmentRange, mode=ReadMode.RLOG):
+  segs = parse_slice(sr)
+
+  return [get_comma_segments_url(sr.route_name, seg) for seg in segs]
+
+def direct_source(file_or_url):
+  return [file_or_url]
+
+def get_invalid_files(files):
+  for f in files:
+    if f is None or not file_exists(f):
+      yield f
+
+def check_source(source, *args):
+  try:
+    files = source(*args)
+    assert next(get_invalid_files(files), None) is None
+    return None, files
+  except Exception as e:
+    return e, None
+
+def auto_source(sr: SegmentRange, mode=ReadMode.RLOG):
+  if mode == ReadMode.SANITIZED:
+    return comma_car_segments_source(sr, mode)
+
+  exceptions = []
+  # Automatically determine viable source
+  for source in [internal_source, openpilotci_source, comma_api_source, comma_car_segments_source]:
+    exception, ret = check_source(source, sr, mode)
+    if exception is None:
+      return ret
+    else:
+      exceptions.append(exception)
+
+  raise Exception(f"auto_source could not find any valid source, exceptions for sources: {exceptions}")
+
+def parse_useradmin(identifier):
+  if "useradmin.comma.ai" in identifier:
+    query = parse_qs(urlparse(identifier).query)
+    return query["onebox"][0]
+  return None
+
+def parse_cabana(identifier):
+  if "cabana.comma.ai" in identifier:
+    query = parse_qs(urlparse(identifier).query)
+    return query["route"][0]
+  return None
+
+def parse_direct(identifier):
+  if identifier.startswith(("http://", "https://", "cd:/")) or pathlib.Path(identifier).exists():
+    return identifier
+  return None
+
+def parse_indirect(identifier):
+  parsed = parse_useradmin(identifier) or parse_cabana(identifier)
+
+  if parsed is not None:
+    return parsed, comma_api_source, True
+
+  return identifier, None, False
+
+
+class LogReader:
+  def _parse_identifiers(self, identifier: str | List[str]):
+    if isinstance(identifier, list):
+      return [i for j in identifier for i in self._parse_identifiers(j)]
+
+    parsed, source, is_indirect = parse_indirect(identifier)
+
+    if not is_indirect:
+      direct_parsed = parse_direct(identifier)
+      if direct_parsed is not None:
+        return direct_source(identifier)
+
+    sr = SegmentRange(parsed)
+    mode = self.default_mode if sr.selector is None else ReadMode(sr.selector)
+    source = self.default_source if source is None else source
+
+    return source(sr, mode)
+
+  def __init__(self, identifier: str | List[str], default_mode=ReadMode.RLOG, default_source=auto_source, sort_by_time=False, only_union_types=False):
+    self.default_mode = default_mode
+    self.default_source = default_source
+    self.identifier = identifier
+
+    self.sort_by_time = sort_by_time
+    self.only_union_types = only_union_types
+
+    self.__lrs: Dict[int, _LogFileReader] = {}
+    self.reset()
+
+  def _get_lr(self, i):
+    if i not in self.__lrs:
+      self.__lrs[i] = _LogFileReader(self.logreader_identifiers[i])
+    return self.__lrs[i]
+
+  def __iter__(self):
+    for i in range(len(self.logreader_identifiers)):
+      yield from self._get_lr(i)
+
+  def _run_on_segment(self, func, i):
+    return func(self._get_lr(i))
+
+  def run_across_segments(self, num_processes, func):
+    with multiprocessing.Pool(num_processes) as pool:
+      ret = []
+      num_segs = len(self.logreader_identifiers)
+      for p in tqdm.tqdm(pool.imap(partial(self._run_on_segment, func), range(num_segs)), total=num_segs):
+        ret.extend(p)
+      return ret
+
+  def reset(self):
+    self.logreader_identifiers = self._parse_identifiers(self.identifier)
+    invalid_count = len(list(get_invalid_files(self.logreader_identifiers)))
+    assert invalid_count == 0, f"{invalid_count}/{len(self.logreader_identifiers)} invalid log(s) found, please ensure all logs \
+are uploaded or auto fallback to qlogs with '/a' selector at the end of the route name."
+
+  @staticmethod
+  def from_bytes(dat):
+    return _LogFileReader("", dat=dat)
+
+  def filter(self, msg_type: str):
+    return (getattr(m, m.which()) for m in filter(lambda m: m.which() == msg_type, self))
+
+  def first(self, msg_type: str):
+    return next(self.filter(msg_type), None)
 
 
 if __name__ == "__main__":

tools/lib/openpilotci.py

@@ -0,0 +1,12 @@
+from openpilot.tools.lib.openpilotcontainers import OpenpilotCIContainer
+
+def get_url(*args, **kwargs):
+  return OpenpilotCIContainer.get_url(*args, **kwargs)
+
+def upload_file(*args, **kwargs):
+  return OpenpilotCIContainer.upload_file(*args, **kwargs)
+
+def upload_bytes(*args, **kwargs):
+  return OpenpilotCIContainer.upload_bytes(*args, **kwargs)
+
+BASE_URL = OpenpilotCIContainer.BASE_URL

tools/lib/openpilotcontainers.py

@@ -0,0 +1,6 @@
+#!/usr/bin/env python3
+from openpilot.tools.lib.azure_container import AzureContainer
+
+OpenpilotCIContainer = AzureContainer("commadataci", "openpilotci")
+DataCIContainer = AzureContainer("commadataci", "commadataci")
+DataProdContainer = AzureContainer("commadata2", "commadata2")

tools/lib/route.py

@@ -1,5 +1,6 @@
 import os
 import re
+from functools import cache
 from urllib.parse import urlparse
 from collections import defaultdict
 from itertools import chain
@@ -229,3 +230,44 @@ class SegmentName:
   def data_dir(self) -> Optional[str]: return self._data_dir
 
   def __str__(self) -> str: return self._canonical_name
+
+
+@cache
+def get_max_seg_number_cached(sr: 'SegmentRange'):
+  try:
+    api = CommaApi(get_token())
+    return api.get("/v1/route/" + sr.route_name.replace("/", "|"))["segment_numbers"][-1]
+  except Exception as e:
+    raise Exception("unable to get max_segment_number. ensure you have access to this route or the route is public.") from e
+
+
+class SegmentRange:
+  def __init__(self, segment_range: str):
+    self.m = re.fullmatch(RE.SEGMENT_RANGE, segment_range)
+    assert self.m, f"Segment range is not valid {segment_range}"
+
+  def get_max_seg_number(self):
+    return get_max_seg_number_cached(self)
+
+  @property
+  def route_name(self):
+    return self.m.group("route_name")
+
+  @property
+  def dongle_id(self):
+    return self.m.group("dongle_id")
+
+  @property
+  def timestamp(self):
+    return self.m.group("timestamp")
+
+  @property
+  def _slice(self):
+    return self.m.group("slice")
+
+  @property
+  def selector(self):
+    return self.m.group("selector")
+
+  def __str__(self):
+    return f"{self.dongle_id}/{self.timestamp}" + (f"/{self._slice}" if self._slice else "") + (f"/{self.selector}" if self.selector else "")

tools/lib/sanitizer.py

@@ -0,0 +1,26 @@
+# Utilities for sanitizing routes of only essential data for testing car ports and doing validation.
+
+from openpilot.tools.lib.logreader import LogIterable, LogMessage
+
+
+def sanitize_vin(vin: str):
+  # (last 6 digits of vin are serial number https://en.wikipedia.org/wiki/Vehicle_identification_number)
+  VIN_SENSITIVE = 6
+  return vin[:-VIN_SENSITIVE] + "X" * VIN_SENSITIVE
+
+
+def sanitize_msg(msg: LogMessage) -> LogMessage:
+  if msg.which() == "carParams":
+    msg = msg.as_builder()
+    msg.carParams.carVin = sanitize_vin(msg.carParams.carVin)
+    msg = msg.as_reader()
+  return msg
+
+
+PRESERVE_SERVICES = ["can", "carParams", "pandaStates", "pandaStateDEPRECATED"]
+
+
+def sanitize(lr: LogIterable) -> LogIterable:
+  filtered = filter(lambda msg: msg.which() in PRESERVE_SERVICES, lr)
+  sanitized = map(sanitize_msg, filtered)
+  return sanitized

tools/lib/tests/test_caching.py

@@ -1,15 +1,58 @@
 #!/usr/bin/env python3
+from functools import wraps
+import http.server
 import os
+import threading
+import time
 import unittest
 
-from pathlib import Path
 from parameterized import parameterized
-from unittest import mock
 
-from openpilot.system.hardware.hw import Paths
 from openpilot.tools.lib.url_file import URLFile
 
 
+class CachingTestRequestHandler(http.server.BaseHTTPRequestHandler):
+  FILE_EXISTS = True
+
+  def do_GET(self):
+    if self.FILE_EXISTS:
+      self.send_response(200, b'1234')
+    else:
+      self.send_response(404)
+    self.end_headers()
+
+  def do_HEAD(self):
+    if self.FILE_EXISTS:
+      self.send_response(200)
+      self.send_header("Content-Length", "4")
+    else:
+      self.send_response(404)
+    self.end_headers()
+
+
+class CachingTestServer(threading.Thread):
+  def run(self):
+    self.server = http.server.HTTPServer(("127.0.0.1", 0), CachingTestRequestHandler)
+    self.port = self.server.server_port
+    self.server.serve_forever()
+
+  def stop(self):
+    self.server.server_close()
+    self.server.shutdown()
+
+def with_caching_server(func):
+  @wraps(func)
+  def wrapper(*args, **kwargs):
+    server = CachingTestServer()
+    server.start()
+    time.sleep(0.25) # wait for server to get it's port
+    try:
+      func(*args, **kwargs, port=server.port)
+    finally:
+      server.stop()
+  return wrapper
+
+
 class TestFileDownload(unittest.TestCase):
 
   def compare_loads(self, url, start=0, length=None):
@@ -66,32 +109,20 @@ class TestFileDownload(unittest.TestCase):
     self.compare_loads(large_file_url)
 
   @parameterized.expand([(True, ), (False, )])
-  def test_recover_from_missing_file(self, cache_enabled):
+  @with_caching_server
+  def test_recover_from_missing_file(self, cache_enabled, port):
     os.environ["FILEREADER_CACHE"] = "1" if cache_enabled else "0"
 
-    file_url = "http://localhost:5001/test.png"
+    file_url = f"http://localhost:{port}/test.png"
 
-    file_exists = False
+    CachingTestRequestHandler.FILE_EXISTS = False
+    length = URLFile(file_url).get_length()
+    self.assertEqual(length, -1)
 
-    def get_length_online_mock(self):
-      if file_exists:
-        return 4
-      return -1
+    CachingTestRequestHandler.FILE_EXISTS = True
+    length = URLFile(file_url).get_length()
+    self.assertEqual(length, 4)
 
-    patch_length = mock.patch.object(URLFile, "get_length_online", get_length_online_mock)
-    patch_length.start()
-    try:
-      length = URLFile(file_url).get_length()
-      self.assertEqual(length, -1)
-
-      file_exists = True
-      length = URLFile(file_url).get_length()
-      self.assertEqual(length, 4)
-    finally:
-      tempfile_length = Path(Paths.download_cache_root()) / "ba2119904385654cb0105a2da174875f8e7648db175f202ecae6d6428b0e838f_length"
-      if tempfile_length.exists():
-        tempfile_length.unlink()
-      patch_length.stop()
 
 
 if __name__ == "__main__":

tools/lib/tests/test_comma_car_segments.py

@@ -0,0 +1,41 @@
+
+
+import unittest
+
+import requests
+from openpilot.tools.lib.comma_car_segments import get_comma_car_segments_database, get_url
+from openpilot.tools.lib.logreader import LogReader
+from openpilot.tools.lib.route import SegmentRange
+
+
+class TestCommaCarSegments(unittest.TestCase):
+  def test_database(self):
+    database = get_comma_car_segments_database()
+
+    platforms = database.keys()
+
+    assert len(platforms) > 100
+
+  def test_download_segment(self):
+    database = get_comma_car_segments_database()
+
+    fp = "SUBARU FORESTER 2019"
+
+    segment = database[fp][0]
+
+    sr = SegmentRange(segment)
+
+    url = get_url(sr.route_name, sr._slice)
+
+    resp = requests.get(url)
+    self.assertEqual(resp.status_code, 200)
+
+    lr = LogReader(url)
+
+    CP = lr.first("carParams")
+
+    self.assertEqual(CP.carFingerprint, fp)
+
+
+if __name__ == "__main__":
+  unittest.main()

tools/lib/tests/test_logreader.py

@@ -0,0 +1,142 @@
+import shutil
+import tempfile
+import numpy as np
+import os
+import unittest
+import pytest
+import requests
+
+from parameterized import parameterized
+from unittest import mock
+
+from openpilot.tools.lib.logreader import LogIterable, LogReader, parse_indirect, parse_slice, ReadMode
+from openpilot.tools.lib.route import SegmentRange
+
+NUM_SEGS = 17 # number of segments in the test route
+ALL_SEGS = list(np.arange(NUM_SEGS))
+TEST_ROUTE = "344c5c15b34f2d8a/2024-01-03--09-37-12"
+QLOG_FILE = "https://commadataci.blob.core.windows.net/openpilotci/0375fdf7b1ce594d/2019-06-13--08-32-25/3/qlog.bz2"
+
+
+def noop(segment: LogIterable):
+  return segment
+
+
+class TestLogReader(unittest.TestCase):
+  @parameterized.expand([
+    (f"{TEST_ROUTE}", ALL_SEGS),
+    (f"{TEST_ROUTE.replace('/', '|')}", ALL_SEGS),
+    (f"{TEST_ROUTE}--0", [0]),
+    (f"{TEST_ROUTE}--5", [5]),
+    (f"{TEST_ROUTE}/0", [0]),
+    (f"{TEST_ROUTE}/5", [5]),
+    (f"{TEST_ROUTE}/0:10", ALL_SEGS[0:10]),
+    (f"{TEST_ROUTE}/0:0", []),
+    (f"{TEST_ROUTE}/4:6", ALL_SEGS[4:6]),
+    (f"{TEST_ROUTE}/0:-1", ALL_SEGS[0:-1]),
+    (f"{TEST_ROUTE}/:5", ALL_SEGS[:5]),
+    (f"{TEST_ROUTE}/2:", ALL_SEGS[2:]),
+    (f"{TEST_ROUTE}/2:-1", ALL_SEGS[2:-1]),
+    (f"{TEST_ROUTE}/-1", [ALL_SEGS[-1]]),
+    (f"{TEST_ROUTE}/-2", [ALL_SEGS[-2]]),
+    (f"{TEST_ROUTE}/-2:-1", ALL_SEGS[-2:-1]),
+    (f"{TEST_ROUTE}/-4:-2", ALL_SEGS[-4:-2]),
+    (f"{TEST_ROUTE}/:10:2", ALL_SEGS[:10:2]),
+    (f"{TEST_ROUTE}/5::2", ALL_SEGS[5::2]),
+    (f"https://useradmin.comma.ai/?onebox={TEST_ROUTE}", ALL_SEGS),
+    (f"https://useradmin.comma.ai/?onebox={TEST_ROUTE.replace('/', '|')}", ALL_SEGS),
+    (f"https://useradmin.comma.ai/?onebox={TEST_ROUTE.replace('/', '%7C')}", ALL_SEGS),
+    (f"https://cabana.comma.ai/?route={TEST_ROUTE}", ALL_SEGS),
+  ])
+  def test_indirect_parsing(self, identifier, expected):
+    parsed, _, _ = parse_indirect(identifier)
+    sr = SegmentRange(parsed)
+    segs = parse_slice(sr)
+    self.assertListEqual(list(segs), expected)
+
+  @parameterized.expand([
+    (f"{TEST_ROUTE}", f"{TEST_ROUTE}"),
+    (f"{TEST_ROUTE.replace('/', '|')}", f"{TEST_ROUTE}"),
+    (f"{TEST_ROUTE}--5", f"{TEST_ROUTE}/5"),
+    (f"{TEST_ROUTE}/0/q", f"{TEST_ROUTE}/0/q"),
+    (f"{TEST_ROUTE}/5:6/r", f"{TEST_ROUTE}/5:6/r"),
+    (f"{TEST_ROUTE}/5", f"{TEST_ROUTE}/5"),
+  ])
+  def test_canonical_name(self, identifier, expected):
+    sr = SegmentRange(identifier)
+    self.assertEqual(str(sr), expected)
+
+  def test_direct_parsing(self):
+    qlog = tempfile.NamedTemporaryFile(mode='wb', delete=False)
+
+    with requests.get(QLOG_FILE, stream=True) as r:
+      with qlog as f:
+        shutil.copyfileobj(r.raw, f)
+
+    for f in [QLOG_FILE, qlog.name]:
+      l = len(list(LogReader(f)))
+      self.assertGreater(l, 100)
+
+  @parameterized.expand([
+    (f"{TEST_ROUTE}///",),
+    (f"{TEST_ROUTE}---",),
+    (f"{TEST_ROUTE}/-4:--2",),
+    (f"{TEST_ROUTE}/-a",),
+    (f"{TEST_ROUTE}/j",),
+    (f"{TEST_ROUTE}/0:1:2:3",),
+    (f"{TEST_ROUTE}/:::3",),
+  ])
+  def test_bad_ranges(self, segment_range):
+    with self.assertRaises(AssertionError):
+      sr = SegmentRange(segment_range)
+      parse_slice(sr)
+
+  @pytest.mark.slow
+  def test_modes(self):
+    qlog_len = len(list(LogReader(f"{TEST_ROUTE}/0", ReadMode.QLOG)))
+    rlog_len = len(list(LogReader(f"{TEST_ROUTE}/0", ReadMode.RLOG)))
+
+    self.assertLess(qlog_len * 6, rlog_len)
+
+  @pytest.mark.slow
+  def test_modes_from_name(self):
+    qlog_len = len(list(LogReader(f"{TEST_ROUTE}/0/q")))
+    rlog_len = len(list(LogReader(f"{TEST_ROUTE}/0/r")))
+
+    self.assertLess(qlog_len * 6, rlog_len)
+
+  @pytest.mark.slow
+  def test_list(self):
+    qlog_len = len(list(LogReader(f"{TEST_ROUTE}/0/q")))
+    qlog_len_2 = len(list(LogReader([f"{TEST_ROUTE}/0/q", f"{TEST_ROUTE}/0/q"])))
+
+    self.assertEqual(qlog_len*2, qlog_len_2)
+
+  @pytest.mark.slow
+  @mock.patch("openpilot.tools.lib.logreader._LogFileReader")
+  def test_multiple_iterations(self, init_mock):
+    lr = LogReader(f"{TEST_ROUTE}/0/q")
+    qlog_len1 = len(list(lr))
+    qlog_len2 = len(list(lr))
+
+    # ensure we don't create multiple instances of _LogFileReader, which means downloading the files twice
+    self.assertEqual(init_mock.call_count, 1)
+
+    self.assertEqual(qlog_len1, qlog_len2)
+
+  @pytest.mark.slow
+  def test_helpers(self):
+    lr = LogReader(f"{TEST_ROUTE}/0/q")
+    self.assertEqual(lr.first("carParams").carFingerprint, "SUBARU OUTBACK 6TH GEN")
+    self.assertTrue(0 < len(list(lr.filter("carParams"))) < len(list(lr)))
+
+  @parameterized.expand([(True,), (False,)])
+  @pytest.mark.slow
+  def test_run_across_segments(self, cache_enabled):
+    os.environ["FILEREADER_CACHE"] = "1" if cache_enabled else "0"
+    lr = LogReader(f"{TEST_ROUTE}/0:4")
+    self.assertEqual(len(lr.run_across_segments(4, noop)), len(list(lr)))
+
+
+if __name__ == "__main__":
+  unittest.main()

tools/lib/url_file.py

@@ -1,16 +1,18 @@
+import logging
 import os
+import socket
 import time
-import threading
-import pycurl
 from hashlib import sha256
-from io import BytesIO
-from tenacity import retry, wait_random_exponential, stop_after_attempt
+from urllib3 import PoolManager, Retry
+from urllib3.util import Timeout
+
 from openpilot.common.file_helpers import atomic_write_in_dir
 from openpilot.system.hardware.hw import Paths
 #  Cache chunk size
 K = 1000
 CHUNK_SIZE = 1000 * K
 
+logging.getLogger("urllib3").setLevel(logging.WARNING)
 
 def hash_256(link):
   hsh = str(sha256((link.split("?")[0]).encode('utf-8')).hexdigest())
@@ -21,11 +23,23 @@ class URLFileException(Exception):
   pass
 
 
-class URLFile:
-  _tlocal = threading.local()
+def new_pool_manager() -> PoolManager:
+  socket_options = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),]
+  retries = Retry(total=5, backoff_factor=0.5, status_forcelist=[409, 429, 503, 504])
+  return PoolManager(num_pools=10, maxsize=100, socket_options=socket_options, retries=retries)
 
-  def __init__(self, url, debug=False, cache=None):
+
+def set_pool_manager():
+  URLFile._pool_manager = new_pool_manager()
+os.register_at_fork(after_in_child=set_pool_manager)
+
+
+class URLFile:
+  _pool_manager = new_pool_manager()
+
+  def __init__(self, url, timeout=10, debug=False, cache=None):
     self._url = url
+    self._timeout = Timeout(connect=timeout, read=timeout)
     self._pos = 0
     self._length = None
     self._local_file = None
@@ -35,10 +49,6 @@ class URLFile:
     if cache is not None:
       self._force_download = not cache
 
-    try:
-      self._curl = self._tlocal.curl
-    except AttributeError:
-      self._curl = self._tlocal.curl = pycurl.Curl()
     if not self._force_download:
       os.makedirs(Paths.download_cache_root(), exist_ok=True)
 
@@ -51,19 +61,15 @@ class URLFile:
       self._local_file.close()
       self._local_file = None
 
-  @retry(wait=wait_random_exponential(multiplier=1, max=5), stop=stop_after_attempt(3), reraise=True)
+  def _request(self, method, url, headers=None):
+    return URLFile._pool_manager.request(method, url, timeout=self._timeout, headers=headers)
+
   def get_length_online(self):
-    c = self._curl
-    c.reset()
-    c.setopt(pycurl.NOSIGNAL, 1)
-    c.setopt(pycurl.TIMEOUT_MS, 500000)
-    c.setopt(pycurl.FOLLOWLOCATION, True)
-    c.setopt(pycurl.URL, self._url)
-    c.setopt(c.NOBODY, 1)
-    c.perform()
-    length = int(c.getinfo(c.CONTENT_LENGTH_DOWNLOAD))
-    c.reset()
-    return length
+    response = self._request('HEAD', self._url)
+    if not (200 <= response.status <= 299):
+      return -1
+    length = response.headers.get('content-length', 0)
+    return int(length)
 
   def get_length(self):
     if self._length is not None:
@@ -114,10 +120,9 @@ class URLFile:
         self._pos = file_end
         return response
 
-  @retry(wait=wait_random_exponential(multiplier=1, max=5), stop=stop_after_attempt(3), reraise=True)
   def read_aux(self, ll=None):
     download_range = False
-    headers = ["Connection: keep-alive"]
+    headers = {}
     if self._pos != 0 or ll is not None:
       if ll is None:
         end = self.get_length() - 1
@@ -125,50 +130,28 @@ class URLFile:
         end = min(self._pos + ll, self.get_length()) - 1
       if self._pos >= end:
         return b""
-      headers.append(f"Range: bytes={self._pos}-{end}")
+      headers['Range'] = f"bytes={self._pos}-{end}"
       download_range = True
 
-    dats = BytesIO()
-    c = self._curl
-    c.setopt(pycurl.URL, self._url)
-    c.setopt(pycurl.WRITEDATA, dats)
-    c.setopt(pycurl.NOSIGNAL, 1)
-    c.setopt(pycurl.TIMEOUT_MS, 500000)
-    c.setopt(pycurl.HTTPHEADER, headers)
-    c.setopt(pycurl.FOLLOWLOCATION, True)
-
     if self._debug:
-      print("downloading", self._url)
-
-      def header(x):
-        if b'MISS' in x:
-          print(x.strip())
-
-      c.setopt(pycurl.HEADERFUNCTION, header)
-
-      def test(debug_type, debug_msg):
-       print("  debug(%d): %s" % (debug_type, debug_msg.strip()))
-
-      c.setopt(pycurl.VERBOSE, 1)
-      c.setopt(pycurl.DEBUGFUNCTION, test)
       t1 = time.time()
 
-    c.perform()
+    response = self._request('GET', self._url, headers=headers)
+    ret = response.data
 
     if self._debug:
       t2 = time.time()
       if t2 - t1 > 0.1:
-        print(f"get {self._url} {headers!r} {t2 - t1:.f} slow")
+        print(f"get {self._url} {headers!r} {t2 - t1:.3f} slow")
 
-    response_code = c.getinfo(pycurl.RESPONSE_CODE)
+    response_code = response.status
     if response_code == 416:  # Requested Range Not Satisfiable
-      raise URLFileException(f"Error, range out of bounds {response_code} {headers} ({self._url}): {repr(dats.getvalue())[:500]}")
+      raise URLFileException(f"Error, range out of bounds {response_code} {headers} ({self._url}): {repr(ret)[:500]}")
     if download_range and response_code != 206:  # Partial Content
-      raise URLFileException(f"Error, requested range but got unexpected response {response_code} {headers} ({self._url}): {repr(dats.getvalue())[:500]}")
+      raise URLFileException(f"Error, requested range but got unexpected response {response_code} {headers} ({self._url}): {repr(ret)[:500]}")
     if (not download_range) and response_code != 200:  # OK
-      raise URLFileException(f"Error {response_code} {headers} ({self._url}): {repr(dats.getvalue())[:500]}")
+      raise URLFileException(f"Error {response_code} {headers} ({self._url}): {repr(ret)[:500]}")
 
-    ret = dats.getvalue()
     self._pos += len(ret)
     return ret