LogReader: formatting and remove unnecessary type alias

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 comma_api_source(sr: SegmentRange, mode: ReadMode) -> LogPaths:
   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 internal_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2") ->
   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,
+  sources: list[Source] = [internal_source, internal_source_zst, openpilotci_source, openpilotci_source_zst,
-                           comma_api_source, comma_car_segments_source, testing_closet_source,]
+                           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 @@ class LogReader:
     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 \
+    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."
+                                "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