Test yolo onnx model

app.py

@@ -19,6 +19,8 @@ from typing import List, Optional, Tuple
 from PIL import Image
 from transformers import AutoModelForObjectDetection, AutoImageProcessor
 from transformers.image_utils import load_image
+from pipeline import build_pipeline
+from utils import cfg, load_config, load_onnx_model
 
 
 # Configuration constants
@@ -564,4 +566,7 @@ with gr.Blocks(theme=gr.themes.Ocean()) as demo:
     )
 
 if __name__ == "__main__":
+    load_config(cfg, 'configs/yolo8n-bytetrack-cpu')
+    pipeline = build_pipeline(cfg.pipeline)
+    load_onnx_model(pipeline.detector, 'downloads/yolo8n-416.onnx')
     demo.queue(max_size=20).launch()

configs/yolo8n-bytetrack-cpu.yaml

@@ -0,0 +1,14 @@
+# YOLOv8n + ByteTrack Configuration
+pipeline: 
+  detector:
+    model: yolov8n
+    categories: ['LightVehicle', 'Person', 'Building', 'UPole', 'Boat', 'Bike', 'Container', 'Truck', 'Gastank', 'Digger', 'Solarpanels', 'Bus']
+    thresholds:
+      confidence: 0.6
+      iou: 0.4
+    slicing:
+      overlap: 0.2
+    device: cpu
+
+  tracker:
+    algorithm: bytetrack

downloads/yolo8n-416.onnx

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:afbf81f217973b0c9e0b256c0b156d2f08b55169b7ac3f9e2e52004a2449fad1
+size 12162747

pipeline/__init__.py

@@ -0,0 +1,14 @@
+from .pipeline import Pipeline
+from .detectors import build_detector
+from .trackers import build_tracker
+
+def build_pipeline(config):
+    """
+    Build and return a pipeline based on the provided configuration.
+    """
+    # Build detector and tracker using the config
+    detector = build_detector(config.detector)
+    tracker = build_tracker(config.tracker)
+
+    # Create and return a Pipeline object with detector and tracker
+    return Pipeline(detector=detector, tracker=tracker)

pipeline/detectors/__init__.py

@@ -0,0 +1,14 @@
+from .yolo import YOLO
+
+
+def build_detector(config):
+    """
+    Build the detection model based on the provided configuration.
+    """
+    # Initialize the YOLO object detection model
+    return YOLO(config.thresholds.confidence,
+                config.thresholds.iou,
+                config.slicing.overlap,
+                config.categories,
+                config.device)
+    

pipeline/detectors/yolo.py

@@ -0,0 +1,80 @@
+import numpy as np
+import supervision as sv
+from sahi import AutoDetectionModel
+from sahi.predict import get_sliced_prediction
+
+class YOLO:
+    def __init__(self, confidence_threshold, iou_threshold, slicing_overlap, categories, device):
+        """
+        YOLO detector wrapper using SAHI for sliced prediction.
+
+        Args:
+            confidence_threshold (float): Minimum confidence for detections.
+            iou_threshold (float): IoU threshold for NMS (not used directly here).
+            slicing_overlap (float): Overlap ratio for slicing.
+            categories (list): List of class names.
+            device (str): Device to run the model on ('cpu' or 'cuda').
+        """
+        self.model = None
+        self.confidence_threshold = confidence_threshold
+        self.iou_threshold = iou_threshold
+        self.slicing_overlap = slicing_overlap
+        self.categories = categories
+        self.category_mapping = {str(i): category for i, category in enumerate(categories)}
+        self.device = device
+
+    def load_onnx_model(self, path):
+        """
+        Loads the ONNX model using SAHI's AutoDetectionModel.
+        """
+        self.model = AutoDetectionModel.from_pretrained(
+            model_type='yolov8onnx',
+            model_path=path,
+            confidence_threshold=self.confidence_threshold,
+            category_mapping=self.category_mapping,
+            device=self.device
+        )
+
+    def __call__(self, frame):
+        """
+        Runs sliced prediction on the input frame and returns a supervision.Detections object.
+        """
+        # Get input shape from ONNX model
+        input_shape = self.model.model.get_inputs()[0].shape[2]
+        result = get_sliced_prediction(
+            frame,
+            self.model,
+            slice_height=input_shape,
+            slice_width=input_shape,
+            overlap_height_ratio=self.slicing_overlap,
+            overlap_width_ratio=self.slicing_overlap,
+            verbose=False,
+        )
+        boxes = []
+        confidences = []
+        class_ids = []
+        for det in result.object_prediction_list:
+            boxes.append(det.bbox.to_xyxy())
+            confidences.append(det.score.value)
+            class_ids.append(det.category.id)
+        if boxes:
+            boxes = np.array(boxes)
+            confidences = np.array(confidences)
+            class_ids = np.array(class_ids)
+        else:
+            boxes = np.zeros((0, 4))
+            confidences = np.zeros((0,))
+            class_ids = np.zeros((0,))
+        detections = sv.Detections(
+            xyxy=boxes,
+            confidence=confidences,
+            class_id=class_ids,
+        )
+        return detections
+    
+    def get_category_mapping(self):
+        """
+        Returns the category mapping.
+        """
+        # Convert string keys to integers
+        return {int(k): v for k, v in self.category_mapping.items()}

