wrapyfi.utils package

Submodules

wrapyfi.utils.core_utils module

wrapyfi.utils.core_utils.deepcopy(obj: Any, exclude_keys: list | tuple | None = None, shallow_keys: list | tuple | None = None)

Deep copy an object, excluding specified keys.

Parameters:

• obj – Any: The object to deep copy

• exclude_keys – Union[list, tuple]: A list of keys to exclude from the deep copy

• shallow_keys – Union[list, tuple]: A list of keys to shallow copy

wrapyfi.utils.core_utils.get_default_args(fnc: Callable[[...], Any])

Get the default arguments for a function.

Parameters:

fnc – Callable[…, Any]: The function to get the default arguments for

wrapyfi.utils.core_utils.match_args(args: list | tuple, kwargs: dict, src_args: list | tuple, src_kwargs: dict)

Match and Substitute Arguments and Keyword Arguments using Specified Source Values.

Navigate through the provided args and kwargs, identifying entries prefixed with “$” and substituting them with values from src_args and src_kwargs respectively, to dynamically modify the function call parameters using the source values.

Parameters:

• args –

  Union[list, tuple]: A list of arguments, potentially containing strings that indicate substitutable entries. Substitutable entries are prefixed with “$” and followed by either:

  • A digit (indicating an index to reference a value from src_args), or

  • Non-digit characters (indicating a key to reference a value from src_kwargs).

• kwargs – dict: A dictionary of keyword arguments, where values might be strings indicating substitutable entries, similar to the entries in args.

• src_args – Union[list, tuple]: A list of source arguments, intended to be referenced by substitutable entries within args.

• src_kwargs – dict: A dictionary of source keyword arguments, intended to be referenced by substitutable entries within args and kwargs.

Returns:

Tuple[list, dict]: A tuple containing: - list: The new arguments, formed by substituting specified entries from args using src_args and src_kwargs. - dict: The new keyword arguments, formed by substituting specified entries from kwargs using src_args and src_kwargs.

wrapyfi.utils.core_utils.dynamic_module_import(modules: List[str], globals: dict)

Dynamically import modules.

Parameters:

• modules – List[str]: A list of module names to import

• globals – dict: The globals dictionary to update

wrapyfi.utils.core_utils.scan_external(module_paths: str, component: str)

Scan external directories specified in module_paths for a specified component directory and dynamically import the modules.

Args: :param module_paths: str: Colon-separated paths for external directories to scan. :param component: str: Type of module to scan for i.e., “listeners”, “publishers”, “servers”, “clients”, “plugins”

class wrapyfi.utils.core_utils.SingletonOptimized

Bases: type

A singleton metaclass that is thread-safe and optimized for speed.

Source: https://stackoverflow.com/a/6798042

class wrapyfi.utils.core_utils.Plugin

Bases: object

Base class for encoding and decoding plugins.

encode(*args, **kwargs)

Encode data into a base64 string.

Parameters:

• args – tuple: Additional arguments

• kwargs – dict: Additional keyword arguments

Returns:

Tuple[bool, dict]: A tuple containing: - bool: True if the encoding was successful, False otherwise - dict: A dictionary containing:

• ’__wrapyfi__’: A tuple containing the class name and encoded data string

decode(*args, **kwargs)

Decode a base64 string back into data.

Parameters:

• args – tuple: Additional arguments

• kwargs – dict: Additional keyword arguments

Returns:

Tuple[bool, object]: A tuple containing: - bool: True if the decoding was successful, False otherwise - object: The decoded data

class wrapyfi.utils.core_utils.PluginRegistrar

Bases: object

Class for registering encoding and decoding plugins.

encoder_registry = {}

decoder_registry = {'AstropyData': <class 'examples.encoders.plugins.astropy_tables.AstropyData'>, 'MXNetTensor': <class 'wrapyfi.plugins.mxnet_tensor.MXNetTensor'>, 'PILImage': <class 'wrapyfi.plugins.pillow_image.PILImage'>, 'PaddleTensor': <class 'wrapyfi.plugins.paddle_tensor.PaddleTensor'>, 'PintData': <class 'wrapyfi.plugins.pint_quantities.PintData'>, 'PytorchTensor': <class 'wrapyfi.plugins.pytorch_tensor.PytorchTensor'>, 'TensorflowTensor': <class 'wrapyfi.plugins.tensorflow_tensor.TensorflowTensor'>}

static register(types=None)

Register a plugin for encoding and decoding a specific type.

Parameters:

types – tuple: The type(s) to register the plugin for

static scan()

Scan the plugins directory (Wrapyfi builtin and external) for plugins to register. This method is called automatically when the module is imported.

wrapyfi.utils.image_encoders module

class wrapyfi.utils.image_encoders.JpegEncoder(quality: int = 95, encoder: str = 'opencv', logging_level: int = 30)

Bases: object

__init__(quality: int = 95, encoder: str = 'opencv', logging_level: int = 30)

Initialize the JPEG encoder with the specified quality and encoder.

Parameters:

• quality – int: JPEG quality (0-100)

• encoder – str: Encoder to use (opencv, pil, or vips)

• logging_level – int: Logging level for the encoder. Defaults to logging.WARNING

encode_jpg_image(img: numpy.ndarray, return_numpy: bool = False)

Encode an image to JPEG using the specified encoder.

Parameters:

• img – np.ndarray: Input image in BGR format (OpenCV default)

• return_numpy – bool: Whether to return the encoded image as a numpy array. Default is False.

Returns:

img_bytes (bytes or np.ndarray): Encoded JPEG image as bytes or numpy array.

wrapyfi.utils.serialization_encoders module

class wrapyfi.utils.serialization_encoders.JsonEncoder(**kwargs)

Bases: JSONEncoder

A custom JSON encoder that can encode: - Sets - Datetime objects - NumPy datetime64 objects - NumPy ndarray objects - Objects registered with the PluginRegistrar

__init__(**kwargs)

Initialize the JsonEncoder.

Parameters:

kwargs – dict: Additional keyword arguments extracting values from the ‘serializer_kwargs’ key and passing them to the base class. All other keyword arguments are passed to the corresponding Plugin.

find_plugin(obj)

Find the plugin for a given object.

Parameters:

obj – Any: The object to find the plugin for

Returns:

Plugin: The plugin for the given object if its type is registered, None otherwise

encode(obj)

Encode an object into a JSON string and ensure that tuples are not encoded as lists.

Parameters:

obj – Any: The object to encode

Returns:

str: The JSON string representation of the object returned by the base class

default(obj)

The default method for the JSON encoder. This method pre-processes the object before encoding it.

Parameters:

obj – Any: The object to encode

Returns:

dict: A dictionary containing the class name and encoded data string

class wrapyfi.utils.serialization_encoders.JsonDecodeHook(**kwargs)

Bases: object

A custom JSON decoder hook that can decode: - Tuples - Sets - Datetime objects - NumPy datetime64 objects - NumPy ndarray objects - Objects registered with the PluginRegistrar

__init__(**kwargs)

Initialize the JsonDecodeHook.

Parameters:

kwargs – dict: Additional keyword arguments are passed to the corresponding Plugin.

object_hook(obj)

The object hook for the JSON decoder. This method post-processes the object after decoding it.

Parameters:

obj – Any: The object to decode if the object is a dictionary containing the class name and encoded data string

Returns:

Any: The decoded object

Module contents