def __init_

def init()

in source/neuropod/backends/python_bridge/_neuropod_native_bootstrap/executor.py [0:0]
41 lines of code
8 McCabe index (conditional complexity)

    def __init__(self, neuropod_path):
        """
        Load a python neuropod

        :param  neuropod_path:  The path to a python neuropod package
        """
        self.neuropod_path = neuropod_path

        # Load the model config
        data_path = os.path.join(neuropod_path, "0", "data")
        with open(os.path.join(neuropod_path, "0", "config.json"), "r") as config_file:
            model_config = json.load(config_file)

        # Load entrypoint info from config
        entrypoint_package_path = model_config["entrypoint_package"]
        entrypoint_fn_name = model_config["entrypoint"]

        # This is running from the native code - set up the requirements if any
        # Note: This is only intended to work when using OPE so this can be problematic
        # when running multiple python models in a single process
        # (which you should try to avoid anyway because of the GIL)
        # For other code paths (e.g. when calling `load_neuropod` from python with `_always_use_native=False`),
        # the user is responsible for ensuring that all dependencies are installed in the environment
        lockfile = os.path.join(neuropod_path, "0", "requirements.lock")
        if os.path.isfile(lockfile):
            load_deps(lockfile)

        # Add the custom op paths to the beginning of the python path
        # Note: there currently isn't a good way to handle multiple custom ops with the same name.
        # Depending on the underlying framework, there can be a global registry that the ops register themselves with
        # If multiple neuropods are loaded that contain ops with the same name, this can cause a conflict or silently
        # use an incorrect op.
        #
        # For complete isolation, out of process execution is the best solution
        custom_op_path = os.path.abspath(os.path.join(neuropod_path, "0", "ops"))

        if os.path.isdir(custom_op_path):
            # Try to avoid silently using the incorrect op
            for item in os.listdir(custom_op_path):
                lib_path = os.path.join(custom_op_path, item)
                lib_hash = sha256sum(lib_path)
                if lib_hash in loaded_op_hashes:
                    # We already loaded this op so it's fine if this is added to the path again
                    # because the op is identical
                    # TODO(vip): This might not be entirely true because of transitive dependencies
                    continue

                loaded_op_hashes.add(lib_hash)

                # If we haven't already loaded the op, make sure there isn't something already loadable with the
                # same name as this op
                op_name = os.path.splitext(item)[0]
                try:
                    importlib.import_module(op_name)
                    raise ValueError(
                        (
                            "Package `{}` is importable before loading the neuropod! "
                            "This means that a custom op in your neuropod package clashes with something already "
                            "accessible by python. Please check your PYTHONPATH and try again"
                        ).format(op_name)
                    )
                except ImportError:
                    pass

            # Add the ops directory to the path
            sys.path.insert(0, custom_op_path)

        # Create a symlink to our code directory
        neuropod_code_path = os.path.abspath(os.path.join(neuropod_path, "0", "code"))
        rand_id = str(uuid.uuid4()).replace("-", "_")
        self.symlink_path = os.path.join(SYMLINKS_DIR, rand_id)
        os.symlink(neuropod_code_path, self.symlink_path)

        # Import the entrypoint package
        if six.PY3:
            # We need to clear the import system caches to make sure it can find the new module
            # See https://docs.python.org/3/library/importlib.html#importlib.import_module
            importlib.invalidate_caches()

        entrypoint_package = importlib.import_module(
            "{}.{}".format(rand_id, entrypoint_package_path)
        )

        # Get the entrypoint function and run it with the data path
        self.model = entrypoint_package.__dict__[entrypoint_fn_name](data_path)
def __init__()

def init()
