From 771ef009e99560b96555a7f66f5408fa4a3b78b8 Mon Sep 17 00:00:00 2001
From: Gregor Kikelj <gregor1234567890@gmail.com>
Date: Tue, 29 Sep 2020 18:52:30 +0200
Subject: [PATCH] wip

---
 selfdrive/loggerd/tests/test_loggerd.py |  9 ++-
 selfdrive/test/test_valgrind_replay.py  | 89 +++++++++++++++++++++++++
 2 files changed, 93 insertions(+), 5 deletions(-)
 create mode 100644 selfdrive/test/test_valgrind_replay.py

diff --git a/selfdrive/loggerd/tests/test_loggerd.py b/selfdrive/loggerd/tests/test_loggerd.py
index 714350425d..defe66fd8f 100755
--- a/selfdrive/loggerd/tests/test_loggerd.py
+++ b/selfdrive/loggerd/tests/test_loggerd.py
@@ -29,7 +29,7 @@ elif TICI:
 else:
   CAMERAS = {}
 
-ALL_CAMERA_COMBINATIONS = [(cameras,) for cameras in [CAMERAS, {k: CAMERAS[k] for k in CAMERAS if k != 'dcamera'}]]
+ALL_CAMERA_COMBINATIONS = [(cameras,) for cameras in [CAMERAS, {k:CAMERAS[k] for k in CAMERAS if k!='dcamera'}]]
 
 FRAME_TOLERANCE = 2
 FILE_SIZE_TOLERANCE = 0.25
@@ -40,8 +40,7 @@ class TestLoggerd(unittest.TestCase):
   @classmethod
   def setUpClass(cls):
     if not (EON or TICI):
-      pass
-      # raise unittest.SkipTest
+      raise unittest.SkipTest
 
   def setUp(self):
     self._clear_logs()
@@ -80,7 +79,7 @@ class TestLoggerd(unittest.TestCase):
     for i in trange(num_segments):
       # check each camera file size
       for camera, size in cameras.items():
-        ext = "ts" if camera == 'qcamera' else "hevc"
+        ext = "ts" if camera=='qcamera' else "hevc"
         file_path = f"{route_prefix_path}--{i}/{camera}.{ext}"
 
         # check file size
@@ -95,7 +94,7 @@ class TestLoggerd(unittest.TestCase):
         # check frame count
         cmd = f"ffprobe -v error -count_frames -select_streams v:0 -show_entries stream=nb_read_frames \
                -of default=nokey=1:noprint_wrappers=1 {file_path}"
-        expected_frames = self.segment_length * CAMERA_FPS // 2 if (EON and camera == 'dcamera') else self.segment_length * CAMERA_FPS
+        expected_frames = self.segment_length * CAMERA_FPS // 2 if (EON and camera=='dcamera') else self.segment_length * CAMERA_FPS
         frame_count = int(subprocess.check_output(cmd, shell=True, encoding='utf8').strip())
         self.assertTrue(abs(expected_frames - frame_count) <= FRAME_TOLERANCE,
                         f"{camera} failed frame count check: expected {expected_frames}, got {frame_count}")
diff --git a/selfdrive/test/test_valgrind_replay.py b/selfdrive/test/test_valgrind_replay.py
new file mode 100644
index 0000000000..a508bf1993
--- /dev/null
+++ b/selfdrive/test/test_valgrind_replay.py
@@ -0,0 +1,89 @@
+import os
+import threading
+import time
+
+if "CI" in os.environ:
+  def tqdm(x):
+    return x
+else:
+  from tqdm import tqdm   # type: ignore
+
+import cereal.messaging as messaging
+from collections import namedtuple
+from tools.lib.logreader import LogReader
+
+ProcessConfig = namedtuple('ProcessConfig', ['proc_name', 'pub_sub', 'ignore', 'command', 'path'])
+
+class SimplePubMaster():
+  def __init__(self, services):  # pylint: disable=super-init-not-called
+    self.sock = {}
+    for s in services:
+      self.sock[s] = messaging.pub_sock(s)
+
+  def send(self, s, dat):
+    # print(dat)
+    self.sock[s].send(dat.to_bytes())
+
+
+CONFIGS = [
+  ProcessConfig(
+    proc_name="ubloxd",
+    pub_sub={
+      "ubloxRaw": ["ubloxGnss", "gpsLocationExternal"],
+    },
+    ignore=[],
+    command="./ubloxd & sleep 20; kill $!",
+    path="../locationd",
+  ),
+]
+
+def valgrindlauncher(arg, cwd):
+  os.chdir(cwd)
+
+  # Run valgrind on a process
+  command = "valgrind --leak-check=full " + arg
+  print(command)
+  output = os.popen(command)
+  while True:
+    s = output.read()
+    if s == "":
+      break
+    print(s)
+
+def replay_process(cfg, lr):
+  pub_sockets = [s for s in cfg.pub_sub.keys() if s != 'can']  # We dump data from logs here
+
+  pm = SimplePubMaster(pub_sockets)
+  print("Sorting logs")
+  all_msgs = sorted(lr, key=lambda msg: msg.logMonoTime)
+  pub_msgs = [msg for msg in all_msgs if msg.which() in list(cfg.pub_sub.keys())]
+  print(len(pub_msgs))
+  thread = threading.Thread(target=valgrindlauncher, args=(cfg.command, cfg.path))
+  thread.daemon = True
+  thread.start()
+  time.sleep(5)  # We give the process time to start
+  for msg in tqdm(pub_msgs):
+    pm.send(msg.which(), msg.as_builder())
+
+BASE_URL = "https://commadataci.blob.core.windows.net/openpilotci/"
+
+def get_segment(segment_name, original=True):
+  route_name, segment_num = segment_name.rsplit("--", 1)
+  if original:
+    rlog_url = BASE_URL + "%s/%s/rlog.bz2" % (route_name.replace("|", "/"), segment_num)
+  else:
+    process_replay_dir = os.path.dirname(os.path.abspath(__file__))
+    model_ref_commit = open(os.path.join(process_replay_dir, "model_ref_commit")).read().strip()
+    rlog_url = BASE_URL + "%s/%s/rlog_%s.bz2" % (route_name.replace("|", "/"), segment_num, model_ref_commit)
+
+  return rlog_url
+
+if __name__ == "__main__":
+  cfg = CONFIGS[0]
+
+  URL = get_segment("0375fdf7b1ce594d|2019-06-13--08-32-25--3")
+  print(URL)
+  lr = LogReader(URL)
+  print(str(cfg))
+  replay_process(cfg, lr)
+  time.sleep(30)