mirror of
https://github.com/infiniteCable2/openpilot.git
synced 2026-08-06 08:46:08 +08:00
@@ -102,7 +102,8 @@ class ReadMode(enum.StrEnum):
|
||||
|
||||
|
||||
LogPath = str | None
|
||||
Source = Callable[[SegmentRange, tuple[str, ...]], list[LogPath]]
|
||||
LogFileName = tuple[str, ...]
|
||||
Source = Callable[[SegmentRange, LogFileName], list[LogPath]]
|
||||
|
||||
InternalUnavailableException = Exception("Internal source not available")
|
||||
|
||||
@@ -111,7 +112,7 @@ class LogsUnavailable(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def comma_api_source(sr: SegmentRange, fns: tuple[str, ...]) -> list[LogPath]:
|
||||
def comma_api_source(sr: SegmentRange, fns: LogFileName) -> list[LogPath]:
|
||||
route = Route(sr.route_name)
|
||||
|
||||
# comma api will have already checked if the file exists
|
||||
@@ -121,7 +122,7 @@ def comma_api_source(sr: SegmentRange, fns: tuple[str, ...]) -> list[LogPath]:
|
||||
return [route.qlog_paths()[seg] for seg in sr.seg_idxs]
|
||||
|
||||
|
||||
def internal_source(sr: SegmentRange, fns: tuple[str, ...], endpoint_url: str = DATA_ENDPOINT) -> list[LogPath]:
|
||||
def internal_source(sr: SegmentRange, fns: LogFileName, endpoint_url: str = DATA_ENDPOINT) -> list[LogPath]:
|
||||
if not internal_source_available(endpoint_url):
|
||||
raise InternalUnavailableException
|
||||
|
||||
@@ -131,11 +132,11 @@ def internal_source(sr: SegmentRange, fns: tuple[str, ...], endpoint_url: str =
|
||||
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: tuple[str, ...]) -> list[LogPath]:
|
||||
def openpilotci_source(sr: SegmentRange, fns: LogFileName) -> 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: tuple[str, ...]) -> list[LogPath]:
|
||||
def comma_car_segments_source(sr: SegmentRange, fns: LogFileName) -> list[LogPath]:
|
||||
return eval_source([get_comma_segments_url(sr.route_name, seg) for seg in sr.seg_idxs])
|
||||
|
||||
|
||||
|
||||
+9
-9
@@ -1,7 +1,6 @@
|
||||
import os
|
||||
import re
|
||||
import requests
|
||||
from typing import TypeAlias
|
||||
from functools import cache
|
||||
from urllib.parse import urlparse
|
||||
from collections import defaultdict
|
||||
@@ -11,15 +10,16 @@ from openpilot.tools.lib.auth_config import get_token
|
||||
from openpilot.tools.lib.api import APIError, CommaApi
|
||||
from openpilot.tools.lib.helpers import RE
|
||||
|
||||
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')
|
||||
RLOG = ("rlog.zst", "rlog.bz2")
|
||||
QLOG = ("qlog.zst", "qlog.bz2")
|
||||
QCAMERA = ('qcamera.ts',)
|
||||
FCAMERA = ('fcamera.hevc',)
|
||||
ECAMERA = ('ecamera.hevc',)
|
||||
DCAMERA = ('dcamera.hevc',)
|
||||
BOOTLOG = ('bootlog.zst', 'bootlog.bz2')
|
||||
|
||||
|
||||
class Route:
|
||||
def __init__(self, name, data_dir=None):
|
||||
|
||||
Reference in New Issue
Block a user