Source code for ndarray_listener._ndl

import weakref

import numpy as np


[docs]class float64(np.float64): r""" Examples -------- .. doctest:: >>> from ndarray_listener import ndl, float64 >>> >>> print(float64(1.5)) 1.5 >>> print(ndl(1.5)) 1.5 """ def __new__(cls, *args): return np.float64.__new__(cls, *args)
[docs] def talk_to(self, me): r"""Not implemented. Array-scalars are immutable. """ pass
[docs]class ndl(np.ndarray): r""" Examples -------- A scalar is stored as a zero-dimensional array much like a NumPy scalar: .. doctest:: >>> from __future__ import print_function >>> from ndarray_listener import ndl >>> from numpy import atleast_1d >>> >>> class Watcher(object): ... def __init__(self, msg): ... self._msg = msg ... ... def __call__(self): ... print(self._msg + " called me") ... >>> scalar = ndl(-0.5) >>> >>> you0 = Watcher("First guy") >>> you1 = Watcher("Second guy") >>> >>> scalar.talk_to(you0) >>> scalar.itemset(-1.0) First guy called me >>> s0 = scalar.copy() >>> s0.itemset(-0.5) First guy called me >>> s0.talk_to(you1) >>> scalar.itemset(0.0) First guy called me Second guy called me >>> >>> s1 = atleast_1d(scalar) >>> s1[0] = 1.0 First guy called me Second guy called me One-dimension arrays are also supported: .. doctest:: >>> from ndarray_listener import ndl >>> from numpy import atleast_1d >>> from numpy import set_printoptions >>> >>> set_printoptions(precision=2, suppress=True) >>> >>> vector = ndl([-0.5, 0.1]) >>> >>> you0 = Watcher("First guy") >>> you1 = Watcher("Second guy") >>> >>> vector.talk_to(you0) >>> >>> vector[0] = 0.0 First guy called me >>> vector[:] = 1.0 First guy called me >>> >>> v0 = vector.copy() >>> v0.itemset(0, 1.1) First guy called me >>> >>> v0.itemset(1, 2.2) First guy called me >>> >>> v1 = v0.ravel() >>> >>> v1.talk_to(you1) >>> vector[-1] = 9.9 First guy called me Second guy called me """ def __new__(cls, input_array): obj = np.asarray(input_array).view(cls) if hasattr(input_array, "_listeners"): obj._listeners = input_array._listeners else: obj._listeners = [] return obj def __array_finalize__(self, obj): if obj is None: return self._listeners = getattr(obj, "_listeners", []) def __setitem__(self, *args, **kwargs): super(ndl, self).__setitem__(*args, **kwargs) self.__notify() def __setattr__(self, *args, **kwargs): super(ndl, self).__setattr__(*args, **kwargs) if len(args) > 0 and args[0] == "_listeners": return self.__notify() def __getitem__(self, *args, **kwargs): v = super(ndl, self).__getitem__(*args, **kwargs) if isinstance(v, ndl): return v if np.isscalar(v): v = float64(v) else: v = ndl(v) for k in self._listeners: v.talk_to(k) return v def talk_to(self, me): self._listeners.append(_create_callback(me)) def __notify(self): dirty = False for k in self._listeners: cb = k() if cb is None: dirty = True else: cb() if dirty: self.__flush() def __flush(self): self._listeners = [k for k in self._listeners if k() is not None]
[docs] def itemset(self, *args, **kwargs): super(ndl, self).itemset(*args, **kwargs) self.__notify()
def _create_callback(cb): try: return weakref.WeakMethod(cb) except TypeError: def _callback(): try: return weakref.proxy(cb) except ReferenceError: return None return _callback