Hooking setting of a value on a numpy array

Question

It is straight forward to hook direct setting of an attribute on a class.

class A(object):

        def __setattr__(self, key, value):
               print(f'__setattr__: {key} = {value}')
               super(A, self).__setattr__(key, value)

        def __getattribute__(self, key):
               print(f'__getattribute__: {key}')

a = A()
a.attribute = np.array([1,2,3])
__setattr__: attribute = [1 2 3]

This catches direct setting of attributes, but does not catch changing of the array.

Current behaviour:

a.attribute[2] = 5
__getattribute__: attribute

I am trying to catch the changing of the array. My understanding of the Python implementation of this is the array is being returned from the class, then modified separately, so it would not be possible. But wanted to check.

Answer 1

What class A stores is not the object itself, but a reference to the array. As such, when you set the index 2 of attribute to 5, you're not actually editing a.attribute, so that no call to __setattribute__ is performed. What A objects does in this case is to provide the interpreter with the array reference by accessing its own attribute (and, indeed, your customized __getattribute__ is invoked); but then, array modification is of ndarray class concern.

You could use some workaround to achieve this, but with a notable overhead, both computational and developmental.

The easiest way is to always prevent the user from modifying the array: you use a mid layer of methods for modifying the array. This is pedantic, difficult to implement and you would loose a lot of straight facilities with array slicing.

On the one hand, you could use memmap to write map an array on a file and, concurrently, you can use a framework (e.g. Qt QFileSystemWatcher) to be signalled for file changes. The issue is that a simple assignment is not enough, you have to flush the array each time you change it.

On the other hand you could create a watcher thread that continuously (and asynchronously) look at your array and store a copy of it, something like (only for didactic purpose! a twin part must be implemented in main thread to receive the event)

from threading import Thread, Event
from time import sleep
class watcher(Thread):
   def __init__(self, array_to_be_watched):
      self.array_changed = Event()
      self.active = True
      self.array = array_to_be_watched
   
   def run(self):
      while self.active:
         array = self.array.copy()
         # determines how frequently you'd like to check
         sleep(0.01)
         if np.array_equal(self.array, array):
            # To be cleared when caught on main loop
            self.array_changed.set()

More info on Event

Some drawback: checking the array is a CPU demanding task. Multithreading is not really efficient for this option. You should keep sleep time not so low! You better use multiprocessing, but with a more complex architecture employing Shared Memory or similar, and you should allow for the delay in IPC.

Some other technique that might be of your interest here