diff --git a/openpilot/selfdrive/modeld/SConscript b/openpilot/selfdrive/modeld/SConscript index 30a31aae271f8e..d2ce07fa155566 100644 --- a/openpilot/selfdrive/modeld/SConscript +++ b/openpilot/selfdrive/modeld/SConscript @@ -10,11 +10,6 @@ from openpilot.selfdrive.modeld.constants import ModelConstants from openpilot.selfdrive.modeld.helpers import TG_INPUT_DEVICES_PATH, usbgpu_present, modeld_pkl_path -CAMERA_CONFIGS = [ - (_ar_ox_fisheye.width, _ar_ox_fisheye.height), # tici: 1928x1208 - (_os_fisheye.width, _os_fisheye.height), # mici: 1344x760 -] - Import('env', 'arch') chunker_file = File("#openpilot/common/file_chunker.py") lenv = env.Clone() @@ -24,20 +19,20 @@ tinygrad_files = ["#"+x for x in glob.glob(env.Dir("#tinygrad_repo").relpath + " if 'pycache' not in x and os.path.isfile(os.path.join(tinygrad_root, x))] def estimate_pickle_max_size(onnx_size): - return 1.2 * onnx_size + 10 * 1024 * 1024 # 20% + 10MB is plenty + return 1.2 * onnx_size + 10 * 1024 * 1024 # 20% + 10MB is plenty if arch == 'comma_arm64': + from openpilot.common.hardware import HARDWARE + camera = _os_fisheye if HARDWARE.get_device_type() == "mici" else _ar_ox_fisheye + camera_configs = [(camera.width, camera.height)] tg_backend = 'QCOM' tg_flags = f'DEV={tg_backend} IMAGE=1 FLOAT16=1 NOLOCALS=1 JIT_BATCH_SIZE=0 OPENPILOT_HACKS=1' else: + camera_configs = [(c.width, c.height) for c in (_ar_ox_fisheye, _os_fisheye)] tg_backend = 'CPU' tg_flags = f'DEV=CPU' if arch == 'Darwin' else 'DEV=CPU:LLVM' tg_devices = { # which device to put jit inputs to at runtime - 'openpilot.selfdrive.modeld.modeld': { - 'default': {'WARP_DEV': tg_backend, 'QUEUE_DEV': tg_backend}, - 'usbgpu': {'WARP_DEV': tg_backend, 'QUEUE_DEV': 'AMD'} - }, 'openpilot.selfdrive.modeld.dmonitoringmodeld': { 'default': {'DEV': tg_backend} }, @@ -45,7 +40,7 @@ tg_devices = { # which device to put jit inputs to at runtime USBGPU = usbgpu_present() if USBGPU: - usbgpu_tg_flags = f'DEBUG=2 DEV=USB+AMD:LLVM WARP_DEV={tg_backend} FLOAT16=1 JIT_BATCH_SIZE=0 GMMU=0 TC_OPT=2' + usbgpu_tg_flags = 'DEBUG=1 DEV=USB+AMD:LLVM FRAME_DEV=CPU FLOAT16=1 JIT_BATCH_SIZE=0 GMMU=0 TC_OPT=2' # the USB+AMD GPU takes an exclusive flock; serialize all targets that touch it usbgpu_lock = File("models/.usb_gpu.lock").abspath @@ -78,7 +73,7 @@ for usbgpu in [False, True] if USBGPU else [False]: # BIG_INTO_SMALL=1 builds the default target from the big model, e.g. to test it without a USB GPU file_prefix, cmd_flags = ('big_', usbgpu_tg_flags) if usbgpu else ('big_' if os.getenv('BIG_INTO_SMALL') else '', tg_flags) driving_onnx_deps = get_existing_chunks(File(f"models/{file_prefix}driving_supercombo.onnx").abspath) - camera_res_args = ' '.join(f'{cw}x{ch}' for cw, ch in CAMERA_CONFIGS) + camera_res_args = ' '.join(f'{cw}x{ch}' for cw, ch in camera_configs) # CPU 7 is isolated with isolcpus on AGNOS, so explicitly pin the compiler to it. taskset = 'taskset -c 7 ' if arch == 'comma_arm64' else '' cmd = (f'{cmd_flags} {mac_brew_string} {taskset}python3 {modeld_dir}/compile_modeld.py ' @@ -106,7 +101,7 @@ for usbgpu in [False, True] if USBGPU else [False]: actions = Action(do_compile, " [USBGPU] $TARGET") if usbgpu else [cmd, Action(do_chunk, " [CHUNK] $TARGET")] node = lenv.Command( chunk_targets, - tinygrad_files + compile_modeld_script + driving_onnx_deps + [Value(chunk_targets), chunker_file], + tinygrad_files + compile_modeld_script + driving_onnx_deps + [Value(camera_res_args), Value(chunk_targets), chunker_file], actions, ) if usbgpu: @@ -120,7 +115,7 @@ lenv.Command(fn + "_metadata.pkl", [fn + ".onnx"] + tinygrad_files + script_file dm_w, dm_h = DM_INPUT_SIZE compile_dm_warp_script = [File(f"{modeld_dir}/compile_dm_warp.py")] -for cam_w, cam_h in CAMERA_CONFIGS: +for cam_w, cam_h in camera_configs: dm_pkl_path = File(f"models/dm_warp_{cam_w}x{cam_h}_tinygrad.pkl").abspath cmd = (f'{tg_flags} {mac_brew_string} python3 {modeld_dir}/compile_dm_warp.py ' f'--camera-resolution {cam_w}x{cam_h} --warp-to {dm_w}x{dm_h} ' diff --git a/openpilot/selfdrive/modeld/compile_modeld.py b/openpilot/selfdrive/modeld/compile_modeld.py index 2d27a41496e8b4..a24c940039655e 100755 --- a/openpilot/selfdrive/modeld/compile_modeld.py +++ b/openpilot/selfdrive/modeld/compile_modeld.py @@ -37,13 +37,12 @@ def fetch_fw(path, name, sha256): NV12Frame = namedtuple("NV12Frame", ['width', 'height', 'stride', 'y_height', 'uv_height', 'size']) -WARP_INPUTS = ['tfm', 'big_tfm'] -POLICY_INPUTS = ['img_q', 'big_img_q', 'feat_q', 'desire_q', 'packed_npy_inputs'] +MODELD_INPUTS = ['img_q', 'big_img_q', 'feat_q', 'desire_q', 'packed_npy_inputs'] -UV_SCALE_MATRIX = np.array([[0.5, 0, 0], [0, 0.5, 0], [0, 0, 1]], dtype=np.float32) -UV_SCALE_MATRIX_INV = np.linalg.inv(UV_SCALE_MATRIX) -WARP_DEV = os.getenv('WARP_DEV') +def nv12_copy_size(stride: int, y_height: int, uv_height: int) -> int: + # Retain the padded Y and UV plane storage, but skip the trailing kernel/guard allocation. + return stride * (y_height + uv_height) def make_random_images(keys, shape, device=None): @@ -99,7 +98,7 @@ def make_frame_prepare(nv12: NV12Frame, model_w, model_h): def frame_prepare_tinygrad(input_frame, M_inv): # UV_SCALE @ M_inv @ UV_SCALE_INV simplifies to elementwise scaling - M_inv_uv = M_inv * Tensor([[1.0, 1.0, 0.5], [1.0, 1.0, 0.5], [2.0, 2.0, 1.0]], device=WARP_DEV) + M_inv_uv = M_inv * Tensor([[1.0, 1.0, 0.5], [1.0, 1.0, 0.5], [2.0, 2.0, 1.0]], device=Device.DEFAULT) # deinterleave NV12 UV plane (UVUV... -> separate U, V) uv = input_frame[uv_offset:uv_offset + uv_height * stride].reshape(uv_height, stride) with Context(SPLIT_REDUCEOP=0): @@ -118,23 +117,6 @@ def frame_prepare_tinygrad(input_frame, M_inv): return frame_prepare_tinygrad -def make_warp_input_queues(vision_input_shapes, frame_skip, device): - img = vision_input_shapes['img'] # (1, 12, 128, 256) - n_frames = img[1] // 6 - img_buf_shape = (frame_skip * (n_frames - 1) + 1, 6, img[2], img[3]) - - npy = { - 'tfm': np.zeros((3, 3), dtype=np.float32), - 'big_tfm': np.zeros((3, 3), dtype=np.float32), - } - input_queues = { - 'img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), - 'big_img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), - **{k: Tensor(v, device='NPY').realize() for k, v in npy.items()}, - } - return input_queues, npy - - def get_policy_npy_shapes(input_shapes): dp = input_shapes['desire_pulse'] # (1, 25, 8) tc = input_shapes['traffic_convention'] # (1, 2) @@ -146,20 +128,25 @@ def get_policy_npy_shapes(input_shapes): def make_input_queues(input_shapes, frame_skip, device): - input_queues, npy = make_warp_input_queues(input_shapes, frame_skip, device) - + img = input_shapes['img'] # (1, 12, 128, 256) fb = input_shapes['features_buffer'] # (1, 24, 512), past features only; the model appends the current frame's feature dp = input_shapes['desire_pulse'] # (1, 25, 8) + n_frames = img[1] // 6 + img_buf_shape = (frame_skip * (n_frames - 1) + 1, 6, img[2], img[3]) - shapes, sizes = get_policy_npy_shapes(input_shapes) + policy_shapes, _ = get_policy_npy_shapes(input_shapes) + shapes = {'tfm': (3, 3), 'big_tfm': (3, 3)} | policy_shapes + sizes = [math.prod(s) for s in shapes.values()] packed_npy_inputs = np.zeros(sum(sizes), dtype=np.float32) # views into the packed inputs, to be refilled at runtime - npy.update({k: v.reshape(s) for (k, s), v in zip(shapes.items(), np.split(packed_npy_inputs, np.cumsum(sizes[:-1])), strict=True)}) - input_queues.update({ + npy = {k: v.reshape(s) for (k, s), v in zip(shapes.items(), np.split(packed_npy_inputs, np.cumsum(sizes[:-1])), strict=True)} + input_queues = { + 'img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), + 'big_img_q': Tensor(np.zeros(img_buf_shape, dtype=np.uint8), device=device).contiguous().realize(), 'feat_q': Tensor(np.zeros((frame_skip * fb[1], fb[0], fb[2]), dtype=np.float32), device=device).contiguous().realize(), 'desire_q': Tensor(np.zeros((frame_skip * dp[1], dp[0], dp[2]), dtype=np.float32), device=device).contiguous().realize(), 'packed_npy_inputs': Tensor(packed_npy_inputs, device='NPY').realize(), - }) + } return input_queues, npy @@ -176,13 +163,17 @@ def sample_desire(buf, frame_skip): return buf.reshape(-1, frame_skip, *buf.shape[1:]).max(1).flatten(0, 1).unsqueeze(0) -def make_warp(nv12, model_w, model_h, frame_skip): +def make_warp(nv12, model_w, model_h): frame_prepare = make_frame_prepare(nv12, model_w, model_h) def warp(tfm, big_tfm, frame, big_frame): - tfm = tfm.to(WARP_DEV) - big_tfm = big_tfm.to(WARP_DEV) - Tensor.realize(tfm, big_tfm) + tfm = tfm.to(Device.DEFAULT) + big_tfm = big_tfm.to(Device.DEFAULT) + # frame and big_frame are exact-sized base buffers. A sliced JIT input is + # pruned as capture-only by tinygrad, which would replay the captured frame. + frame = frame.to(Device.DEFAULT) + big_frame = big_frame.to(Device.DEFAULT) + Tensor.realize(tfm, big_tfm, frame, big_frame) warped_frame = frame_prepare(frame, tfm).unsqueeze(0) warped_big_frame = frame_prepare(big_frame, big_tfm).unsqueeze(0) @@ -195,10 +186,10 @@ def make_run_policy(model_runner, model_metadata, frame_skip): sample_desire_fn = partial(sample_desire, frame_skip=frame_skip) sample_skip_fn = partial(sample_skip, frame_skip=frame_skip) npy_shapes, npy_sizes = get_policy_npy_shapes(model_metadata['input_shapes']) + model_input_dtypes = {name: spec.dtype for name, spec in model_runner.graph_inputs.items()} def run_policy(warped, img_q, big_img_q, feat_q, desire_q, packed_npy_inputs): packed_npy_inputs = packed_npy_inputs.to(Device.DEFAULT) - warped = warped.to(Device.DEFAULT) Tensor.realize(packed_npy_inputs, warped) img = shift_and_sample(img_q, warped[0:1], sample_skip_fn) @@ -216,11 +207,24 @@ def run_policy(warped, img_q, big_img_q, feat_q, desire_q, packed_npy_inputs): 'traffic_convention': traffic_convention, 'action_t': action_t, } + inputs = {name: value.cast(model_input_dtypes[name]) for name, value in inputs.items()} out = next(iter(model_runner(inputs).values())).cast('float32') return out, return run_policy +def make_run_model(warp, run_policy, model_metadata): + _, policy_sizes = get_policy_npy_shapes(model_metadata['input_shapes']) + + def run_model(img_q, big_img_q, feat_q, desire_q, packed_npy_inputs, frame, big_frame): + packed_npy_inputs = packed_npy_inputs.to(Device.DEFAULT) + Tensor.realize(packed_npy_inputs) + tfm, big_tfm, policy_inputs = packed_npy_inputs.split([9, 9, sum(policy_sizes)]) + warped = warp(tfm.reshape(3, 3), big_tfm.reshape(3, 3), frame, big_frame) + return run_policy(warped, img_q, big_img_q, feat_q, desire_q, policy_inputs) + return run_model + + def compile_jit(jit, make_random_inputs, input_keys, make_queues): SEED = 42 def random_inputs_run(fn, seed, test_val=None, test_buffers=None, expect_match=True): @@ -261,9 +265,10 @@ def random_inputs_run(fn, seed, test_val=None, test_buffers=None, expect_match=T with tempfile.TemporaryFile(dir=".") as f: dump_oob(jit, f) f.seek(0) - jit = load_oob(f) - random_inputs_run(jit, SEED, test_val, test_buffers, expect_match=True) - random_inputs_run(jit, SEED+1, test_val, test_buffers, expect_match=False) + loaded_jit = load_oob(f) + random_inputs_run(loaded_jit, SEED, test_val, test_buffers, expect_match=True) + random_inputs_run(loaded_jit, SEED+1, test_val, test_buffers, expect_match=False) + # Keep the original so per-resolution JITs share model weight buffers in the final pickle. return jit @@ -293,26 +298,32 @@ def read_file_chunked_to_disk(path): p.add_argument('--output', required=True) p.add_argument('--frame-skip', type=int, required=True) args = p.parse_args() + frame_device = Device.canonicalize(os.getenv('FRAME_DEV', Device.DEFAULT)) model_path = read_file_chunked_to_disk(args.onnx) model_w, model_h = args.model_size model_runner = OnnxRunner(model_path) - out = {'metadata': make_metadata_dict(model_path)} - - run_policy_jit = TinyJit(make_run_policy(model_runner, out['metadata'], args.frame_skip), prune=True) + out = { + 'metadata': make_metadata_dict(model_path), + 'input_devices': { + 'frame': frame_device, + 'model': Device.DEFAULT, + }, + 'run_model': {}, + } - make_policy_queues = partial(make_input_queues, out['metadata']['input_shapes'], args.frame_skip) - make_random_model_inputs = partial(make_random_images, keys=['warped'], shape=(2, 6, *out['metadata']['input_shapes']['img'][2:]), device=WARP_DEV) - out['run_policy'] = compile_jit(run_policy_jit, make_random_model_inputs, POLICY_INPUTS, - make_policy_queues) + run_policy = make_run_policy(model_runner, out['metadata'], args.frame_skip) + make_model_queues = partial(make_input_queues, out['metadata']['input_shapes'], args.frame_skip) for cam_w, cam_h in args.camera_resolutions: nv12 = NV12Frame(cam_w, cam_h, *get_nv12_info(cam_w, cam_h)) - make_random_warp_inputs = partial(make_random_images, keys=['frame', 'big_frame'], shape=nv12.size, device=WARP_DEV) - warp = TinyJit(make_warp(nv12, model_w, model_h, args.frame_skip), prune=True) - make_warp_queues = partial(make_warp_input_queues, out['metadata']['input_shapes'], args.frame_skip) - out[(cam_w,cam_h)] = compile_jit(warp, make_random_warp_inputs, WARP_INPUTS, make_warp_queues) + frame_copy_size = nv12_copy_size(nv12.stride, nv12.y_height, nv12.uv_height) + make_random_frame_inputs = partial(make_random_images, keys=['frame', 'big_frame'], shape=frame_copy_size, device=frame_device) + warp = make_warp(nv12, model_w, model_h) + run_model_jit = TinyJit(make_run_model(warp, run_policy, out['metadata']), prune=True) + out['run_model'][(cam_w,cam_h)] = compile_jit(run_model_jit, make_random_frame_inputs, MODELD_INPUTS, + make_model_queues) with open(args.output, "wb") as f: dump_oob(out, f) diff --git a/openpilot/selfdrive/modeld/modeld.py b/openpilot/selfdrive/modeld/modeld.py index 7f049912ddfd8a..a816d53b438ddd 100755 --- a/openpilot/selfdrive/modeld/modeld.py +++ b/openpilot/selfdrive/modeld/modeld.py @@ -26,13 +26,12 @@ from openpilot.selfdrive.controls.lib.desire_helper import DesireHelper from openpilot.selfdrive.controls.lib.drive_helpers import get_accel_from_plan, should_stop, smooth_value, get_curvature_from_plan from openpilot.selfdrive.modeld.parse_model_outputs import Parser -from openpilot.selfdrive.modeld.compile_modeld import make_input_queues, WARP_INPUTS, POLICY_INPUTS +from openpilot.selfdrive.modeld.compile_modeld import make_input_queues, nv12_copy_size, MODELD_INPUTS from openpilot.selfdrive.modeld.fill_model_msg import fill_model_msg, fill_driving_model_data, fill_pose_msg, PublishState from openpilot.common.file_chunker import open_file_chunked from openpilot.selfdrive.modeld.constants import ModelConstants, Plan -from openpilot.selfdrive.modeld.helpers import usbgpu_present, usbgpu_compiled, modeld_pkl_path, get_tg_input_devices, load_oob +from openpilot.selfdrive.modeld.helpers import usbgpu_present, usbgpu_compiled, modeld_pkl_path, load_oob -PROCESS_NAME = "openpilot.selfdrive.modeld.modeld" SEND_RAW_PRED = os.getenv('SEND_RAW_PRED') LAT_SMOOTH_SECONDS = 0.0 @@ -138,9 +137,9 @@ class ModelState: prev_desire: np.ndarray # for tracking the rising edge of the pulse def __init__(self, cam_w: int, cam_h: int, usbgpu: bool): - input_devices = get_tg_input_devices(PROCESS_NAME, usbgpu) - self.WARP_DEV, self.QUEUE_DEV = input_devices['WARP_DEV'], input_devices['QUEUE_DEV'] jits = load_oob(open_file_chunked(modeld_pkl_path(usbgpu))) + input_devices = jits['input_devices'] + self.frame_device, self.model_device = input_devices['frame'], input_devices['model'] metadata = jits['metadata'] self.input_shapes = metadata['input_shapes'] self.vision_input_names = [k for k in self.input_shapes if 'img' in k] @@ -150,13 +149,12 @@ def __init__(self, cam_w: int, cam_h: int, usbgpu: bool): self.usbgpu = usbgpu self.frame_skip = ModelConstants.MODEL_RUN_FREQ // ModelConstants.MODEL_CONTEXT_FREQ - self.input_queues, self.npy = make_input_queues(self.input_shapes, self.frame_skip, device=self.QUEUE_DEV) + self.input_queues, self.npy = make_input_queues(self.input_shapes, self.frame_skip, device=self.model_device) self.full_frames: dict[str, Tensor] = {} self._blob_cache: dict[tuple[str, int], Tensor] = {} self.parser = Parser() self.frame_buf_params = {k: get_nv12_info(cam_w, cam_h) for k in ('img', 'big_img')} - self.run_policy = jits['run_policy'] - self.warp = jits[(cam_w,cam_h)] + self.run_model = jits['run_model'][(cam_w,cam_h)] def slice_outputs(self, model_outputs: np.ndarray, output_slices: dict[str, slice]) -> dict[str, np.ndarray]: parsed_model_outputs = {k: model_outputs[np.newaxis, v] for k,v in output_slices.items()} @@ -166,11 +164,12 @@ def run(self, bufs: dict[str, VisionBuf], transforms: dict[str, np.ndarray], inputs: dict[str, np.ndarray]) -> dict[str, np.ndarray] | None: for key in bufs.keys(): ptr = np.frombuffer(bufs[key].data, dtype=np.uint8).ctypes.data - yuv_size = self.frame_buf_params[key][3] + stride, y_height, uv_height, _ = self.frame_buf_params[key] + frame_copy_size = nv12_copy_size(stride, y_height, uv_height) # There is a ringbuffer of imgs, just cache tensors pointing to all of them cache_key = (key, ptr) if cache_key not in self._blob_cache: - self._blob_cache[cache_key] = Tensor.from_blob(ptr, (yuv_size,), dtype='uint8', device=self.WARP_DEV) + self._blob_cache[cache_key] = Tensor.from_blob(ptr, (frame_copy_size,), dtype='uint8', device=self.frame_device) self.full_frames[key] = self._blob_cache[cache_key] # Model decides when action is completed, so desire input is just a pulse triggered on rising edge @@ -182,10 +181,9 @@ def run(self, bufs: dict[str, VisionBuf], transforms: dict[str, np.ndarray], self.npy['tfm'][:,:] = transforms['img'][:,:] self.npy['big_tfm'][:,:] = transforms['big_img'][:,:] - warped = self.warp(**{k: self.input_queues[k] for k in WARP_INPUTS}, frame=self.full_frames['img'], big_frame=self.full_frames['big_img']) - - outs, = self.run_policy( - **{k: self.input_queues[k] for k in POLICY_INPUTS if k in self.input_queues}, warped=warped + outs, = self.run_model( + **{k: self.input_queues[k] for k in MODELD_INPUTS}, + frame=self.full_frames['img'], big_frame=self.full_frames['big_img'], ) model_output = outs.numpy()[0] if self.usbgpu and not np.all(np.isfinite(model_output)): @@ -200,11 +198,14 @@ def run(self, bufs: dict[str, VisionBuf], transforms: dict[str, np.ndarray], return outputs_dict def warmup(self) -> None: - dummy_frames = {k: np.zeros(self.frame_buf_params[k][3], dtype=np.uint8) for k in self.vision_input_names} + dummy_frames = { + k: np.zeros(nv12_copy_size(*params[:3]), dtype=np.uint8) + for k, params in self.frame_buf_params.items() + } eye = np.eye(3, dtype=np.float32) dims = {'desire_pulse': ModelConstants.DESIRE_LEN, 'traffic_convention': 2, 'action_t': 2} self.run(dummy_frames, dict.fromkeys(self.vision_input_names, eye), {k: np.zeros(v, dtype=np.float32) for k, v in dims.items()}) - self.input_queues, self.npy = make_input_queues(self.input_shapes, self.frame_skip, device=self.QUEUE_DEV) + self.input_queues, self.npy = make_input_queues(self.input_shapes, self.frame_skip, device=self.model_device) self.prev_desire[:] = 0 self.full_frames.clear() self._blob_cache.clear() diff --git a/tinygrad_repo b/tinygrad_repo index 138fb4a783d82f..07268b724fe63e 160000 --- a/tinygrad_repo +++ b/tinygrad_repo @@ -1 +1 @@ -Subproject commit 138fb4a783d82f4e877ad2fe3692aaf8d1de2e46 +Subproject commit 07268b724fe63e45ba33be193ee679dbf02b163f