models: Consolidate model helpers & get_model_path filename fix (#884)

consolidate run_helpers.py & helpers.py
bugfix for get_model_path

sunnypilot/modeld/modeld.py

@@ -23,7 +23,7 @@ from openpilot.sunnypilot.modeld.parse_model_outputs import Parser
 from openpilot.sunnypilot.modeld.fill_model_msg import fill_model_msg, fill_pose_msg, PublishState
 from openpilot.sunnypilot.modeld.constants import ModelConstants
 from openpilot.sunnypilot.modeld.models.commonmodel_pyx import ModelFrame, CLContext
-from openpilot.sunnypilot.modeld.runners.run_helpers import get_model_path, load_metadata, prepare_inputs, load_meta_constants
+from openpilot.sunnypilot.models.helpers import get_model_path, load_metadata, prepare_inputs, load_meta_constants
 
 PROCESS_NAME = "selfdrive.modeld.modeld_snpe"
 SEND_RAW_PRED = os.getenv('SEND_RAW_PRED')

sunnypilot/modeld/runners/run_helpers.py

@@ -1,97 +0,0 @@
-"""
-Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
-
-This file is part of sunnypilot and is licensed under the MIT License.
-See the LICENSE.md file in the root directory for more details.
-"""
-
-import os
-import pickle
-import numpy as np
-
-from cereal import custom
-from openpilot.sunnypilot.modeld.constants import Meta, MetaTombRaider, MetaSimPose
-from openpilot.sunnypilot.modeld.runners import ModelRunner
-from openpilot.sunnypilot.models.helpers import get_active_bundle
-from openpilot.system.hardware import PC
-from openpilot.system.hardware.hw import Paths
-from pathlib import Path
-
-USE_ONNX = os.getenv('USE_ONNX', PC)
-
-CUSTOM_MODEL_PATH = Paths.model_root()
-METADATA_PATH = Path(__file__).parent / '../models/supercombo_metadata.pkl'
-
-ModelManager = custom.ModelManagerSP
-
-
-def _get_model():
-  if bundle := get_active_bundle():
-    drive_model = next(model for model in bundle.models if model.type == ModelManager.Model.Type.supercombo)
-    return drive_model
-
-  return None
-
-def get_model_path():
-  if USE_ONNX:
-    return {ModelRunner.ONNX: Path(__file__).parent / '../models/supercombo.onnx'}
-
-  if model := _get_model():
-    return {ModelRunner.THNEED: f"{CUSTOM_MODEL_PATH}/{model.fileName}"}
-
-  return {ModelRunner.THNEED: Path(__file__).parent / '../models/supercombo.thneed'}
-
-
-def load_metadata():
-  metadata_path = METADATA_PATH
-
-  if model := _get_model():
-    metadata_path = f"{CUSTOM_MODEL_PATH}/{model.metadata.fileName}"
-
-  with open(metadata_path, 'rb') as f:
-    return pickle.load(f)
-
-
-def prepare_inputs(model_metadata) -> dict[str, np.ndarray]:
-  # img buffers are managed in openCL transform code so we don't pass them as inputs
-  inputs = {
-    k: np.zeros(v, dtype=np.float32).flatten()
-    for k, v in model_metadata['input_shapes'].items()
-    if 'img' not in k
-  }
-
-  return inputs
-
-
-def load_meta_constants(model_metadata):
-  """
-  Determines and loads the appropriate meta model class based on the metadata provided. The function checks
-  specific keys and conditions within the provided metadata dictionary to identify the corresponding meta
-  model class to return.
-
-  :param model_metadata: Dictionary containing metadata about the model. It includes
-      details such as input shapes, output slices, and other configurations for identifying
-      metadata-dependent meta model classes.
-  :type model_metadata: dict
-  :return: The appropriate meta model class (Meta, MetaSimPose, or MetaTombRaider)
-      based on the conditions and metadata provided.
-  :rtype: type
-  """
-  meta = Meta  # Default Meta
-
-  if 'sim_pose' in model_metadata['input_shapes'].keys():
-    # Meta for models with sim_pose input
-    meta = MetaSimPose
-  else:
-    # Meta for Tomb Raider, it does not include sim_pose input but has the same meta slice as previous models
-    meta_slice = model_metadata['output_slices']['meta']
-    meta_tf_slice = slice(5868, 5921, None)
-
-    if (
-          meta_slice.start == meta_tf_slice.start and
-          meta_slice.stop == meta_tf_slice.stop and
-          meta_slice.step == meta_tf_slice.step
-    ):
-      meta = MetaTombRaider
-
-  return meta

sunnypilot/models/helpers.py

@@ -7,13 +7,28 @@ See the LICENSE.md file in the root directory for more details.
 
 import hashlib
 import os
+import pickle
+import numpy as np
+import json
+
 from openpilot.common.params import Params
 from cereal import custom
-import json
+from openpilot.sunnypilot.modeld.constants import Meta, MetaTombRaider, MetaSimPose
+from openpilot.sunnypilot.modeld.runners import ModelRunner
+from openpilot.system.hardware import PC
+from openpilot.system.hardware.hw import Paths
+from pathlib import Path
 
 CURRENT_SELECTOR_VERSION = 2
 REQUIRED_MIN_SELECTOR_VERSION = 2
 
+USE_ONNX = os.getenv('USE_ONNX', PC)
+
+CUSTOM_MODEL_PATH = Paths.model_root()
+METADATA_PATH = Path(__file__).parent / '../models/supercombo_metadata.pkl'
+
+ModelManager = custom.ModelManagerSP
+
 
 async def verify_file(file_path: str, expected_hash: str) -> bool:
   """Verifies file hash against expected hash"""
@@ -98,3 +113,74 @@ def get_active_model_runner(params: Params = None, force_check=False) -> custom.
     params.put("ModelRunnerTypeCache", str(int(runner_type)))
 
   return runner_type
+
+def _get_model():
+  if bundle := get_active_bundle():
+    drive_model = next(model for model in bundle.models if model.type == ModelManager.Model.Type.supercombo)
+    return drive_model
+
+  return None
+
+def get_model_path():
+  if USE_ONNX:
+    return {ModelRunner.ONNX: Path(__file__).parent / '../models/supercombo.onnx'}
+
+  if model := _get_model():
+    return {ModelRunner.THNEED: f"{CUSTOM_MODEL_PATH}/{model.artifact.fileName}"}
+
+  return {ModelRunner.THNEED: Path(__file__).parent / '../models/supercombo.thneed'}
+
+
+def load_metadata():
+  metadata_path = METADATA_PATH
+
+  if model := _get_model():
+    metadata_path = f"{CUSTOM_MODEL_PATH}/{model.metadata.fileName}"
+
+  with open(metadata_path, 'rb') as f:
+    return pickle.load(f)
+
+
+def prepare_inputs(model_metadata) -> dict[str, np.ndarray]:
+  # img buffers are managed in openCL transform code so we don't pass them as inputs
+  inputs = {
+    k: np.zeros(v, dtype=np.float32).flatten()
+    for k, v in model_metadata['input_shapes'].items()
+    if 'img' not in k
+  }
+
+  return inputs
+
+
+def load_meta_constants(model_metadata):
+  """
+  Determines and loads the appropriate meta model class based on the metadata provided. The function checks
+  specific keys and conditions within the provided metadata dictionary to identify the corresponding meta
+  model class to return.
+
+  :param model_metadata: Dictionary containing metadata about the model. It includes
+      details such as input shapes, output slices, and other configurations for identifying
+      metadata-dependent meta model classes.
+  :type model_metadata: dict
+  :return: The appropriate meta model class (Meta, MetaSimPose, or MetaTombRaider)
+      based on the conditions and metadata provided.
+  :rtype: type
+  """
+  meta = Meta  # Default Meta
+
+  if 'sim_pose' in model_metadata['input_shapes'].keys():
+    # Meta for models with sim_pose input
+    meta = MetaSimPose
+  else:
+    # Meta for Tomb Raider, it does not include sim_pose input but has the same meta slice as previous models
+    meta_slice = model_metadata['output_slices']['meta']
+    meta_tf_slice = slice(5868, 5921, None)
+
+    if (
+            meta_slice.start == meta_tf_slice.start and
+            meta_slice.stop == meta_tf_slice.stop and
+            meta_slice.step == meta_tf_slice.step
+    ):
+      meta = MetaTombRaider
+
+  return meta