pipeline/pipeline.py

@@ -0,0 +1,26 @@
+class Pipeline:
+    def __init__(self, detector, tracker):
+        """
+        Initialize the Pipeline class with a detector and tracker.
+
+        Args:
+            detector (object): The object detection model.
+            tracker (object): The object tracking model.
+        """
+        self.detector = detector
+        self.tracker = tracker
+
+    def load_state_dict(self, onnx_path):
+        self.detector.load_state_dict(onnx_path)
+
+    def __call__(self, frame):
+        """
+        Run the detection and tracking on the input image.
+
+        Args:
+            frame (np.ndarray): The input image to process.
+
+        Returns:
+            supervision.Detections: Detections object after tracking.
+        """
+        return self.tracker(self.detector(frame))

pipeline/trackers/__init__.py

@@ -0,0 +1,8 @@
+from .bytetrack import ByteTrack
+
+def build_tracker(cfg):
+    """
+    Build the tracking model based on the provided configuration.
+    """
+    # Initialize the tracker
+    return ByteTrack()

pipeline/trackers/bytetrack.py

@@ -0,0 +1,14 @@
+import supervision as sv
+
+
+class ByteTrack:
+    def __init__(self):
+        self.tracker = sv.ByteTrack()
+        self.smoother = sv.DetectionsSmoother()
+
+    def __call__(self, detections):
+        """Process detections using ByteTrack."""
+        # supervision_detections = self._convert_pytorch_to_supervision(detections)
+        tracked_detections = self.tracker.update_with_detections(detections)
+        smoothed_detections = self.smoother.update_with_detections(tracked_detections)
+        return smoothed_detections

requirements.txt

@@ -10,3 +10,5 @@ supervision
 trackers[deepsort] @ git+https://github.com/roboflow/trackers
 spaces
 imageio[pyav]
+onnxruntime
+sahi

utils/__init__.py

@@ -0,0 +1,6 @@
+
+from .config import cfg, load_config
+from .model_loader import load_onnx_model
+
+
+__all__ = ["cfg", "load_config", "load_onnx_model"]

utils/config.py

