"""
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 emmitted.
"""
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):
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):
return self._transform_output(self._current)
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)