mawei/StarPilot
1
0
Fork 0
mirror of https://github.com/firestar5683/StarPilot.git synced 2026-08-05 16:26:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

LogReader: formatting and remove unnecessary type alias

Browse Source 
old-commit-hash: 34305bec091c1088249fe6fe8df8de64a7ce1f86
This commit is contained in:
Shane Smiskol
2024-09-03 17:58:50 -07:00
parent 72d7ada7e4
commit 730924e7f2
1 changed files with 20 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/lib/logreader.py
+20 -21
View File
@@ -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,
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 @@ 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 \
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