@@ -0,0 +1,40 @@
+from .yacs import CfgNode
+
+cfg = CfgNode(new_allowed=True)
+cfg.save_dir = "./"
+# common params for NETWORK
+cfg.model = CfgNode(new_allowed=True)
+cfg.model.arch = CfgNode(new_allowed=True)
+cfg.model.arch.backbone = CfgNode(new_allowed=True)
+cfg.model.arch.fpn = CfgNode(new_allowed=True)
+cfg.model.arch.head = CfgNode(new_allowed=True)
+
+# DATASET related params
+cfg.data = CfgNode(new_allowed=True)
+cfg.data.train = CfgNode(new_allowed=True)
+cfg.data.val = CfgNode(new_allowed=True)
+cfg.device = CfgNode(new_allowed=True)
+cfg.device.precision = 32
+# train
+cfg.schedule = CfgNode(new_allowed=True)
+
+# logger
+cfg.log = CfgNode()
+cfg.log.interval = 50
+
+# testing
+cfg.test = CfgNode()
+# size of images for each device
+
+
+def load_config(cfg, args_cfg):
+    cfg.defrost()
+    cfg.merge_from_file(args_cfg)
+    cfg.freeze()
+
+
+if __name__ == "__main__":
+    import sys
+
+    with open(sys.argv[1], "w") as f:
+        print(cfg, file=f)

utils/model_loader.py

@@ -0,0 +1,2 @@
+def load_onnx_model(detector, path):
+    detector.load_onnx_model(path)

utils/yacs.py

@@ -0,0 +1,531 @@
+# Copyright (c) 2018-present, Facebook, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+##############################################################################
+"""YACS -- Yet Another Configuration System is designed to be a simple
+configuration management system for academic and industrial research
+projects.
+
+See README.md for usage and examples.
+"""
+
+import copy
+import io
+import logging
+import os
+import sys
+from ast import literal_eval
+
+import yaml
+
+# Flag for py2 and py3 compatibility to use when separate code paths are necessary
+# When _PY2 is False, we assume Python 3 is in use
+_PY2 = sys.version_info.major == 2
+
+# Filename extensions for loading configs from files
+_YAML_EXTS = {"", ".yaml", ".yml"}
+_PY_EXTS = {".py"}
+
+_FILE_TYPES = (io.IOBase,)
+
+# CfgNodes can only contain a limited set of valid types
+_VALID_TYPES = {tuple, list, str, int, float, bool, type(None)}
+# py2 allow for str and unicode
+if _PY2:
+    _VALID_TYPES = _VALID_TYPES.union({unicode})  # noqa: F821
+
+# Utilities for importing modules from file paths
+if _PY2:
+    # imp is available in both py2 and py3 for now, but is deprecated in py3
+    import imp
+else:
+    import importlib.util
+
+logger = logging.getLogger(__name__)
+
+
+class CfgNode(dict):
+    """
+    CfgNode represents an internal node in the configuration tree. It's a simple
+    dict-like container that allows for attribute-based access to keys.
+    """
+
+    IMMUTABLE = "__immutable__"
+    DEPRECATED_KEYS = "__deprecated_keys__"
+    RENAMED_KEYS = "__renamed_keys__"
+    NEW_ALLOWED = "__new_allowed__"
+
+    def __init__(self, init_dict=None, key_list=None, new_allowed=False):
+        """
+        Args:
+            init_dict (dict): the possibly-nested dictionary to initailize the
+                CfgNode.
+            key_list (list[str]): a list of names which index this CfgNode from
+                the root.
+                Currently only used for logging purposes.
+            new_allowed (bool): whether adding new key is allowed when merging with
+                other configs.
+        """
+        # Recursively convert nested dictionaries in init_dict into CfgNodes
+        init_dict = {} if init_dict is None else init_dict
+        key_list = [] if key_list is None else key_list
+        init_dict = self._create_config_tree_from_dict(init_dict, key_list)
+        super(CfgNode, self).__init__(init_dict)
+        # Manage if the CfgNode is frozen or not
+        self.__dict__[CfgNode.IMMUTABLE] = False
+        # Deprecated options
+        # If an option is removed from the code and you don't want to break existing
+        # yaml configs, you can add the full config key as a string to the set below.
+        self.__dict__[CfgNode.DEPRECATED_KEYS] = set()
+        # Renamed options
+        # If you rename a config option, record the mapping from the old name to the
+        # new name in the dictionary below. Optionally, if the type also changed, you
+        # can make the value a tuple that specifies first the renamed key and then
+        # instructions for how to edit the config file.
+        self.__dict__[CfgNode.RENAMED_KEYS] = {
+            # 'EXAMPLE.OLD.KEY': 'EXAMPLE.NEW.KEY',  # Dummy example to follow
+            # 'EXAMPLE.OLD.KEY': (                   # A more complex example to follow
+            #     'EXAMPLE.NEW.KEY',
+            #     "Also convert to a tuple, e.g., 'foo' -> ('foo',) or "
+            #     + "'foo:bar' -> ('foo', 'bar')"
+            # ),
+        }
+
+        # Allow new attributes after initialisation
+        self.__dict__[CfgNode.NEW_ALLOWED] = new_allowed
+
+    @classmethod
+    def _create_config_tree_from_dict(cls, dic, key_list):
+        """
+        Create a configuration tree using the given dict.
+        Any dict-like objects inside dict will be treated as a new CfgNode.
+
+        Args:
+            dic (dict):
+            key_list (list[str]): a list of names which index this CfgNode from
+                the root. Currently only used for logging purposes.
+        """
+        dic = copy.deepcopy(dic)
+        for k, v in dic.items():
+            if isinstance(v, dict):
+                # Convert dict to CfgNode
+                dic[k] = cls(v, key_list=key_list + [k])
+            else:
+                # Check for valid leaf type or nested CfgNode
+                _assert_with_logging(
+                    _valid_type(v, allow_cfg_node=False),
+                    "Key {} with value {} is not a valid type; valid types: {}".format(
+                        ".".join(key_list + [k]), type(v), _VALID_TYPES
+                    ),
+                )
+        return dic
+
+    def __getattr__(self, name):
+        if name in self:
+            return self[name]
+        else:
+            raise AttributeError(name)
+
+    def __setattr__(self, name, value):
+        if self.is_frozen():
+            raise AttributeError(
+                "Attempted to set {} to {}, but CfgNode is immutable".format(
+                    name, value
+                )
+            )
+
+        _assert_with_logging(
+            name not in self.__dict__,
+            "Invalid attempt to modify internal CfgNode state: {}".format(name),
+        )
+        _assert_with_logging(
+            _valid_type(value, allow_cfg_node=True),
+            "Invalid type {} for key {}; valid types = {}".format(
+                type(value), name, _VALID_TYPES
+            ),
+        )
+
+        self[name] = value
+
+    def __str__(self):
+        def _indent(s_, num_spaces):
+            s = s_.split("\n")
+            if len(s) == 1:
+                return s_
+            first = s.pop(0)
+            s = [(num_spaces * " ") + line for line in s]
+            s = "\n".join(s)
+            s = first + "\n" + s
+            return s
+
+        r = ""
+        s = []
+        for k, v in sorted(self.items()):
+            seperator = "\n" if isinstance(v, CfgNode) else " "
+            attr_str = "{}:{}{}".format(str(k), seperator, str(v))
+            attr_str = _indent(attr_str, 2)
+            s.append(attr_str)
+        r += "\n".join(s)
+        return r
+
+    def __repr__(self):
+        return "{}({})".format(self.__class__.__name__, super(CfgNode, self).__repr__())
+
+    def dump(self, **kwargs):
+        """Dump to a string."""
+
+        def convert_to_dict(cfg_node, key_list):
+            if not isinstance(cfg_node, CfgNode):
+                _assert_with_logging(
+                    _valid_type(cfg_node),
+                    "Key {} with value {} is not a valid type; valid types: {}".format(
+                        ".".join(key_list), type(cfg_node), _VALID_TYPES
+                    ),
+                )
+                return cfg_node
+            else:
+                cfg_dict = dict(cfg_node)
+                for k, v in cfg_dict.items():
+                    cfg_dict[k] = convert_to_dict(v, key_list + [k])
+                return cfg_dict
+
+        self_as_dict = convert_to_dict(self, [])
+        return yaml.safe_dump(self_as_dict, **kwargs)
+
+    def merge_from_file(self, cfg_filename):
+        """Load a yaml config file and merge it this CfgNode."""
+        with open(cfg_filename, "r", encoding="utf-8") as f:
+            cfg = self.load_cfg(f)
+        self.merge_from_other_cfg(cfg)
+
+    def merge_from_other_cfg(self, cfg_other):
+        """Merge `cfg_other` into this CfgNode."""
+        _merge_a_into_b(cfg_other, self, self, [])
+
+    def merge_from_list(self, cfg_list):
+        """Merge config (keys, values) in a list (e.g., from command line) into
+        this CfgNode. For example, `cfg_list = ['FOO.BAR', 0.5]`.
+        """
+        _assert_with_logging(
+            len(cfg_list) % 2 == 0,
+            "Override list has odd length: {}; it must be a list of pairs".format(
+                cfg_list
+            ),
+        )
+        root = self
+        for full_key, v in zip(cfg_list[0::2], cfg_list[1::2]):
+            if root.key_is_deprecated(full_key):
+                continue
+            if root.key_is_renamed(full_key):
+                root.raise_key_rename_error(full_key)
+            key_list = full_key.split(".")
+            d = self
+            for subkey in key_list[:-1]:
+                _assert_with_logging(
+                    subkey in d, "Non-existent key: {}".format(full_key)
+                )
+                d = d[subkey]
+            subkey = key_list[-1]
+            _assert_with_logging(subkey in d, "Non-existent key: {}".format(full_key))
+            value = self._decode_cfg_value(v)
+            value = _check_and_coerce_cfg_value_type(value, d[subkey], subkey, full_key)
+            d[subkey] = value
+
+    def freeze(self):
+        """Make this CfgNode and all of its children immutable."""
+        self._immutable(True)
+
+    def defrost(self):
+        """Make this CfgNode and all of its children mutable."""
+        self._immutable(False)
+
+    def is_frozen(self):
+        """Return mutability."""
+        return self.__dict__[CfgNode.IMMUTABLE]
+
+    def _immutable(self, is_immutable):
+        """Set immutability to is_immutable and recursively apply the setting
+        to all nested CfgNodes.
+        """
+        self.__dict__[CfgNode.IMMUTABLE] = is_immutable
+        # Recursively set immutable state
+        for v in self.__dict__.values():
+            if isinstance(v, CfgNode):
+                v._immutable(is_immutable)
+        for v in self.values():
+            if isinstance(v, CfgNode):
+                v._immutable(is_immutable)
+
+    def clone(self):
+        """Recursively copy this CfgNode."""
+        return copy.deepcopy(self)
+
+    def register_deprecated_key(self, key):
+        """Register key (e.g. `FOO.BAR`) a deprecated option. When merging deprecated
+        keys a warning is generated and the key is ignored.
+        """
+        _assert_with_logging(
+            key not in self.__dict__[CfgNode.DEPRECATED_KEYS],
+            "key {} is already registered as a deprecated key".format(key),
+        )
+        self.__dict__[CfgNode.DEPRECATED_KEYS].add(key)
+
+    def register_renamed_key(self, old_name, new_name, message=None):
+        """Register a key as having been renamed from `old_name` to `new_name`.
+        When merging a renamed key, an exception is thrown alerting to user to
+        the fact that the key has been renamed.
+        """
+        _assert_with_logging(
+            old_name not in self.__dict__[CfgNode.RENAMED_KEYS],
+            "key {} is already registered as a renamed cfg key".format(old_name),
+        )
+        value = new_name
+        if message:
+            value = (new_name, message)
+        self.__dict__[CfgNode.RENAMED_KEYS][old_name] = value
+
+    def key_is_deprecated(self, full_key):
+        """Test if a key is deprecated."""
+        if full_key in self.__dict__[CfgNode.DEPRECATED_KEYS]:
+            logger.warning("Deprecated config key (ignoring): {}".format(full_key))
+            return True
+        return False
+
+    def key_is_renamed(self, full_key):
+        """Test if a key is renamed."""
+        return full_key in self.__dict__[CfgNode.RENAMED_KEYS]
+
+    def raise_key_rename_error(self, full_key):
+        new_key = self.__dict__[CfgNode.RENAMED_KEYS][full_key]
+        if isinstance(new_key, tuple):
+            msg = " Note: " + new_key[1]
+            new_key = new_key[0]
+        else:
+            msg = ""
+        raise KeyError(
+            "Key {} was renamed to {}; please update your config.{}".format(
+                full_key, new_key, msg
+            )
+        )
+
+    def is_new_allowed(self):
+        return self.__dict__[CfgNode.NEW_ALLOWED]
+
+    @classmethod
+    def load_cfg(cls, cfg_file_obj_or_str):
+        """
+        Load a cfg.
+        Args:
+            cfg_file_obj_or_str (str or file):
+                Supports loading from:
+                - A file object backed by a YAML file
+                - A file object backed by a Python source file that exports an attribute
+                  "cfg" that is either a dict or a CfgNode
+                - A string that can be parsed as valid YAML
+        """
+        _assert_with_logging(
+            isinstance(cfg_file_obj_or_str, _FILE_TYPES + (str,)),
+            "Expected first argument to be of type {} or {}, but it was {}".format(
+                _FILE_TYPES, str, type(cfg_file_obj_or_str)
+            ),
+        )
+        if isinstance(cfg_file_obj_or_str, str):
+            return cls._load_cfg_from_yaml_str(cfg_file_obj_or_str)
+        elif isinstance(cfg_file_obj_or_str, _FILE_TYPES):
+            return cls._load_cfg_from_file(cfg_file_obj_or_str)
+        else:
+            raise NotImplementedError("Impossible to reach here (unless there's a bug)")
+
+    @classmethod
+    def _load_cfg_from_file(cls, file_obj):
+        """Load a config from a YAML file or a Python source file."""
+        _, file_extension = os.path.splitext(file_obj.name)
+        if file_extension in _YAML_EXTS:
+            return cls._load_cfg_from_yaml_str(file_obj.read())
+        elif file_extension in _PY_EXTS:
+            return cls._load_cfg_py_source(file_obj.name)
+        else:
+            raise Exception(
+                "Attempt to load from an unsupported file type {}; "
+                "only {} are supported".format(file_obj, _YAML_EXTS.union(_PY_EXTS))
+            )
+
+    @classmethod
+    def _load_cfg_from_yaml_str(cls, str_obj):
+        """Load a config from a YAML string encoding."""
+        cfg_as_dict = yaml.safe_load(str_obj)
+        return cls(cfg_as_dict)
+
+    @classmethod
+    def _load_cfg_py_source(cls, filename):
+        """Load a config from a Python source file."""
+        module = _load_module_from_file("yacs.config.override", filename)
+        _assert_with_logging(
+            hasattr(module, "cfg"),
+            "Python module from file {} must have 'cfg' attr".format(filename),
+        )
+        VALID_ATTR_TYPES = {dict, CfgNode}
+        _assert_with_logging(
+            type(module.cfg) in VALID_ATTR_TYPES,
+            "Imported module 'cfg' attr must be in {} but is {} instead".format(
+                VALID_ATTR_TYPES, type(module.cfg)
+            ),
+        )
+        return cls(module.cfg)
+
+    @classmethod
+    def _decode_cfg_value(cls, value):
+        """
+        Decodes a raw config value (e.g., from a yaml config files or command
+        line argument) into a Python object.
+
+        If the value is a dict, it will be interpreted as a new CfgNode.
+        If the value is a str, it will be evaluated as literals.
+        Otherwise it is returned as-is.
+        """
+        # Configs parsed from raw yaml will contain dictionary keys that need to be
+        # converted to CfgNode objects
+        if isinstance(value, dict):
+            return cls(value)
+        # All remaining processing is only applied to strings
+        if not isinstance(value, str):
+            return value
+        # Try to interpret `value` as a:
+        #   string, number, tuple, list, dict, boolean, or None
+        try:
+            value = literal_eval(value)
+        # The following two excepts allow v to pass through when it represents a
+        # string.
+        #
+        # Longer explanation:
+        # The type of v is always a string (before calling literal_eval), but
+        # sometimes it *represents* a string and other times a data structure, like
+        # a list. In the case that v represents a string, what we got back from the
+        # yaml parser is 'foo' *without quotes* (so, not '"foo"'). literal_eval is
+        # ok with '"foo"', but will raise a ValueError if given 'foo'. In other
+        # cases, like paths (v = 'foo/bar' and not v = '"foo/bar"'), literal_eval
+        # will raise a SyntaxError.
+        except ValueError:
+            pass
+        except SyntaxError:
+            pass
+        return value
+
+
+load_cfg = (
+    CfgNode.load_cfg
+)  # keep this function in global scope for backward compatibility
+
+
+def _valid_type(value, allow_cfg_node=False):
+    return (type(value) in _VALID_TYPES) or (
+        allow_cfg_node and isinstance(value, CfgNode)
+    )
+
+
+def _merge_a_into_b(a, b, root, key_list):
+    """Merge config dictionary a into config dictionary b, clobbering the
+    options in b whenever they are also specified in a.
+    """
+    _assert_with_logging(
+        isinstance(a, CfgNode),
+        "`a` (cur type {}) must be an instance of {}".format(type(a), CfgNode),
+    )
+    _assert_with_logging(
+        isinstance(b, CfgNode),
+        "`b` (cur type {}) must be an instance of {}".format(type(b), CfgNode),
+    )
+
+    for k, v_ in a.items():
+        full_key = ".".join(key_list + [k])
+
+        v = copy.deepcopy(v_)
+        v = b._decode_cfg_value(v)
+
+        if k in b:
+            v = _check_and_coerce_cfg_value_type(v, b[k], k, full_key)
+            # Recursively merge dicts
+            if isinstance(v, CfgNode):
+                try:
+                    _merge_a_into_b(v, b[k], root, key_list + [k])
+                except BaseException:
+                    raise
+            else:
+                b[k] = v
+        elif b.is_new_allowed():
+            b[k] = v
+        else:
+            if root.key_is_deprecated(full_key):
+                continue
+            elif root.key_is_renamed(full_key):
+                root.raise_key_rename_error(full_key)
+            else:
+                raise KeyError("Non-existent config key: {}".format(full_key))
+
+
+def _check_and_coerce_cfg_value_type(replacement, original, key, full_key):
+    """Checks that `replacement`, which is intended to replace `original` is of
+    the right type. The type is correct if it matches exactly or is one of a few
+    cases in which the type can be easily coerced.
+    """
+    original_type = type(original)
+    replacement_type = type(replacement)
+
+    # The types must match (with some exceptions)
+    if replacement_type == original_type:
+        return replacement
+
+    # Cast replacement from from_type to to_type if the replacement and original
+    # types match from_type and to_type
+    def conditional_cast(from_type, to_type):
+        if replacement_type == from_type and original_type == to_type:
+            return True, to_type(replacement)
+        else:
+            return False, None
+
+    # Conditionally casts
+    # list <-> tuple
+    casts = [(tuple, list), (list, tuple)]
+    # For py2: allow converting from str (bytes) to a unicode string
+    try:
+        casts.append((str, unicode))  # noqa: F821
+    except Exception:
+        pass
+
+    for (from_type, to_type) in casts:
+        converted, converted_value = conditional_cast(from_type, to_type)
+        if converted:
+            return converted_value
+
+    raise ValueError(
+        "Type mismatch ({} vs. {}) with values ({} vs. {}) for config "
+        "key: {}".format(
+            original_type, replacement_type, original, replacement, full_key
+        )
+    )
+
+
+def _assert_with_logging(cond, msg):
+    if not cond:
+        logger.debug(msg)
+    assert cond, msg
+
+
+def _load_module_from_file(name, filename):
+    if _PY2:
+        module = imp.load_source(name, filename)
+    else:
+        spec = importlib.util.spec_from_file_location(name, filename)
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+    return module