diff --git a/tools/lib/logreader.py b/tools/lib/logreader.py index c12b14ebd4286c..1f5309c1f33318 100755 --- a/tools/lib/logreader.py +++ b/tools/lib/logreader.py @@ -93,9 +93,8 @@ class ReadMode(enum.StrEnum): LogPath = str | None -LogPaths = list[LogPath] ValidFileCallable = Callable[[LogPath], bool] -Source = Callable[[SegmentRange, ReadMode], LogPaths] +Source = Callable[[SegmentRange, ReadMode], list[LogPath]] InternalUnavailableException = Exception("Internal source not available") @@ -105,7 +104,7 @@ def default_valid_file(fn: LogPath) -> bool: return fn is not None and file_exists(fn) -def auto_strategy(rlog_paths: LogPaths, qlog_paths: LogPaths, interactive: bool, valid_file: ValidFileCallable) -> LogPaths: +def auto_strategy(rlog_paths: list[LogPath], qlog_paths: list[LogPath], interactive: bool, valid_file: ValidFileCallable) -> list[LogPath]: # auto select logs based on availability missing_rlogs = [rlog is None or not valid_file(rlog) for rlog in rlog_paths].count(True) if missing_rlogs != 0: @@ -120,7 +119,7 @@ def auto_strategy(rlog_paths: LogPaths, qlog_paths: LogPaths, interactive: bool, return rlog_paths -def apply_strategy(mode: ReadMode, rlog_paths: LogPaths, qlog_paths: LogPaths, valid_file: ValidFileCallable = default_valid_file) -> LogPaths: +def apply_strategy(mode: ReadMode, rlog_paths: list[LogPath], qlog_paths: list[LogPath], valid_file: ValidFileCallable = default_valid_file) -> list[LogPath]: if mode == ReadMode.RLOG: return rlog_paths elif mode == ReadMode.QLOG: @@ -132,7 +131,7 @@ def apply_strategy(mode: ReadMode, rlog_paths: LogPaths, qlog_paths: LogPaths, v raise Exception(f"invalid mode: {mode}") -def comma_api_source(sr: SegmentRange, mode: ReadMode) -> LogPaths: +def comma_api_source(sr: SegmentRange, mode: ReadMode) -> list[LogPath]: route = Route(sr.route_name) rlog_paths = [route.log_paths()[seg] for seg in sr.seg_idxs] @@ -145,7 +144,7 @@ def valid_file(fn): return apply_strategy(mode, rlog_paths, qlog_paths, valid_file=valid_file) -def internal_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2") -> LogPaths: +def internal_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2") -> list[LogPath]: if not internal_source_available(): raise InternalUnavailableException @@ -159,32 +158,32 @@ def get_internal_url(sr: SegmentRange, seg, file): return apply_strategy(mode, rlog_paths, qlog_paths) -def internal_source_zst(sr: SegmentRange, mode: ReadMode, file_ext: str = "zst") -> LogPaths: +def internal_source_zst(sr: SegmentRange, mode: ReadMode, file_ext: str = "zst") -> list[LogPath]: return internal_source(sr, mode, file_ext) -def openpilotci_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2") -> LogPaths: +def openpilotci_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2") -> list[LogPath]: rlog_paths = [get_url(sr.route_name, seg, f"rlog.{file_ext}") for seg in sr.seg_idxs] qlog_paths = [get_url(sr.route_name, seg, f"qlog.{file_ext}") for seg in sr.seg_idxs] return apply_strategy(mode, rlog_paths, qlog_paths) -def openpilotci_source_zst(sr: SegmentRange, mode: ReadMode) -> LogPaths: +def openpilotci_source_zst(sr: SegmentRange, mode: ReadMode) -> list[LogPath]: return openpilotci_source(sr, mode, "zst") -def comma_car_segments_source(sr: SegmentRange, mode=ReadMode.RLOG) -> LogPaths: +def comma_car_segments_source(sr: SegmentRange, mode=ReadMode.RLOG) -> list[LogPath]: return [get_comma_segments_url(sr.route_name, seg) for seg in sr.seg_idxs] -def testing_closet_source(sr: SegmentRange, mode=ReadMode.RLOG) -> LogPaths: +def testing_closet_source(sr: SegmentRange, mode=ReadMode.RLOG) -> list[LogPath]: if not internal_source_available('http://testing.comma.life'): raise InternalUnavailableException return [f"http://testing.comma.life/download/{sr.route_name.replace('|', '/')}/{seg}/rlog" for seg in sr.seg_idxs] -def direct_source(file_or_url: str) -> LogPaths: +def direct_source(file_or_url: str) -> list[LogPath]: return [file_or_url] @@ -194,31 +193,31 @@ def get_invalid_files(files): yield f -def check_source(source: Source, *args) -> LogPaths: +def check_source(source: Source, *args) -> list[LogPath]: files = source(*args) assert len(files) > 0, "No files on source" assert next(get_invalid_files(files), False) is False, "Some files are invalid" return files -def auto_source(sr: SegmentRange, mode=ReadMode.RLOG) -> LogPaths: +def auto_source(sr: SegmentRange, mode=ReadMode.RLOG) -> list[LogPath]: if mode == ReadMode.SANITIZED: return comma_car_segments_source(sr, mode) - SOURCES: list[Source] = [internal_source, internal_source_zst, openpilotci_source, openpilotci_source_zst, - comma_api_source, comma_car_segments_source, testing_closet_source,] + sources: list[Source] = [internal_source, internal_source_zst, openpilotci_source, openpilotci_source_zst, + comma_api_source, comma_car_segments_source, testing_closet_source] exceptions = {} # for automatic fallback modes, auto_source needs to first check if rlogs exist for any source if mode in [ReadMode.AUTO, ReadMode.AUTO_INTERACTIVE]: - for source in SOURCES: + for source in sources: try: return check_source(source, sr, ReadMode.RLOG) except Exception: pass # Automatically determine viable source - for source in SOURCES: + for source in sources: try: return check_source(source, sr, mode) except Exception as e: @@ -276,12 +275,12 @@ def _parse_identifiers(self, identifier: str | list[str]): identifiers = source(sr, mode) invalid_count = len(list(get_invalid_files(identifiers))) - assert invalid_count == 0, f"{invalid_count}/{len(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." + assert invalid_count == 0, (f"{invalid_count}/{len(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.") return identifiers def __init__(self, identifier: str | list[str], default_mode: ReadMode = ReadMode.RLOG, - default_source=auto_source, sort_by_time=False, only_union_types=False): + default_source: Source = auto_source, sort_by_time=False, only_union_types=False): self.default_mode = default_mode self.default_source = default_source self.identifier = identifier