LogReader: add data endpoint option (#35751)

* add data endpoint option

* clean up

tools/lib/filereader.py

@@ -8,14 +8,14 @@ from openpilot.tools.lib.url_file import URLFile
 DATA_ENDPOINT = os.getenv("DATA_ENDPOINT", "http://data-raw.comma.internal/")
 
 
-def internal_source_available(url=DATA_ENDPOINT):
+def internal_source_available(url: str) -> bool:
   if os.path.isdir(url):
     return True
 
   try:
     hostname = urlparse(url).hostname
     port = urlparse(url).port or 80
-    with socket.socket(socket.AF_INET,socket.SOCK_STREAM) as s:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
       s.settimeout(0.5)
       s.connect((hostname, port))
     return True

tools/lib/logreader.py

@@ -19,7 +19,7 @@ from cereal import log as capnp_log
 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 FileReader, file_exists, internal_source_available
+from openpilot.tools.lib.filereader import DATA_ENDPOINT, FileReader, file_exists, internal_source_available
 from openpilot.tools.lib.route import Route, SegmentRange
 from openpilot.tools.lib.log_time_series import msgs_to_time_series
 
@@ -157,12 +157,13 @@ def comma_api_source(sr: SegmentRange, mode: ReadMode) -> list[LogPath]:
   return apply_strategy(mode, rlog_paths, qlog_paths, valid_file=valid_file)
 
 
-def internal_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2") -> list[LogPath]:
-  if not internal_source_available():
+def internal_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2",
+                    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"cd:/{sr.dongle_id}/{sr.log_id}/{seg}/{file}.{file_ext}"
+    return f"{endpoint_url.rstrip('/')}/{sr.dongle_id}/{sr.log_id}/{seg}/{file}.{file_ext}"
 
   # TODO: list instead of using static URLs to support routes with multiple file extensions
   rlog_paths = [get_internal_url(sr, seg, "rlog") for seg in sr.seg_idxs]
@@ -171,8 +172,9 @@ 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") -> list[LogPath]:
-  return internal_source(sr, mode, file_ext)
+def internal_source_zst(sr: SegmentRange, mode: ReadMode, file_ext: str = "zst",
+                        endpoint_url: str = DATA_ENDPOINT) -> list[LogPath]:
+  return internal_source(sr, mode, file_ext, endpoint_url)
 
 
 def openpilotci_source(sr: SegmentRange, mode: ReadMode, file_ext: str = "bz2") -> list[LogPath]: