Source code for param.reactive

"""
reactive API

`rx` is a wrapper around a Python object that lets users create
reactive expression pipelines by calling existing APIs on an object with dynamic
parameters or widgets.

An `rx` instance watches what operations are applied to the object
and records these on each instance, which are then strung together
into a chain.

The original input to an `rx` object is stored in a mutable list and can be
accessed via the `_obj` property. The shared mutable data structure
ensures that all `rx` instances created from the same object can
hold a shared reference that can be updated, e.g. via the `.value`
property or because the input was itself a reference to some object that
can potentially be updated.

When an operation is applied to an `rx` instance, it will
record the operation and create a new instance using the `_clone` method,
e.g. `dfi.head()` first records that the `'head'` attribute is
accessed, which is achieved by overriding `__getattribute__`. A new
reactive object is returned, which will then record that it is
being called, and that new object will be itself called, as
`rx` implements `__call__`. `__call__` returns another
`rx` instance. To be able to watch all the potential
operations that may be applied to an object, `rx` implements:

- `__getattribute__`: Watching for attribute accesses
- `__call__`: Intercepting both actual calls or method calls if an
  attribute was previously accessed
- `__getitem__`: Intercepting indexing operations
- Operators: Implementing all valid operators `__gt__`, `__add__`, etc.
- `__array_ufunc__`: Intercepting numpy universal function calls

The `rx` object evaluates operations lazily, but whenever the
current value is needed the operations are automatically
evaluated. Note that even attribute access or tab-completion
operations can result in evaluation of the pipeline. This is very
useful in a REPL, as this allows inspecting the transformed
object at any point of the pipeline, and as such provide correct
auto-completion and docstrings. E.g. executing `dfi.A.max?` in an
interactive REPL or notebook where it allows returning the docstring
of the method being accessed.

The actual operations are stored as a dictionary on the `_operation`
attribute of each instance. They contain 4 keys:

- `fn`: The function to apply (either an actual function or a string
        indicating the operation is a method on the object)
- `args`: Any arguments to supply to the `fn`.
- `kwargs`: Any keyword arguments to supply to the `fn`.
- `reverse`: If the function is not a method this indicates whether
             the first arg and the input object should be supplied in
             reverse order.

The `_depth` attribute starts at 0 and is incremented by 1 every time
a new `rx` instance is created part of a chain. The root
instance in a reactive expression  has a `_depth` of 0. A reactive
expression can consist of multiple chains, such as `dfi[dfi.A > 1]`,
as the `rx` instance is referenced twice in the expression. As a
consequence `_depth` is not the total count of `rx` instance
creations of a pipeline, it is the count of instances created in the
outer chain. In the example, that would be `dfi[]`. Each `rx`
instance keeps a reference to the previous instance in the chain and
each instance tracks whether its current value is up-to-date via the
`_dirty` attribute, which is set to False if any dependency changes.

The `_method` attribute is a string that temporarily stores the
method/attr accessed on the object, e.g. `_method` is 'head' in
`dfi.head()`, until the `rx` instance created in the pipeline
is called at which point `_method` is reset to None. In cases such as
`dfi.head` or `dfi.A`, `_method` is not (yet) reset to None. At this
stage the `rx` instance returned has its `_current` attribute
not updated, e.g. `dfi.A._current` is still the original dataframe,
not the 'A' series. Keeping `_method` is thus useful for instance to
display `dfi.A`, as the evaluation of the object will check whether
`_method` is set or not, and if it's set it will use it to compute the
object returned, e.g. the series `df.A` or the method `df.head`, and
display its repr.
"""
from __future__ import annotations

import asyncio
import inspect
import math
import operator

from collections.abc import Iterable, Iterator
from functools import partial
from types import FunctionType, MethodType
from typing import Any, Callable, Optional

from .depends import depends
from .display import _display_accessors, _reactive_display_objs
from .parameterized import (
    Parameter, Parameterized, Skip, Undefined, eval_function_with_deps, get_method_owner,
    register_reference_transform, resolve_ref, resolve_value, transform_reference
)
from .parameters import Boolean, Event
from ._utils import _to_async_gen, iscoroutinefunction, full_groupby


class Wrapper(Parameterized):
    """
    Helper class to allow updating literal values easily.
    """

    object = Parameter(allow_refs=False)


class GenWrapper(Parameterized):
    """
    Helper class to allow streaming from generator functions.
    """

    object = Parameter(allow_refs=True)


