Source code for wolframclient.serializers.normalizer

# -*- coding: utf-8 -*-

from __future__ import absolute_import, print_function, unicode_literals

from collections import defaultdict

from wolframclient.utils.dispatch import ClassDispatch
from wolframclient.utils.functional import composition, iterate
from wolframclient.utils.importutils import safe_import_string
from wolframclient.utils import six
import inspect
import sys
from wolframclient.utils.api import multiprocessing

dispatch = ClassDispatch()


@dispatch.default()
def normalizer(self, o):
    if not inspect.isclass(o) and hasattr(o, '__iter__'):
        return self.serialize_iterable(self.normalize(value) for value in o)
    if self.allow_external_objects:
        return self.serialize_external_object(o)

    raise NotImplementedError(
        'Cannot serialize object of class %s' % o.__class__)


[docs]class DispatchUpdater(object): def __init__(self, dispatch): self.registry = defaultdict(list) self.modules = set() self.dispatch = dispatch
[docs] def register_modules(self, **handlers): for module, handlers in handlers.items(): self.modules.add(module) self.registry[module].extend(iterate(handlers))
def _update_dispatch(self): if self.modules: for module in self.modules.intersection(sys.modules.keys()): for handler in self.registry[module]: safe_import_string(handler)(self.dispatch) del self.registry[module] self.modules.remove(module) if not six.JYTHON: # global lock to avoid multiple dispatcher updating in multithreaded programs. _lock = multiprocessing.Lock() def update_dispatch(self): with self._lock: self._update_dispatch() else:
[docs] def update_dispatch(self): self._update_dispatch()
updater = DispatchUpdater(dispatch) updater.register_modules( #builtin libraries sys='wolframclient.serializers.normalizer.builtin.update_dispatch', decimal='wolframclient.serializers.normalizer.decimal.update_dispatch', datetime='wolframclient.serializers.normalizer.datetime.update_dispatch', fractions='wolframclient.serializers.normalizer.fractions.update_dispatch', #wolfram language support wolframclient= 'wolframclient.serializers.normalizer.wolfram.update_dispatch', #third party libraries numpy='wolframclient.serializers.normalizer.numpy.update_dispatch', PIL='wolframclient.serializers.normalizer.pil.update_dispatch', )
[docs]class Normalizer(object): default_normalizer = normalizer default_updater = updater def __init__(self, normalizer=None, allow_external_objects=False, target_kernel_version=None): self.normalize = self.chain_normalizers(normalizer) self.allow_external_objects = allow_external_objects self.target_kernel_version = target_kernel_version or 11.3
[docs] def chain_normalizers(self, func): self.default_updater.update_dispatch() return composition(*map(safe_import_string, iterate(func or (), self.default_normalizer)))