segment regen (#21148)

* start seg regen

* send vipc

* fix up some paths

* set fp

* fix no entries

* engages

* seperate camera procs

* send real frames

* regen test routes

* nice argparser

* fix valgrind test

* move that

* fix that

selfdrive/common/params.cc

@@ -159,6 +159,7 @@ std::unordered_map<std::string, uint32_t> keys = {
     {"CarVin", CLEAR_ON_MANAGER_START | CLEAR_ON_PANDA_DISCONNECT | CLEAR_ON_IGNITION_ON},
     {"CommunityFeaturesToggle", PERSISTENT},
     {"ControlsReady", CLEAR_ON_MANAGER_START | CLEAR_ON_PANDA_DISCONNECT | CLEAR_ON_IGNITION_ON},
+    {"CurrentRoute", CLEAR_ON_MANAGER_START | CLEAR_ON_IGNITION_ON},
     {"DisableRadar", PERSISTENT}, // WARNING: THIS DISABLES AEB
     {"EndToEndToggle", PERSISTENT},
     {"CompletedTrainingVersion", PERSISTENT},

selfdrive/debug/check_freq.py

@@ -20,6 +20,7 @@ if __name__ == "__main__":
   sockets = {}
 
   rcv_times = defaultdict(lambda: deque(maxlen=100))
+  valids = defaultdict(lambda: deque(maxlen=100))
 
   t = sec_since_boot()
   for name in socket_names:
@@ -34,12 +35,13 @@ if __name__ == "__main__":
 
       t = sec_since_boot()
       rcv_times[name].append(msg.logMonoTime / 1e9)
+      valids[name].append(msg.valid)
 
     if t - prev_print > 1:
       print()
       for name in socket_names:
         dts = np.diff(rcv_times[name])
         mean = np.mean(dts)
-        print("%s: Freq %.2f Hz, Min %.2f%%, Max %.2f%%" % (name, 1.0 / mean, np.min(dts) / mean * 100, np.max(dts) / mean * 100))
+        print("%s: Freq %.2f Hz, Min %.2f%%, Max %.2f%%, valid " % (name, 1.0 / mean, np.min(dts) / mean * 100, np.max(dts) / mean * 100), all(valids[name]))
 
       prev_print = t

selfdrive/loggerd/logger.h

@@ -15,9 +15,12 @@
 #include "selfdrive/common/swaglog.h"
 #include "selfdrive/hardware/hw.h"
 
-const std::string LOG_ROOT =
+const std::string DEFAULT_LOG_ROOT =
     Hardware::PC() ? util::getenv_default("HOME", "/.comma/media/0/realdata", "/data/media/0/realdata")
                    : "/data/media/0/realdata";
+
+const std::string LOG_ROOT = util::getenv_default("LOG_ROOT", "", DEFAULT_LOG_ROOT.c_str());
+
 #define LOGGER_MAX_HANDLES 16
 
 class BZFile {

selfdrive/loggerd/loggerd.cc

@@ -345,8 +345,11 @@ int main(int argc, char** argv) {
     qlog_states[sock] = {.counter = 0, .freq = it.decimation};
   }
 
+  Params params;
+
   // init logger
   logger_init(&s.logger, "rlog", true);
+  params.put("CurrentRoute", s.logger.route_name);
 
   // init encoders
   pthread_mutex_init(&s.rotate_lock, NULL);
@@ -356,7 +359,7 @@ int main(int argc, char** argv) {
   encoder_threads.push_back(std::thread(encoder_thread, LOG_CAMERA_ID_FCAMERA));
   s.rotate_state[LOG_CAMERA_ID_FCAMERA].enabled = true;
 
-  if (!Hardware::PC() && Params().getBool("RecordFront")) {
+  if (!Hardware::PC() && params.getBool("RecordFront")) {
     encoder_threads.push_back(std::thread(encoder_thread, LOG_CAMERA_ID_DCAMERA));
     s.rotate_state[LOG_CAMERA_ID_DCAMERA].enabled = true;
   }

selfdrive/modeld/models/dmonitoring.cc

@@ -20,7 +20,7 @@
 #endif
 
 void dmonitoring_init(DMonitoringModelState* s) {
-  const char *model_path = "../../models/dmonitoring_model_q.dlc";
+  const char *model_path = Hardware::PC() ? "../../models/dmonitoring_model.dlc" : "../../models/dmonitoring_model_q.dlc";
   int runtime = USE_DSP_RUNTIME;
   s->m = new DefaultRunModel(model_path, &s->output[0], OUTPUT_SIZE, runtime);
   s->is_rhd = Params().getBool("IsRHD");

selfdrive/test/process_replay/.gitignore

@@ -0,0 +1 @@
+fakedata/

selfdrive/test/process_replay/inject_model.py

@@ -1,83 +0,0 @@
-#!/usr/bin/env python3
-
-import time
-
-from tqdm import tqdm
-
-from selfdrive.manager.process_config import managed_processes
-from cereal.messaging import PubMaster, recv_one, sub_sock
-from tools.lib.framereader import FrameReader
-
-
-def rreplace(s, old, new, occurrence):
-  li = s.rsplit(old, occurrence)
-  return new.join(li)
-
-
-def regen_model(msgs, pm, frame_reader, model_sock):
-  # Send some livecalibration messages to initalize visiond
-  for msg in msgs:
-    if msg.which() == 'liveCalibration':
-      pm.send('liveCalibration', msg.as_builder())
-
-  out_msgs = []
-  fidx = 0
-  for msg in tqdm(msgs):
-    w = msg.which()
-
-    if w == 'roadCameraState':
-      msg = msg.as_builder()
-
-      img = frame_reader.get(fidx, pix_fmt="rgb24")[0][:,:,::-1]
-
-      msg.roadCameraState.image = img.flatten().tobytes()
-
-      pm.send(w, msg)
-      model = recv_one(model_sock)
-      fidx += 1
-      out_msgs.append(model)
-    elif w == 'liveCalibration':
-      pm.send(w, msg.as_builder())
-
-  return out_msgs
-
-
-def inject_model(msgs, segment_name):
-  if segment_name.count('--') == 2:
-    segment_name = rreplace(segment_name, '--', '/', 1)
-  frame_reader = FrameReader('cd:/'+segment_name.replace("|", "/") + "/fcamera.hevc")
-
-  managed_processes['camerad'].start()
-  managed_processes['modeld'].start()
-  # TODO do better than just wait for modeld to boot
-  time.sleep(5)
-
-  pm = PubMaster(['liveCalibration', 'roadCameraState'])
-  model_sock = sub_sock('model')
-  try:
-    out_msgs = regen_model(msgs, pm, frame_reader, model_sock)
-  except (KeyboardInterrupt, SystemExit, Exception) as e:
-    managed_processes['modeld'].stop()
-    time.sleep(2)
-    managed_processes['camerad'].stop()
-    raise e
-  managed_processes['modeld'].stop()
-  time.sleep(2)
-  managed_processes['camerad'].stop()
-
-  new_msgs = []
-  midx = 0
-  for msg in msgs:
-    if (msg.which() == 'model') and (midx < len(out_msgs)):
-      model = out_msgs[midx].as_builder()
-      model.logMonoTime = msg.logMonoTime
-      model = model.as_reader()
-      new_msgs.append(model)
-      midx += 1
-    else:
-      new_msgs.append(msg)
-
-  print(len(new_msgs), len(list(msgs)))
-  assert abs(len(new_msgs) - len(list(msgs))) < 2
-
-  return new_msgs

selfdrive/test/process_replay/regen.py

@@ -0,0 +1,184 @@
+#!/usr/bin/env python3
+import argparse
+import os
+import time
+import multiprocessing
+from tqdm import tqdm
+
+# run DM procs
+os.environ["USE_WEBCAM"] = "1"
+
+import cereal.messaging as messaging
+from cereal.services import service_list
+from cereal.visionipc.visionipc_pyx import VisionIpcServer, VisionStreamType  # pylint: disable=no-name-in-module, import-error
+from common.params import Params
+from common.realtime import Ratekeeper, DT_MDL, DT_DMON
+from common.transformations.camera import eon_f_frame_size, eon_d_frame_size
+from selfdrive.car.fingerprints import FW_VERSIONS
+from selfdrive.manager.process import ensure_running
+from selfdrive.manager.process_config import managed_processes
+from selfdrive.test.update_ci_routes import upload_route
+from tools.lib.route import Route
+from tools.lib.framereader import FrameReader
+from tools.lib.logreader import LogReader
+
+
+process_replay_dir = os.path.dirname(os.path.abspath(__file__))
+FAKEDATA = os.path.join(process_replay_dir, "fakedata/")
+
+
+def replay_service(s, msgs):
+  pm = messaging.PubMaster([s, ])
+  rk = Ratekeeper(service_list[s].frequency, print_delay_threshold=None)
+  smsgs = [m for m in msgs if m.which() == s]
+  for m in smsgs:
+    pm.send(s, m.as_builder())
+    rk.keep_time()
+
+vs = None
+def replay_cameras(lr, frs):
+  cameras = [
+    ("roadCameraState", DT_MDL, eon_f_frame_size, VisionStreamType.VISION_STREAM_YUV_BACK),
+    ("driverCameraState", DT_DMON, eon_d_frame_size, VisionStreamType.VISION_STREAM_YUV_FRONT),
+  ]
+
+  def replay_camera(s, stream, dt, vipc_server, fr, size):
+    pm = messaging.PubMaster([s, ])
+    rk = Ratekeeper(1 / dt, print_delay_threshold=None)
+
+    img = b"\x00" * int(size[0]*size[1]*3/2)
+    while True:
+      if fr is not None and False:
+        img = fr.get(rk.frame % fr.frame_count, pix_fmt='yuv420p')[0]
+        img = img.flatten().tobytes()
+        print("got img")
+
+      rk.keep_time()
+
+      m = messaging.new_message(s)
+      msg = getattr(m, s)
+      msg.frameId = rk.frame
+      pm.send(s, m)
+
+      vipc_server.send(stream, img, msg.frameId, msg.timestampSof, msg.timestampEof)
+
+  # init vipc server and cameras
+  p = []
+  global vs
+  vs = VisionIpcServer("camerad")
+  for (s, dt, size, stream) in cameras:
+    fr = frs.get(s, None)
+    vs.create_buffers(stream, 40, False, size[0], size[1])
+    p.append(multiprocessing.Process(target=replay_camera,
+                                     args=(s, stream, dt, vs, fr, size)))
+
+  # hack to make UI work
+  vs.create_buffers(VisionStreamType.VISION_STREAM_RGB_BACK, 4, True, eon_f_frame_size[0], eon_f_frame_size[1])
+  vs.start_listener()
+  return p
+
+
+def regen_segment(lr, frs=None, outdir=FAKEDATA):
+
+  lr = list(lr)
+  if frs is None:
+    frs = dict()
+
+  # setup env
+  params = Params()
+  params.clear_all()
+  params.put_bool("Passive", False)
+  params.put_bool("OpenpilotEnabledToggle", True)
+  params.put_bool("CommunityFeaturesToggle", True)
+  params.put_bool("CommunityFeaturesToggle", True)
+  cal = messaging.new_message('liveCalibration')
+  cal.liveCalibration.validBlocks = 20
+  cal.liveCalibration.rpyCalib = [0.0, 0.0, 0.0]
+  params.put("CalibrationParams", cal.to_bytes())
+
+  os.environ["LOG_ROOT"] = outdir
+  os.environ["SIMULATION"] = "1"
+
+  os.environ['SKIP_FW_QUERY'] = ""
+  os.environ['FINGERPRINT'] = ""
+  for msg in lr:
+    if msg.which() == 'carParams':
+      car_fingerprint = msg.carParams.carFingerprint
+      if len(msg.carParams.carFw) and (car_fingerprint in FW_VERSIONS):
+        params.put("CarParamsCache", msg.carParams.as_builder().to_bytes())
+      else:
+        os.environ['SKIP_FW_QUERY'] = "1"
+        os.environ['FINGERPRINT'] = car_fingerprint
+
+  fake_daemons = {
+    'sensord': [
+      multiprocessing.Process(target=replay_service, args=('sensorEvents', lr)),
+    ],
+    'pandad': [
+      multiprocessing.Process(target=replay_service, args=('can', lr)),
+      multiprocessing.Process(target=replay_service, args=('pandaState', lr)),
+    ],
+    #'managerState': [
+    #  multiprocessing.Process(target=replay_service, args=('managerState', lr)),
+    #],
+    'thermald': [
+      multiprocessing.Process(target=replay_service, args=('deviceState', lr)),
+    ],
+    'camerad': [
+      *replay_cameras(lr, frs),
+    ],
+
+    # TODO: fix these and run them
+    'paramsd': [
+      multiprocessing.Process(target=replay_service, args=('liveParameters', lr)),
+    ],
+    'locationd': [
+      multiprocessing.Process(target=replay_service, args=('liveLocationKalman', lr)),
+    ],
+  }
+
+  try:
+    # start procs up
+    ignore = list(fake_daemons.keys()) + ['ui', 'manage_athenad', 'uploader']
+    ensure_running(managed_processes.values(), started=True, not_run=ignore)
+    for procs in fake_daemons.values():
+      for p in procs:
+        p.start()
+
+    for _ in tqdm(range(60)):
+      # ensure all procs are running
+      for d, procs in fake_daemons.items():
+        for p in procs:
+          if not p.is_alive():
+            raise Exception(f"{d}'s {p.name} died")
+      time.sleep(1)
+  finally:
+    # kill everything
+    for p in managed_processes.values():
+      p.stop()
+    for procs in fake_daemons.values():
+      for p in procs:
+        p.terminate()
+
+  r = params.get("CurrentRoute", encoding='utf-8')
+  return os.path.join(outdir, r + "--0")
+
+if __name__ == "__main__":
+
+  parser = argparse.ArgumentParser(description="Generate new segments from old ones")
+  parser.add_argument("--upload", action="store_true", help="Upload the new segment to the CI bucket")
+  parser.add_argument("route", type=str, help="The source route")
+  parser.add_argument("seg", type=int, help="Segment in source route")
+  args = parser.parse_args()
+
+  r = Route(args.route)
+  lr = LogReader(r.log_paths()[args.seg])
+  fr = FrameReader(r.camera_paths()[args.seg])
+  rpath = regen_segment(lr, {'roadCameraState': fr})
+  relr = os.path.relpath(rpath)
+
+  print("\n\n", "*"*30, "\n\n")
+  print("New route:", relr, "\n")
+  if args.upload:
+    upload_route(relr)
+

selfdrive/test/process_replay/test_processes.py

@@ -5,11 +5,11 @@ import sys
 from typing import Any
 
 from selfdrive.car.car_helpers import interface_names
+from selfdrive.test.openpilotci import get_url
 from selfdrive.test.process_replay.compare_logs import compare_logs
 from selfdrive.test.process_replay.process_replay import CONFIGS, replay_process
 from tools.lib.logreader import LogReader
 
-INJECT_MODEL = 0
 
 segments = [
   ("HYUNDAI", "02c45f73a2e5c6e9|2021-01-01--19-08-22--1"),     # HYUNDAI.SONATA
@@ -37,18 +37,6 @@ BASE_URL = "https://commadataci.blob.core.windows.net/openpilotci/"
 FULL_TEST = len(sys.argv) <= 1
 
 
-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
-
-
 def test_process(cfg, lr, cmp_log_fn, ignore_fields=None, ignore_msgs=None):
   if ignore_fields is None:
     ignore_fields = []
@@ -153,8 +141,8 @@ if __name__ == "__main__":
 
     results[segment] = {}
 
-    rlog_fn = get_segment(segment)
-    lr = LogReader(rlog_fn)
+    r, n = segment.rsplit("--", 1)
+    lr = LogReader(get_url(r, n))
 
     for cfg in CONFIGS:
       if (procs_whitelisted and cfg.proc_name not in args.whitelist_procs) or \

selfdrive/test/process_replay/update_model.py

@@ -1,47 +0,0 @@
-#!/usr/bin/env python3
-import os
-import sys
-
-from selfdrive.test.openpilotci import upload_file
-from selfdrive.test.process_replay.compare_logs import save_log
-from selfdrive.test.process_replay.test_processes import segments, get_segment
-from selfdrive.version import get_git_commit
-from tools.lib.logreader import LogReader
-from selfdrive.test.process_replay.inject_model import inject_model
-
-if __name__ == "__main__":
-
-  no_upload = "--no-upload" in sys.argv
-
-  process_replay_dir = os.path.dirname(os.path.abspath(__file__))
-  ref_commit_fn = os.path.join(process_replay_dir, "model_ref_commit")
-
-  ref_commit = get_git_commit()
-  if ref_commit is None:
-    raise Exception("couldn't get ref commit")
-  with open(ref_commit_fn, "w") as f:
-    f.write(ref_commit)
-
-  for car_brand, segment in segments:
-    rlog_fn = get_segment(segment, original=True)
-
-    if rlog_fn is None:
-      print("failed to get segment %s" % segment)
-      sys.exit(1)
-
-    lr = LogReader(rlog_fn)
-    print('injecting model into % s' % segment)
-    lr = inject_model(lr, segment)
-
-    route_name, segment_num = segment.rsplit("--", 1)
-    log_fn = "%s/%s/rlog_%s.bz2" % (route_name.replace("|", "/"), segment_num, ref_commit)
-    tmp_name = 'tmp_%s_%s' % (route_name, segment_num)
-    save_log(tmp_name, lr)
-
-    if not no_upload:
-      upload_file(tmp_name, log_fn)
-      print('uploaded %s', log_fn)
-      os.remove(tmp_name)
-    os.remove(rlog_fn)
-
-  print("done")

selfdrive/test/process_replay/update_refs.py

@@ -2,10 +2,10 @@
 import os
 import sys
 
-from selfdrive.test.openpilotci import upload_file
+from selfdrive.test.openpilotci import upload_file, get_url
 from selfdrive.test.process_replay.compare_logs import save_log
 from selfdrive.test.process_replay.process_replay import replay_process, CONFIGS
-from selfdrive.test.process_replay.test_processes import segments, get_segment
+from selfdrive.test.process_replay.test_processes import segments
 from selfdrive.version import get_git_commit
 from tools.lib.logreader import LogReader
 
@@ -23,13 +23,8 @@ if __name__ == "__main__":
     f.write(ref_commit)
 
   for car_brand, segment in segments:
-    rlog_fn = get_segment(segment)
-
-    if rlog_fn is None:
-      print("failed to get segment %s" % segment)
-      sys.exit(1)
-
-    lr = LogReader(rlog_fn)
+    r, n = segment.rsplit("--", 1)
+    lr = LogReader(get_url(r, n))
 
     for cfg in CONFIGS:
       log_msgs = replay_process(cfg, lr)

selfdrive/test/test_valgrind_replay.py

@@ -14,7 +14,7 @@ else:
 import cereal.messaging as messaging
 from collections import namedtuple
 from tools.lib.logreader import LogReader
-from selfdrive.test.process_replay.test_processes import get_segment
+from selfdrive.test.openpilotci import get_url
 from common.basedir import BASEDIR
 
 ProcessConfig = namedtuple('ProcessConfig', ['proc_name', 'pub_sub', 'ignore', 'command', 'path', 'segment', 'wait_for_response'])
@@ -98,8 +98,8 @@ class TestValgrind(unittest.TestCase):
 
     for cfg in CONFIGS:
       self.done = False
-      URL = cfg.segment
-      lr = LogReader(get_segment(URL))
+      r, n = cfg.segment.rsplit("--", 1)
+      lr = LogReader(get_url(r, n))
       self.replay_process(cfg, lr)
       time.sleep(1)  # Wait for the logs to get written
       self.assertFalse(self.leak)

selfdrive/test/update_ci_routes.py

@@ -18,6 +18,20 @@ DEST_KEY = azureutil.get_user_token(_DATA_ACCOUNT_CI, "openpilotci")
 SOURCE_KEYS = [azureutil.get_user_token(account, bucket) for account, bucket in SOURCES]
 SERVICE = BlockBlobService(_DATA_ACCOUNT_CI, sas_token=DEST_KEY)
 
+def upload_route(path):
+  r, n = path.rsplit("--", 1)
+  destpath = f"{r}/{n}"
+  cmd = [
+    "azcopy",
+    "copy",
+    f"{path}/*",
+    "https://{}.blob.core.windows.net/{}/{}?{}".format(_DATA_ACCOUNT_CI, "openpilotci", destpath, DEST_KEY),
+    "--recursive=false",
+    "--overwrite=false",
+    "--exclude-pattern=*/dcamera.hevc",
+  ]
+  subprocess.check_call(cmd)
+
 def sync_to_ci_public(route):
   key_prefix = route.replace('|', '/')
   dongle_id = key_prefix.split('/')[0]