How to use the imjoy.workers.utils.dotdict function in imjoy

To help you get started, we’ve selected a few imjoy examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github oeway / ImJoy-Engine / imjoy / workers / python_worker.py View on Github external
def __init__(self, client, opt):
        """Set up connection instance."""
        self.secret = opt.secret
        self.id = opt.id  # pylint: disable=invalid-name
        self.local = {}
        self._remote = dotdict()
        self.interface = {}
        self.plugin_interfaces = {}
        self.remote_set = False
        self.store = ReferenceStore()
        self.executed = False
        self.init = False
        self.abort = threading.Event()
        self.work_dir = opt.work_dir
        self.opt = opt
        self._registered_plugins[self.id] = self
        self.client = client

        def emit(_):
            raise NotImplementedError

        self.emit = emit
github oeway / ImJoy-Engine / imjoy / workers / python_worker.py View on Github external
except Exception as exc:
                    logger.debug("Error in converting: %s", exc)
                    b_object = a_object
                    raise exc
            elif a_object["__jailed_type__"] == "error":
                b_object = Exception(a_object["__value__"])
            elif a_object["__jailed_type__"] == "argument":
                b_object = a_object["__value__"]
            else:
                b_object = a_object["__value__"]
            return b_object

        if isinstance(a_object, tuple):
            a_object = list(a_object)
        isarray = isinstance(a_object, list)
        b_object = [] if isarray else dotdict()
        keys = range(len(a_object)) if isarray else a_object.keys()
        for key in keys:
            if isarray or key in a_object:
                val = a_object[key]
                if isinstance(val, (dict, list)):
                    if isarray:
                        b_object.append(self._decode(val, callback_id, with_promise))
                    else:
                        b_object[key] = self._decode(val, callback_id, with_promise)
        return b_object
github oeway / ImJoy-Engine / imjoy / workers / python_worker.py View on Github external
def set_remote(self, api):
        """Set remote."""
        _remote = dotdict()
        for i, _ in enumerate(api):
            if isinstance(api[i], dict) and "name" in api[i]:
                name = api[i]["name"]
                data = api[i].get("data", None)
                if data is not None:
                    if isinstance(data, dict):
                        data2 = dotdict()
                        for key in data:
                            if key in data:
                                if data[key] == "**@@FUNCTION@@**:" + key:
                                    data2[key] = self._gen_remote_method(
                                        name + "." + key
                                    )
                                else:
                                    data2[key] = data[key]
                        _remote[name] = data2
github oeway / ImJoy-Engine / imjoy / workers / python_worker.py View on Github external
"""Return a ndarray."""
    _dtype = type(typed_array)
    if dtype and dtype != _dtype:
        raise Exception(
            "dtype doesn't match the type of the array: " + _dtype + " != " + dtype
        )
    shape = shape or (len(typed_array),)
    return {
        "__jailed_type__": "ndarray",
        "__value__": typed_array,
        "__shape__": shape,
        "__dtype__": _dtype,
    }


API_UTILS = dotdict(
    ndarray=ndarray, kill=kill, debounce=debounce, set_interval=set_interval
)


class PluginConnection:
    """Represent a plugin connection."""

    # pylint:disable=too-many-instance-attributes
    _registered_plugins = {}

    @staticmethod
    def get_plugin(plugin_id):
        return PluginConnection._registered_plugins.get(plugin_id)

    @staticmethod
    def add_plugin(plugin_id, client_id):
github oeway / ImJoy-Engine / imjoy / workers / python_worker.py View on Github external
def add_plugin(plugin_id, client_id):
        opt = dotdict(id=plugin_id, secret="", work_dir="")
        client = BaseClient.get_client(client_id)
        p = PluginConnection(client, opt)
        return p
github oeway / ImJoy-Engine / imjoy / workers / python_worker.py View on Github external
def set_remote(self, api):
        """Set remote."""
        _remote = dotdict()
        for i, _ in enumerate(api):
            if isinstance(api[i], dict) and "name" in api[i]:
                name = api[i]["name"]
                data = api[i].get("data", None)
                if data is not None:
                    if isinstance(data, dict):
                        data2 = dotdict()
                        for key in data:
                            if key in data:
                                if data[key] == "**@@FUNCTION@@**:" + key:
                                    data2[key] = self._gen_remote_method(
                                        name + "." + key
                                    )
                                else:
                                    data2[key] = data[key]
                        _remote[name] = data2
                    else:
                        _remote[name] = data
                else:
                    _remote[name] = self._gen_remote_method(name)

        self._set_local_api(_remote)
        return _remote