diff --git a/Jenkinsfile b/Jenkinsfile index 6f81755d4..e58ac817e 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -219,15 +219,13 @@ node { 'camerad OX03C10': { deviceStage("OX03C10", "tizi-ox03c10", ["UNSAFE=1"], [ step("build", "cd system/manager && ./build.py"), - step("test camerad", "pytest system/camerad/test/test_camerad.py", [timeout: 60]), - step("test exposure", "pytest system/camerad/test/test_exposure.py"), + step("test camerad", "pytest system/camerad/test/test_camerad.py", [timeout: 90]), ]) }, 'camerad OS04C10': { deviceStage("OS04C10", "tici-os04c10", ["UNSAFE=1"], [ step("build", "cd system/manager && ./build.py"), - step("test camerad", "pytest system/camerad/test/test_camerad.py", [timeout: 60]), - step("test exposure", "pytest system/camerad/test/test_exposure.py"), + step("test camerad", "pytest system/camerad/test/test_camerad.py", [timeout: 90]), ]) }, 'sensord': { diff --git a/selfdrive/test/helpers.py b/selfdrive/test/helpers.py index 81635aa31..5dfc1c3ec 100644 --- a/selfdrive/test/helpers.py +++ b/selfdrive/test/helpers.py @@ -37,6 +37,43 @@ def release_only(f): return wrap +def collect_logs(services, duration): + socks = [messaging.sub_sock(s, conflate=False, timeout=100) for s in services] + logs = [] + start = time.monotonic() + while time.monotonic() - start < duration: + for s in socks: + logs.extend(messaging.drain_sock(s)) + return logs + + +@contextlib.contextmanager +def log_collector(services): + """Background thread that continuously drains messages from services. + Use when the main thread needs to do blocking work (e.g. capturing images).""" + socks = [messaging.sub_sock(s, conflate=False, timeout=100) for s in services] + raw_logs = [] + lock = threading.Lock() + stop_event = threading.Event() + + def _drain(): + while not stop_event.is_set(): + for s in socks: + msgs = messaging.drain_sock(s) + if msgs: + with lock: + raw_logs.extend(msgs) + time.sleep(0.01) + + thread = threading.Thread(target=_drain, daemon=True) + thread.start() + try: + yield raw_logs, lock + finally: + stop_event.set() + thread.join(timeout=2) + + @contextlib.contextmanager def processes_context(processes, init_time=0, ignore_stopped=None): ignore_stopped = [] if ignore_stopped is None else ignore_stopped diff --git a/system/camerad/test/test_camerad.py b/system/camerad/test/test_camerad.py index 1f3f97b08..5f8de8689 100644 --- a/system/camerad/test/test_camerad.py +++ b/system/camerad/test/test_camerad.py @@ -3,51 +3,103 @@ import time import pytest import numpy as np -import cereal.messaging as messaging from cereal.services import SERVICE_LIST -from openpilot.system.manager.process_config import managed_processes from openpilot.tools.lib.log_time_series import msgs_to_time_series +from openpilot.system.camerad.snapshot import get_snapshots +from openpilot.selfdrive.test.helpers import collect_logs, log_collector, processes_context TEST_TIMESPAN = 10 CAMERAS = ('roadCameraState', 'driverCameraState', 'wideRoadCameraState') +EXPOSURE_STABLE_COUNT = 3 +EXPOSURE_RANGE = (0.15, 0.35) +MAX_TEST_TIME = 25 + + +def _numpy_rgb2gray(im): + return np.clip(im[:,:,2] * 0.114 + im[:,:,1] * 0.587 + im[:,:,0] * 0.299, 0, 255).astype(np.uint8) + +def _exposure_stats(im): + h, w = im.shape[:2] + gray = _numpy_rgb2gray(im[h//10:9*h//10, w//10:9*w//10]) + return float(np.median(gray) / 255.), float(np.mean(gray) / 255.) + +def _in_range(median, mean): + lo, hi = EXPOSURE_RANGE + return lo < median < hi and lo < mean < hi + +def _exposure_stable(results): + return all( + len(v) >= EXPOSURE_STABLE_COUNT and all(_in_range(*s) for s in v[-EXPOSURE_STABLE_COUNT:]) + for v in results.values() + ) def run_and_log(procs, services, duration): - logs = [] - - try: - for p in procs: - managed_processes[p].start() - socks = [messaging.sub_sock(s, conflate=False, timeout=100) for s in services] - - start_time = time.monotonic() - while time.monotonic() - start_time < duration: - for s in socks: - logs.extend(messaging.drain_sock(s)) - for p in procs: - assert managed_processes[p].proc.is_alive() - finally: - for p in procs: - managed_processes[p].stop() - - return logs + with processes_context(procs): + return collect_logs(services, duration) @pytest.fixture(scope="module") -def logs(): - logs = run_and_log(["camerad", ], CAMERAS, TEST_TIMESPAN) - ts = msgs_to_time_series(logs) +def _camera_session(): + """Single camerad session that collects logs and exposure data. + Runs until exposure stabilizes (min TEST_TIMESPAN seconds for enough log data).""" + with processes_context(["camerad"]), log_collector(CAMERAS) as (raw_logs, lock): + exposure = {cam: [] for cam in CAMERAS} + start = time.monotonic() + while time.monotonic() - start < MAX_TEST_TIME: + rpic, dpic = get_snapshots(frame="roadCameraState", front_frame="driverCameraState") + wpic, _ = get_snapshots(frame="wideRoadCameraState") + for cam, img in zip(CAMERAS, [rpic, dpic, wpic], strict=True): + exposure[cam].append(_exposure_stats(img)) + + if time.monotonic() - start >= TEST_TIMESPAN and _exposure_stable(exposure): + break + + elapsed = time.monotonic() - start + + with lock: + ts = msgs_to_time_series(raw_logs) for cam in CAMERAS: - expected_frames = SERVICE_LIST[cam].frequency * TEST_TIMESPAN + expected_frames = SERVICE_LIST[cam].frequency * elapsed cnt = len(ts[cam]['t']) assert expected_frames*0.8 < cnt < expected_frames*1.2, f"unexpected frame count {cam}: {expected_frames=}, got {cnt}" dts = np.abs(np.diff([ts[cam]['timestampSof']/1e6]) - 1000/SERVICE_LIST[cam].frequency) assert (dts < 1.0).all(), f"{cam} dts(ms) out of spec: max diff {dts.max()}, 99 percentile {np.percentile(dts, 99)}" - return ts + + return ts, exposure + +@pytest.fixture(scope="module") +def logs(_camera_session): + return _camera_session[0] + +@pytest.fixture(scope="module") +def exposure_data(_camera_session): + return _camera_session[1] @pytest.mark.tici class TestCamerad: + @pytest.mark.parametrize("cam", CAMERAS) + def test_camera_exposure(self, exposure_data, cam): + lo, hi = EXPOSURE_RANGE + checks = exposure_data[cam] + assert len(checks) >= EXPOSURE_STABLE_COUNT, f"{cam}: only got {len(checks)} samples" + + # check that exposure converges into the valid range + passed = sum(_in_range(med, mean) for med, mean in checks) + assert passed >= EXPOSURE_STABLE_COUNT, \ + f"{cam}: only {passed}/{len(checks)} checks in range. " + \ + " | ".join(f"#{i+1}: med={m:.4f} mean={u:.4f}" for i, (m, u) in enumerate(checks)) + + # check that exposure is stable once converged (no regressions) + in_range = False + for i, (median, mean) in enumerate(checks): + ok = _in_range(median, mean) + if in_range and not ok: + pytest.fail(f"{cam}: exposure regressed on sample {i+1} " + + f"(median={median:.4f}, mean={mean:.4f}, expected: ({lo}, {hi}))") + in_range = ok + def test_frame_skips(self, logs): for c in CAMERAS: assert set(np.diff(logs[c]['frameId'])) == {1, }, f"{c} has frame skips" @@ -91,7 +143,10 @@ class TestCamerad: def test_stress_test(self): os.environ['SPECTRA_ERROR_PROB'] = '0.008' - logs = run_and_log(["camerad", ], CAMERAS, 10) + try: + logs = run_and_log(["camerad", ], CAMERAS, 10) + finally: + del os.environ['SPECTRA_ERROR_PROB'] ts = msgs_to_time_series(logs) # we should see some jumps from introduced errors diff --git a/system/camerad/test/test_exposure.py b/system/camerad/test/test_exposure.py deleted file mode 100644 index 6f89e0480..000000000 --- a/system/camerad/test/test_exposure.py +++ /dev/null @@ -1,51 +0,0 @@ -import time -import numpy as np -import pytest - -from openpilot.selfdrive.test.helpers import with_processes -from openpilot.system.camerad.snapshot import get_snapshots - -TEST_TIME = 45 -REPEAT = 5 - -@pytest.mark.tici -class TestCamerad: - @classmethod - def setup_class(cls): - pass - - def _numpy_rgb2gray(self, im): - ret = np.clip(im[:,:,2] * 0.114 + im[:,:,1] * 0.587 + im[:,:,0] * 0.299, 0, 255).astype(np.uint8) - return ret - - def _is_exposure_okay(self, i, med_mean=None): - if med_mean is None: - med_mean = np.array([[0.18,0.3],[0.18,0.3]]) - h, w = i.shape[:2] - i = i[h//10:9*h//10,w//10:9*w//10] - med_ex, mean_ex = med_mean - i = self._numpy_rgb2gray(i) - i_median = np.median(i) / 255. - i_mean = np.mean(i) / 255. - print([i_median, i_mean]) - return med_ex[0] < i_median < med_ex[1] and mean_ex[0] < i_mean < mean_ex[1] - - @with_processes(['camerad']) - def test_camera_operation(self): - passed = 0 - start = time.monotonic() - while time.monotonic() - start < TEST_TIME and passed < REPEAT: - rpic, dpic = get_snapshots(frame="roadCameraState", front_frame="driverCameraState") - wpic, _ = get_snapshots(frame="wideRoadCameraState") - - res = self._is_exposure_okay(rpic) - res = res and self._is_exposure_okay(dpic) - res = res and self._is_exposure_okay(wpic) - - if passed > 0 and not res: - passed = -passed # fails test if any failure after first sus - break - - passed += int(res) - time.sleep(2) - assert passed >= REPEAT