class Trigger(Parameterized):
    """
    Helper class to allow triggering an event under some condition.
    """

    value = Event()

    def __init__(self, parameters=None, internal=False, **params):
        super().__init__(**params)
        self.internal = internal
        self.parameters = parameters

class Resolver(Parameterized):
    """
    Helper class to allow (recursively) resolving references.
    """

    object = Parameter(allow_refs=True)

    recursive = Boolean(default=False)

    value = Parameter()

    def __init__(self, **params):
        self._watchers = []
        super().__init__(**params)

    def _resolve_value(self, *events):
        nested = self.param.object.nested_refs
        refs = resolve_ref(self.object, nested)
        value = resolve_value(self.object, nested)
        if self.recursive:
            new_refs = [r for r in resolve_ref(value, nested) if r not in refs]
            while new_refs:
                refs += new_refs
                value = resolve_value(value, nested)
                new_refs = [r for r in resolve_ref(value, nested) if r not in refs]
            if events:
                self._update_refs(refs)
        self.value = value
        return refs

    @depends('object', watch=True, on_init=True)
    def _resolve_object(self):
        refs = self._resolve_value()
        self._update_refs(refs)

    def _update_refs(self, refs):
        for w in self._watchers:
            (w.inst or w.cls).param.unwatch(w)
        self._watchers = []
        for _, params in full_groupby(refs, lambda x: id(x.owner)):
            self._watchers.append(
                params[0].owner.param.watch(self._resolve_value, [p.name for p in params])
            )


class NestedResolver(Resolver):

    object = Parameter(allow_refs=True, nested_refs=True)


class reactive_ops:
    """
    Namespace for reactive operators.

    Implements operators that cannot be implemented using regular
    Python syntax.
    """

    def __init__(self, reactive):
        self._reactive = reactive

    def _as_rx(self):
        return self._reactive if isinstance(self._reactive, rx) else self()

    def __call__(self):
        rxi = self._reactive
        return rxi if isinstance(rx, rx) else rx(rxi)

    def and_(self, other):
        """
        Replacement for the ``and`` statement.
        """
        return self._as_rx()._apply_operator(lambda obj, other: obj and other, other)

[docs] def bool(self): """ __bool__ cannot be implemented so it is provided as a method. """ return self._as_rx()._apply_operator(bool)
def buffer(self, n): """ Collects the last n items that were emitted. """ items = [] def collect(new, n): items.append(new) while len(items) > n: items.pop(0) return items return self._as_rx()._apply_operator(collect, n)
[docs] def in_(self, other): """ Replacement for the ``in`` statement. """ return self._as_rx()._apply_operator(operator.contains, other, reverse=True)
[docs] def is_(self, other): """ Replacement for the ``is`` statement. """ return self._as_rx()._apply_operator(operator.is_, other)
[docs] def is_not(self, other): """ Replacement for the ``is not`` statement. """ return self._as_rx()._apply_operator(operator.is_not, other)
[docs] def len(self): """ __len__ cannot be implemented so it is provided as a method. """ return self._as_rx()._apply_operator(len)
def map(self, func, /, *args, **kwargs): """ Apply a function to each item. Arguments --------- func: function Function to apply. args: iterable, optional Positional arguments to pass to `func`. kwargs: mapping, optional A dictionary of keywords to pass to `func`. """ if inspect.isasyncgenfunction(func) or inspect.isgeneratorfunction(func): raise TypeError( "Cannot map a generator function. Only regular function " "or coroutine functions are permitted." ) if inspect.iscoroutinefunction(func): async def apply(vs, *args, **kwargs): return list(await asyncio.gather(*(func(v, *args, **kwargs) for v in vs))) else: def apply(vs, *args, **kwargs): return [func(v, *args, **kwargs) for v in vs] return self._as_rx()._apply_operator(apply, *args, **kwargs) def not_(self): """ __bool__ cannot be implemented so not has to be provided as a method. """ return self._as_rx()._apply_operator(operator.not_) def or_(self, other): """ Replacement for the ``or`` statement. """ return self._as_rx()._apply_operator(lambda obj, other: obj or other, other)
[docs] def pipe(self, func, /, *args, **kwargs): """ Apply chainable functions. Arguments --------- func: function Function to apply. args: iterable, optional Positional arguments to pass to `func`. kwargs: mapping, optional A dictionary of keywords to pass to `func`. """ return self._as_rx()._apply_operator(func, *args, **kwargs)
def resolve(self, nested=True, recursive=False): """ Resolves references held by the expression. As an example if the expression returns a list of parameters this operation will return a list of the parameter values. Arguments --------- nested: bool Whether to resolve references contained within nested objects, i.e. tuples, lists, sets and dictionaries. recursive: bool Whether to recursively resolve references, i.e. if a reference itself returns a reference we recurse into it until no more references can be resolved. """ resolver_type = NestedResolver if nested else Resolver resolver = resolver_type(object=self._reactive, recursive=recursive) return resolver.param.value.rx()
[docs] def updating(self): """ Returns a new expression that is True while the expression is updating. """ wrapper = Wrapper(object=False) self._watch(lambda e: wrapper.param.update(object=True), precedence=-999) self._watch(lambda e: wrapper.param.update(object=False), precedence=999) return wrapper.param.object.rx()
[docs] def when(self, *dependencies, initial=Undefined): """ Returns a reactive expression that emits the contents of this expression only when the dependencies change. If initial value is provided and the dependencies are all param.Event types the expression will not be evaluated until the first event is triggered. Arguments --------- dependencies: param.Parameter | rx A dependency that will trigger an update in the output. initial: object Object that will stand in for the actual value until the first time a param.Event in the dependencies is triggered. """ deps = [p for d in dependencies for p in resolve_ref(d)] is_event = all(isinstance(dep, Event) for dep in deps) def eval(*_, evaluated=[]): if is_event and initial is not Undefined and not evaluated: # Abuse mutable default value to keep track of evaluation state evaluated.append(True) return initial else: return self.value return bind(eval, *deps).rx()
[docs] def where(self, x, y): """ Returns either x or y depending on the current state of the expression, i.e. replaces a ternary if statement. Arguments --------- x: object The value to return if the expression evaluates to True. y: object The value to return if the expression evaluates to False. """ xrefs = resolve_ref(x) yrefs = resolve_ref(y) if isinstance(self._reactive, rx): params = self._reactive._params else: params = resolve_ref(self._reactive) trigger = Trigger(parameters=params) if xrefs: def trigger_x(*args): if self.value: trigger.param.trigger('value') bind(trigger_x, *xrefs, watch=True) if yrefs: def trigger_y(*args): if not self.value: trigger.param.trigger('value') bind(trigger_y, *yrefs, watch=True) def ternary(condition, _): return resolve_value(x) if condition else resolve_value(y) return bind(ternary, self._reactive, trigger.param.value)
# Operations to get the output and set the input of an expression @property def value(self): """ Returns the current state of the reactive expression by evaluating the pipeline. """ if isinstance(self._reactive, rx): return self._reactive._resolve() elif isinstance(self._reactive, Parameter): return getattr(self._reactive.owner, self._reactive.name) else: return self._reactive() @value.setter def value(self, new): """ Allows overriding the original input to the pipeline. """ if isinstance(self._reactive, Parameter): raise AttributeError( "`Parameter.rx.value = value` is not supported. Cannot override " "parameter value." ) elif not isinstance(self._reactive, rx): raise AttributeError( "`bind(...).rx.value = value` is not supported. Cannot override " "the output of a function." ) elif self._reactive._root is not self._reactive: raise AttributeError( "The value of a derived expression cannot be set. Ensure you " "set the value on the root node wrapping a concrete value, e.g.:" "\n\n a = rx(1)\n b = a + 1\n a.rx.value = 2\n\n " "is valid but you may not set `b.rx.value = 2`." ) if self._reactive._wrapper is None: raise AttributeError( "Setting the value of a reactive expression is only " "supported if it wraps a concrete value. A reactive " "expression wrapping a Parameter or another dynamic " "reference cannot be updated." ) self._reactive._wrapper.object = resolve_value(new)
[docs] def watch(self, fn=None, onlychanged=True, queued=False, precedence=0): """ Adds a callable that observes the output of the pipeline. If no callable is provided this simply causes the expression to be eagerly evaluated. """ if precedence < 0: raise ValueError("User-defined watch callbacks must declare " "a positive precedence. Negative precedences " "are reserved for internal Watchers.") self._watch(fn, onlychanged=onlychanged, queued=queued, precedence=precedence)
def _watch(self, fn=None, onlychanged=True, queued=False, precedence=0): def cb(value): from .parameterized import async_executor if iscoroutinefunction(fn): async_executor(partial(fn, value)) elif fn is not None: fn(value) bind(cb, self._reactive, watch=True)
[docs]def bind(function, *args, watch=False, **kwargs): """ Given a function, returns a wrapper function that binds the values of some or all arguments to Parameter values and expresses Param dependencies on those values, so that the function can be invoked whenever the underlying values change and the output will reflect those updated values. As for functools.partial, arguments can also be bound to constants, which allows all of the arguments to be bound, leaving a simple callable object. Arguments --------- function: callable The function to bind constant or dynamic args and kwargs to. args: object, param.Parameter Positional arguments to bind to the function. watch: boolean Whether to evaluate the function automatically whenever one of the bound parameters changes. kwargs: object, param.Parameter Keyword arguments to bind to the function. Returns ------- Returns a new function with the args and kwargs bound to it and annotated with all dependencies. """ args, kwargs = ( tuple(transform_reference(arg) for arg in args), {key: transform_reference(arg) for key, arg in kwargs.items()} ) dependencies = {} # If the wrapped function has a dependency add it fn_dep = transform_reference(function) if isinstance(fn_dep, Parameter) or hasattr(fn_dep, '_dinfo'): dependencies['__fn'] = fn_dep # Extract dependencies from args and kwargs for i, p in enumerate(args): if hasattr(p, '_dinfo'): for j, arg in enumerate(p._dinfo['dependencies']): dependencies[f'__arg{i}_arg{j}'] = arg for kw, kwarg in p._dinfo['kw'].items(): dependencies[f'__arg{i}_arg_{kw}'] = kwarg elif isinstance(p, Parameter): dependencies[f'__arg{i}'] = p for kw, v in kwargs.items(): if hasattr(v, '_dinfo'): for j, arg in enumerate(v._dinfo['dependencies']): dependencies[f'__kwarg_{kw}_arg{j}'] = arg for pkw, kwarg in v._dinfo['kw'].items(): dependencies[f'__kwarg_{kw}_{pkw}'] = kwarg elif isinstance(v, Parameter): dependencies[kw] = v def combine_arguments(wargs, wkwargs, asynchronous=False): combined_args = [] for arg in args: if hasattr(arg, '_dinfo'): arg = eval_function_with_deps(arg) elif isinstance(arg, Parameter): arg = getattr(arg.owner, arg.name) combined_args.append(arg) combined_args += list(wargs) combined_kwargs = {} for kw, arg in kwargs.items(): if hasattr(arg, '_dinfo'): arg = eval_function_with_deps(arg) elif isinstance(arg, Parameter): arg = getattr(arg.owner, arg.name) combined_kwargs[kw] = arg for kw, arg in wkwargs.items(): if asynchronous: if kw.startswith('__arg'): index = kw[5:] if index.isdigit(): combined_args[int(index)] = arg elif kw.startswith('__kwarg'): substring = kw[8:] if substring in combined_kwargs: combined_kwargs[substring] = arg continue elif kw.startswith('__arg') or kw.startswith('__kwarg') or kw.startswith('__fn'): continue combined_kwargs[kw] = arg return combined_args, combined_kwargs def eval_fn(): if callable(function): fn = function else: p = transform_reference(function) if isinstance(p, Parameter): fn = getattr(p.owner, p.name) else: fn = eval_function_with_deps(p) return fn if inspect.isgeneratorfunction(function): def wrapped(*wargs, **wkwargs): combined_args, combined_kwargs = combine_arguments( wargs, wkwargs, asynchronous=True ) evaled = eval_fn()(*combined_args, **combined_kwargs) for val in evaled: yield val wrapper_fn = depends(**dependencies, watch=watch)(wrapped) wrapped._dinfo = wrapper_fn._dinfo elif inspect.isasyncgenfunction(function): async def wrapped(*wargs, **wkwargs): combined_args, combined_kwargs = combine_arguments( wargs, wkwargs, asynchronous=True ) evaled = eval_fn()(*combined_args, **combined_kwargs) async for val in evaled: yield val wrapper_fn = depends(**dependencies, watch=watch)(wrapped) wrapped._dinfo = wrapper_fn._dinfo elif iscoroutinefunction(function): @depends(**dependencies, watch=watch) async def wrapped(*wargs, **wkwargs): combined_args, combined_kwargs = combine_arguments( wargs, wkwargs, asynchronous=True ) evaled = eval_fn()(*combined_args, **combined_kwargs) return await evaled else: @depends(**dependencies, watch=watch) def wrapped(*wargs, **wkwargs): combined_args, combined_kwargs = combine_arguments(wargs, wkwargs) return eval_fn()(*combined_args, **combined_kwargs) wrapped.__bound_function__ = function wrapped.rx = reactive_ops(wrapped) _reactive_display_objs.add(wrapped) for name, accessor in _display_accessors.items(): setattr(wrapped, name, accessor(wrapped)) return wrapped
[docs]class rx: """ `rx` allows wrapping objects and then operating on them interactively while recording any operations applied to them. By recording all arguments or operands in the operations the recorded pipeline can be replayed if an operand represents a dynamic value. Parameters ---------- obj: any A supported data structure object Examples -------- Instantiate it from an object: >>> ifloat = rx(3.14) >>> ifloat * 2 6.28 Then update the original value and see the new result: >>> ifloat.value = 1 2 """ _accessors: dict[str, Callable[[rx], Any]] = {} _display_options: tuple[str] = () _display_handlers: dict[type, tuple[Any, dict[str, Any]]] = {} _method_handlers: dict[str, Callable] = {} @classmethod def register_accessor( cls, name: str, accessor: Callable[[rx], Any], predicate: Optional[Callable[[Any], bool]] = None ): """ Registers an accessor that extends rx with custom behavior. Arguments --------- name: str The name of the accessor will be attribute-accessible under. accessor: Callable[[rx], any] A callable that will return the accessor namespace object given the rx object it is registered on. predicate: Callable[[Any], bool] | None """ cls._accessors[name] = (accessor, predicate) @classmethod def register_display_handler(cls, obj_type, handler, **kwargs): """ Registers a display handler for a specific type of object, making it possible to define custom display options for specific objects. Arguments --------- obj_type: type | callable The type to register a custom display handler on. handler: Viewable | callable A Viewable or callable that is given the object to be displayed and the custom keyword arguments. kwargs: dict[str, Any] Additional display options to register for this type. """ cls._display_handlers[obj_type] = (handler, kwargs) @classmethod def register_method_handler(cls, method, handler): """ Registers a handler that is called when a specific method on an object is called. """ cls._method_handlers[method] = handler def __new__(cls, obj=None, **kwargs): wrapper = None obj = transform_reference(obj) if kwargs.get('fn'): # rx._clone codepath fn = kwargs.pop('fn') wrapper = kwargs.pop('_wrapper', None) elif inspect.isgeneratorfunction(obj) or iscoroutinefunction(obj): # Resolves generator and coroutine functions lazily wrapper = GenWrapper(object=obj) fn = bind(lambda obj: obj, wrapper.param.object) obj = Undefined elif isinstance(obj, (FunctionType, MethodType)) and hasattr(obj, '_dinfo'): # Bound functions and methods are resolved on access fn = obj obj = None elif isinstance(obj, Parameter): fn = bind(lambda obj: obj, obj) obj = getattr(obj.owner, obj.name) else: # For all other objects wrap them so they can be updated # via .rx.value property wrapper = Wrapper(object=obj) fn = bind(lambda obj: obj, wrapper.param.object) inst = super(rx, cls).__new__(cls) inst._fn = fn inst._shared_obj = kwargs.get('_shared_obj', None if obj is None else [obj]) inst._wrapper = wrapper return inst
[docs] def __init__( self, obj=None, operation=None, fn=None, depth=0, method=None, prev=None, _shared_obj=None, _current=None, _wrapper=None, **kwargs ): # _init is used to prevent to __getattribute__ to execute its # specialized code. self._init = False display_opts = {} for _, opts in self._display_handlers.values(): for k, o in opts.items(): display_opts[k] = o display_opts.update({ dopt: kwargs.pop(dopt) for dopt in self._display_options + tuple(display_opts) if dopt in kwargs }) self._display_opts = display_opts self._method = method self._operation = operation self._depth = depth self._dirty = _current is None self._dirty_obj = False self._current_task = None self._error_state = None self._current_ = _current if isinstance(obj, rx) and not prev: self._prev = obj else: self._prev = prev # Define special trigger parameter if operation has to be lazily evaluated if operation and (iscoroutinefunction(operation['fn']) or inspect.isgeneratorfunction(operation['fn'])): self._trigger = Trigger(internal=True) self._current_ = Undefined else: self._trigger = None self._root = self._compute_root() self._fn_params = self._compute_fn_params() self._internal_params = self._compute_params() # Filter params that external objects depend on, ensuring # that Trigger parameters do not cause double execution self._params = [ p for p in self._internal_params if (not isinstance(p.owner, Trigger) or p.owner.internal) or any (p not in self._internal_params for p in p.owner.parameters) ] self._setup_invalidations(depth) self._kwargs = kwargs self.rx = reactive_ops(self) self._init = True for name, accessor in _display_accessors.items(): setattr(self, name, accessor(self)) for name, (accessor, predicate) in rx._accessors.items(): if predicate is None or predicate(self._current): setattr(self, name, accessor(self))
@property def _obj(self): if self._shared_obj is None: self._obj = eval_function_with_deps(self._fn) elif self._root._dirty_obj: root = self._root root._shared_obj[0] = eval_function_with_deps(root._fn) root._dirty_obj = False return self._shared_obj[0] @_obj.setter def _obj(self, obj): if self._shared_obj is None: self._shared_obj = [obj] else: self._shared_obj[0] = obj @property def _current(self): if self._error_state: raise self._error_state elif self._dirty or self._root._dirty_obj: self._resolve() return self._current_ def _compute_root(self): if self._prev is None: return self root = self while root._prev is not None: root = root._prev return root def _compute_fn_params(self) -> list[Parameter]: if self._fn is None: return [] owner = get_method_owner(self._fn) if owner is not None: deps = [ dep.pobj for dep in owner.param.method_dependencies(self._fn.__name__) ] return deps dinfo = getattr(self._fn, '_dinfo', {}) args = list(dinfo.get('dependencies', [])) kwargs = list(dinfo.get('kw', {}).values()) return args + kwargs def _compute_params(self) -> list[Parameter]: ps = list(self._fn_params) if self._trigger: ps.append(self._trigger.param.value) # Collect parameters on previous objects in chain prev = self._prev while prev is not None: for p in prev._params: if p not in ps: ps.append(p) prev = prev._prev if self._operation is None: return ps # Accumulate dependencies in args and/or kwargs for ref in resolve_ref(self._operation['fn']): if ref not in ps: ps.append(ref) for arg in list(self._operation['args'])+list(self._operation['kwargs'].values()): for ref in resolve_ref(arg, recursive=True): if ref not in ps: ps.append(ref) return ps def _setup_invalidations(self, depth: int = 0): """ Since the parameters of the pipeline can change at any time we have to invalidate the internal state of the pipeline. To handle both invalidations of the inputs of the pipeline and the pipeline itself we set up watchers on both. 1. The first invalidation we have to set up is to re-evaluate the function that feeds the pipeline. Only the root node of a pipeline has to perform this invalidation because all leaf nodes inherit the same shared_obj. This avoids evaluating the same function for every branch of the pipeline. 2. The second invalidation is for the pipeline itself, i.e. if any parameter changes we have to notify the pipeline that it has to re-evaluate the pipeline. This is done by marking the pipeline as `_dirty`. The next time the `_current` value is requested the value is resolved by re-executing the pipeline. """ if self._fn is not None: for _, params in full_groupby(self._fn_params, lambda x: id(x.owner)): fps = [p.name for p in params if p in self._root._fn_params] if fps: params[0].owner.param._watch(self._invalidate_obj, fps, precedence=-1) for _, params in full_groupby(self._internal_params, lambda x: id(x.owner)): params[0].owner.param._watch(self._invalidate_current, [p.name for p in params], precedence=-1) def _invalidate_current(self, *events): if all(event.obj is self._trigger for event in events): return self._dirty = True self._error_state = None def _invalidate_obj(self, *events): self._root._dirty_obj = True self._error_state = None async def _resolve_async(self, obj): self._current_task = task = asyncio.current_task() if inspect.isasyncgen(obj): async for val in obj: if self._current_task is not task: break self._current_ = val self._trigger.param.trigger('value') else: value = await obj if self._current_task is task: self._current_ = value self._trigger.param.trigger('value') def _lazy_resolve(self, obj): from .parameterized import async_executor if inspect.isgenerator(obj): obj = _to_async_gen(obj) async_executor(partial(self._resolve_async, obj)) def _resolve(self): if self._error_state: raise self._error_state elif self._dirty or self._root._dirty_obj: try: obj = self._obj if self._prev is None else self._prev._resolve() if obj is Skip or obj is Undefined: self._current_ = Undefined raise Skip operation = self._operation if operation: obj = self._eval_operation(obj, operation) if inspect.isasyncgen(obj) or inspect.iscoroutine(obj) or inspect.isgenerator(obj): self._lazy_resolve(obj) obj = Skip if obj is Skip: raise Skip except Skip: self._dirty = False return self._current_ except Exception as e: self._error_state = e raise e self._current_ = current = obj else: current = self._current_ self._dirty = False if self._method: # E.g. `pi = dfi.A` leads to `pi._method` equal to `'A'`. current = getattr(current, self._method, current) if hasattr(current, '__call__'): self.__call__.__func__.__doc__ = self.__call__.__doc__ return current def _transform_output(self, obj): """ Applies custom display handlers before their output. """ applies = False for predicate, (handler, opts) in self._display_handlers.items(): display_opts = { k: v for k, v in self._display_opts.items() if k in opts } display_opts.update(self._kwargs) try: applies = predicate(obj, **display_opts) except TypeError: applies = predicate(obj) if applies: new = handler(obj, **display_opts) if new is not obj: return new return obj @property def _callback(self): params = self._params def evaluate(*args, **kwargs): out = self._current if self._method: out = getattr(out, self._method) return self._transform_output(out) if params: return bind(evaluate, *params) return evaluate def _clone(self, operation=None, copy=False, **kwargs): operation = operation or self._operation depth = self._depth + 1 if copy: kwargs = dict( self._kwargs, _current=self._current, method=self._method, prev=self._prev, **kwargs ) else: kwargs = dict(prev=self, **dict(self._kwargs, **kwargs)) kwargs = dict(self._display_opts, **kwargs) return type(self)( self._obj, operation=operation, depth=depth, fn=self._fn, _shared_obj=self._shared_obj, _wrapper=self._wrapper, **kwargs ) def __dir__(self): current = self._current if self._method: current = getattr(current, self._method) extras = {attr for attr in dir(current) if not attr.startswith('_')} try: return sorted(set(super().__dir__()) | extras) except Exception: return sorted(set(dir(type(self))) | set(self.__dict__) | extras) def _resolve_accessor(self): if not self._method: # No method is yet set, as in `dfi.A`, so return a copied clone. return self._clone(copy=True) # This is executed when one runs e.g. `dfi.A > 1`, in which case after # dfi.A the _method 'A' is set (in __getattribute__) which allows # _resolve_accessor to record the attribute access as an operation. operation = { 'fn': getattr, 'args': (self._method,), 'kwargs': {}, 'reverse': False } self._method = None return self._clone(operation) def __getattribute__(self, name): self_dict = super().__getattribute__('__dict__') if not self_dict.get('_init') or name == 'rx' or name.startswith('_'): return super().__getattribute__(name) current = self_dict['_current_'] dirty = self_dict['_dirty'] if dirty: self._resolve() current = self_dict['_current_'] method = self_dict['_method'] if method: current = getattr(current, method) # Getting all the public attributes available on the current object, # e.g. `sum`, `head`, etc. extras = [d for d in dir(current) if not d.startswith('_')] if (name in extras or current is Undefined) and name not in super().__dir__(): new = self._resolve_accessor() # Setting the method name for a potential use later by e.g. an # operator or method, as in `dfi.A > 2`. or `dfi.A.max()` new._method = name try: new.__doc__ = getattr(current, name).__doc__ except Exception: pass return new return super().__getattribute__(name) def __call__(self, *args, **kwargs): new = self._clone(copy=True) method = new._method or '__call__' if method == '__call__' and self._depth == 0 and not hasattr(self._current, '__call__'): return self.set_display(*args, **kwargs) if method in rx._method_handlers: handler = rx._method_handlers[method] method = handler(self) new._method = None kwargs = dict(kwargs) operation = { 'fn': method, 'args': args, 'kwargs': kwargs, 'reverse': False } return new._clone(operation) #---------------------------------------------------------------- # rx pipeline APIs #---------------------------------------------------------------- def __array_ufunc__(self, ufunc, method, *args, **kwargs): new = self._resolve_accessor() operation = { 'fn': getattr(ufunc, method), 'args': args[1:], 'kwargs': kwargs, 'reverse': False } return new._clone(operation) def _apply_operator(self, operator, *args, reverse=False, **kwargs): new = self._resolve_accessor() operation = { 'fn': operator, 'args': args, 'kwargs': kwargs, 'reverse': reverse } return new._clone(operation) # Builtin functions def __abs__(self): return self._apply_operator(abs) def __str__(self): return self._apply_operator(str) def __round__(self, ndigits=None): args = () if ndigits is None else (ndigits,) return self._apply_operator(round, *args) # Unary operators def __ceil__(self): return self._apply_operator(math.ceil) def __floor__(self): return self._apply_operator(math.floor) def __invert__(self): return self._apply_operator(operator.inv) def __neg__(self): return self._apply_operator(operator.neg) def __pos__(self): return self._apply_operator(operator.pos) def __trunc__(self): return self._apply_operator(math.trunc) # Binary operators def __add__(self, other): return self._apply_operator(operator.add, other) def __and__(self, other): return self._apply_operator(operator.and_, other) def __contains_(self, other): return self._apply_operator(operator.contains, other) def __divmod__(self, other): return self._apply_operator(divmod, other) def __eq__(self, other): return self._apply_operator(operator.eq, other) def __floordiv__(self, other): return self._apply_operator(operator.floordiv, other) def __ge__(self, other): return self._apply_operator(operator.ge, other) def __gt__(self, other): return self._apply_operator(operator.gt, other) def __le__(self, other): return self._apply_operator(operator.le, other) def __lt__(self, other): return self._apply_operator(operator.lt, other) def __lshift__(self, other): return self._apply_operator(operator.lshift, other) def __matmul__(self, other): return self._apply_operator(operator.matmul, other) def __mod__(self, other): return self._apply_operator(operator.mod, other) def __mul__(self, other): return self._apply_operator(operator.mul, other) def __ne__(self, other): return self._apply_operator(operator.ne, other) def __or__(self, other): return self._apply_operator(operator.or_, other) def __rshift__(self, other): return self._apply_operator(operator.rshift, other) def __pow__(self, other): return self._apply_operator(operator.pow, other) def __sub__(self, other): return self._apply_operator(operator.sub, other) def __truediv__(self, other): return self._apply_operator(operator.truediv, other) def __xor__(self, other): return self._apply_operator(operator.xor, other) # Reverse binary operators def __radd__(self, other): return self._apply_operator(operator.add, other, reverse=True) def __rand__(self, other): return self._apply_operator(operator.and_, other, reverse=True) def __rdiv__(self, other): return self._apply_operator(operator.div, other, reverse=True) def __rdivmod__(self, other): return self._apply_operator(divmod, other, reverse=True) def __rfloordiv__(self, other): return self._apply_operator(operator.floordiv, other, reverse=True) def __rlshift__(self, other): return self._apply_operator(operator.rlshift, other) def __rmod__(self, other): return self._apply_operator(operator.mod, other, reverse=True) def __rmul__(self, other): return self._apply_operator(operator.mul, other, reverse=True) def __ror__(self, other): return self._apply_operator(operator.or_, other, reverse=True) def __rpow__(self, other): return self._apply_operator(operator.pow, other, reverse=True) def __rrshift__(self, other): return self._apply_operator(operator.rrshift, other) def __rsub__(self, other): return self._apply_operator(operator.sub, other, reverse=True) def __rtruediv__(self, other): return self._apply_operator(operator.truediv, other, reverse=True) def __rxor__(self, other): return self._apply_operator(operator.xor, other, reverse=True) def __getitem__(self, other): return self._apply_operator(operator.getitem, other) def __iter__(self): if isinstance(self._current, Iterator): while True: try: new = self._apply_operator(next) new.rx.value except RuntimeError: break yield new return elif not isinstance(self._current, Iterable): raise TypeError(f'cannot unpack non-iterable {type(self._current).__name__} object.') items = self._apply_operator(list) for i in range(len(self._current)): yield items[i] def _eval_operation(self, obj, operation): fn, args, kwargs = operation['fn'], operation['args'], operation['kwargs'] resolved_args = [] for arg in args: val = resolve_value(arg) if val is Skip or val is Undefined: raise Skip resolved_args.append(val) resolved_kwargs = {} for k, arg in kwargs.items(): val = resolve_value(arg) if val is Skip or val is Undefined: raise Skip resolved_kwargs[k] = val if isinstance(fn, str): obj = getattr(obj, fn)(*resolved_args, **resolved_kwargs) elif operation.get('reverse'): obj = fn(resolved_args[0], obj, *resolved_args[1:], **resolved_kwargs) else: obj = fn(obj, *resolved_args, **resolved_kwargs) return obj
def _rx_transform(obj): if not isinstance(obj, rx): return obj return bind(lambda *_: obj.rx.value, *obj._params) register_reference_transform(_rx_transform)