diff --git a/tools/lib/logreader.py b/tools/lib/logreader.py index 1075bd2f08..fa6e6357eb 100755 --- a/tools/lib/logreader.py +++ b/tools/lib/logreader.py @@ -21,8 +21,7 @@ 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 DATA_ENDPOINT, FileReader, file_exists, internal_source_available -from openpilot.tools.lib.route import QCAMERA_FILENAMES, CAMERA_FILENAMES, DCAMERA_FILENAMES, \ - ECAMERA_FILENAMES, BOOTLOG_FILENAMES, Route, SegmentRange +from openpilot.tools.lib.route import Route, SegmentRange, FileName from openpilot.tools.lib.log_time_series import msgs_to_time_series LogMessage = type[capnp._DynamicStructReader] @@ -102,19 +101,8 @@ class ReadMode(enum.StrEnum): AUTO_INTERACTIVE = "i" # default to rlogs, fallback to qlogs with a prompt from the user -class FileName(enum.Enum): - #TODO use the ones from route.py - RLOG = ("rlog.zst", "rlog.bz2") - QLOG = ("qlog.zst", "qlog.bz2") - QCAMERA = QCAMERA_FILENAMES - FCAMERA = CAMERA_FILENAMES - ECAMERA = ECAMERA_FILENAMES - DCAMERA = DCAMERA_FILENAMES - BOOTLOG = BOOTLOG_FILENAMES - - LogPath = str | None -Source = Callable[[SegmentRange, FileName], list[LogPath]] +Source = Callable[[SegmentRange, tuple[str, ...]], list[LogPath]] InternalUnavailableException = Exception("Internal source not available") @@ -123,7 +111,7 @@ class LogsUnavailable(Exception): pass -def comma_api_source(sr: SegmentRange, fns: FileName) -> list[LogPath]: +def comma_api_source(sr: SegmentRange, fns: tuple[str, ...]) -> list[LogPath]: route = Route(sr.route_name) # comma api will have already checked if the file exists @@ -133,21 +121,21 @@ def comma_api_source(sr: SegmentRange, fns: FileName) -> list[LogPath]: return [route.qlog_paths()[seg] for seg in sr.seg_idxs] -def internal_source(sr: SegmentRange, fns: FileName, endpoint_url: str = DATA_ENDPOINT) -> list[LogPath]: +def internal_source(sr: SegmentRange, fns: tuple[str, ...], endpoint_url: str = DATA_ENDPOINT) -> list[LogPath]: if not internal_source_available(endpoint_url): raise InternalUnavailableException def get_internal_url(sr: SegmentRange, seg, file): return f"{endpoint_url.rstrip('/')}/{sr.dongle_id}/{sr.log_id}/{seg}/{file}" - return eval_source([[get_internal_url(sr, seg, fn) for fn in fns.value] for seg in sr.seg_idxs]) + return eval_source([[get_internal_url(sr, seg, fn) for fn in fns] for seg in sr.seg_idxs]) -def openpilotci_source(sr: SegmentRange, fns: FileName) -> list[LogPath]: - return eval_source([[get_url(sr.route_name, seg, fn) for fn in fns.value] for seg in sr.seg_idxs]) +def openpilotci_source(sr: SegmentRange, fns: tuple[str, ...]) -> list[LogPath]: + return eval_source([[get_url(sr.route_name, seg, fn) for fn in fns] for seg in sr.seg_idxs]) -def comma_car_segments_source(sr: SegmentRange, fns: FileName) -> list[LogPath]: +def comma_car_segments_source(sr: SegmentRange, fns: tuple[str, ...]) -> list[LogPath]: return eval_source([get_comma_segments_url(sr.route_name, seg) for seg in sr.seg_idxs]) diff --git a/tools/lib/route.py b/tools/lib/route.py index 55ea956c5f..dc8bb60e8f 100644 --- a/tools/lib/route.py +++ b/tools/lib/route.py @@ -1,6 +1,7 @@ import os import re import requests +from typing import TypeAlias from functools import cache from urllib.parse import urlparse from collections import defaultdict @@ -10,14 +11,15 @@ from openpilot.tools.lib.auth_config import get_token from openpilot.tools.lib.api import APIError, CommaApi from openpilot.tools.lib.helpers import RE -QLOG_FILENAMES = ('qlog.bz2', 'qlog.zst', 'qlog') -QCAMERA_FILENAMES = ('qcamera.ts',) -LOG_FILENAMES = ('rlog.bz2', 'raw_log.bz2', 'rlog.zst', 'rlog') -CAMERA_FILENAMES = ('fcamera.hevc', 'video.hevc') -DCAMERA_FILENAMES = ('dcamera.hevc',) -ECAMERA_FILENAMES = ('ecamera.hevc',) -BOOTLOG_FILENAMES = ('bootlog.zst', 'bootlog.bz2', 'bootlog') - +FileNameTuple: TypeAlias = tuple[str, ...] +class FileName: + RLOG: FileNameTuple = ("rlog.zst", "rlog.bz2") + QLOG: FileNameTuple = ("qlog.zst", "qlog.bz2") + QCAMERA: FileNameTuple = ('qcamera.ts',) + FCAMERA: FileNameTuple = ('fcamera.hevc',) + ECAMERA: FileNameTuple = ('ecamera.hevc',) + DCAMERA: FileNameTuple = ('dcamera.hevc',) + BOOTLOG: FileNameTuple = ('bootlog.zst', 'bootlog.bz2') class Route: def __init__(self, name, data_dir=None): @@ -82,23 +84,23 @@ class Route: if segments.get(segment_name): segments[segment_name] = Segment( segment_name, - url if fn in LOG_FILENAMES else segments[segment_name].log_path, - url if fn in QLOG_FILENAMES else segments[segment_name].qlog_path, - url if fn in CAMERA_FILENAMES else segments[segment_name].camera_path, - url if fn in DCAMERA_FILENAMES else segments[segment_name].dcamera_path, - url if fn in ECAMERA_FILENAMES else segments[segment_name].ecamera_path, - url if fn in QCAMERA_FILENAMES else segments[segment_name].qcamera_path, + url if fn in FileName.RLOG else segments[segment_name].log_path, + url if fn in FileName.QLOG else segments[segment_name].qlog_path, + url if fn in FileName.FCAMERA else segments[segment_name].camera_path, + url if fn in FileName.DCAMERA else segments[segment_name].dcamera_path, + url if fn in FileName.ECAMERA else segments[segment_name].ecamera_path, + url if fn in FileName.QCAMERA else segments[segment_name].qcamera_path, self.metadata['url'], ) else: segments[segment_name] = Segment( segment_name, - url if fn in LOG_FILENAMES else None, - url if fn in QLOG_FILENAMES else None, - url if fn in CAMERA_FILENAMES else None, - url if fn in DCAMERA_FILENAMES else None, - url if fn in ECAMERA_FILENAMES else None, - url if fn in QCAMERA_FILENAMES else None, + url if fn in FileName.RLOG else None, + url if fn in FileName.QLOG else None, + url if fn in FileName.FCAMERA else None, + url if fn in FileName.DCAMERA else None, + url if fn in FileName.ECAMERA else None, + url if fn in FileName.QCAMERA else None, self.metadata['url'], ) @@ -136,32 +138,32 @@ class Route: for segment, files in segment_files.items(): try: - log_path = next(path for path, filename in files if filename in LOG_FILENAMES) + log_path = next(path for path, filename in files if filename in FileName.RLOG) except StopIteration: log_path = None try: - qlog_path = next(path for path, filename in files if filename in QLOG_FILENAMES) + qlog_path = next(path for path, filename in files if filename in FileName.QLOG) except StopIteration: qlog_path = None try: - camera_path = next(path for path, filename in files if filename in CAMERA_FILENAMES) + camera_path = next(path for path, filename in files if filename in FileName.FCAMERA) except StopIteration: camera_path = None try: - dcamera_path = next(path for path, filename in files if filename in DCAMERA_FILENAMES) + dcamera_path = next(path for path, filename in files if filename in FileName.DCAMERA) except StopIteration: dcamera_path = None try: - ecamera_path = next(path for path, filename in files if filename in ECAMERA_FILENAMES) + ecamera_path = next(path for path, filename in files if filename in FileName.ECAMERA) except StopIteration: ecamera_path = None try: - qcamera_path = next(path for path, filename in files if filename in QCAMERA_FILENAMES) + qcamera_path = next(path for path, filename in files if filename in FileName.QCAMERA) except StopIteration: qcamera_path = None