Source code for imagepypelines

# @Email: jmaggio14@gmail.com
# @Website: https://www.imagepypelines.org/
# @License: https://github.com/jmaggio14/imagepypelines/blob/master/LICENSE
# @github: https://github.com/jmaggio14/imagepypelines
#
# Copyright (c) 2018 - 2020 Jeff Maggio, Jai Mehra, Ryan Hartzell

# -------- setup a uuid for this imagepypelines session -------
import time
from uuid import uuid4
import os
import pkg_resources
import sys

init_time = time.time()
"""unix time initiatization time for this imagepypelines session"""
session_uuid = uuid4().hex
"""a universally unique id for this imagepypelines session"""

# ----------- Setup the Root ImagePypelines Logger ---------------
# import the master logger
from .Logger import MASTER_LOGGER, get_logger, ImagepypelinesLogger, set_log_level, get_master_logger
# import master logger convienence function
# NOTE: import logging constants our users can modify to change color behavior


# ---------- import imagepypelines ----------
from .version_info import *
from .core import *

# ---------- import plugins ----------
from collections import OrderedDict
LOADED_PLUGINS = OrderedDict()
"""module level OrderedDict that contains the all loaded modules in the order in
which they were loaded"""


[docs]def add_plugin(plugin_name, plugin_module, add_to_namespace=True): """adds the given plugin to ImagePypelines Args: plugin_name(str): the name of the desired plugin namespace plugin_module(module): the valid module object for the plugin add_to_namespace(bool): whether or not to add the plugin to the ImagePypelines namespace so it's available as `ip.plugin_name`. If False, then you will only be access your plugin namespace using `ip.get_plugin_by_name`. Defaults to True. Returns: None """ import sys ip_module = sys.modules[__name__] get_master_logger().debug( "loading plugin '{0}' - it will be available as imagepypelines.{0}"\ .format(plugin_name)) if add_to_namespace: # add the plugin to the current namespace setattr(ip_module, plugin_name, plugin_module) # update the default shape functions if the plugin provides new ones SHAPE_FUNCS.update( getattr(plugin_module, 'SHAPE_FUNCS', {}) ) # update the default homogenus containers if the plugin provides new ones HOMOGENUS_CONTAINERS.extend( getattr(plugin_module, 'HOMOGENUS_CONTAINERS', []) ) # add the plugin name to a global list for debugging LOADED_PLUGINS[plugin_name] = plugin_module
# define a function to load all the plugins so it's easier to keep the namespace # clean
[docs]def load_plugins(): """Load all installed plugins to the imagepypelines namespace""" # import these again in case this function is called multiple times and it's # deleted from the global namespace import pkg_resources # load in all installed python packages with our plugin entry_point required_objects = [] plugins = { entry_point.name: entry_point.load() for entry_point in pkg_resources.iter_entry_points('imagepypelines.plugins') } for plugin_name in sorted( plugins.keys() ): plugin_module = plugins[plugin_name] add_plugin(plugin_name, plugin_module, True)
# load all of our plugins load_plugins() # define a function to check if a plugin is loaded
[docs]def require(*plugins): """check to make sure the given plugin(s) are loaded and raise an error if they can't be found Args: *plugins: names of plugins """ master = get_master_logger() not_loaded = [] for plg in plugins: if not (plg in LOADED_PLUGINS): not_loaded.append(plg) if not_loaded: for plg in not_loaded: master.error(f'unable to find "{plg}"') raise RuntimeError(f"unable to find the required plugin(s): {not_loaded}")
[docs]def get_plugin_by_name(plugin_name): """fetches the plugin module using its name""" require(plugin_name) return LOADED_PLUGINS[plugin_name]
# ---------- delete namespace pollutants ---------- del pkg_resources, os, uuid4, time, OrderedDict, sys