Warm tip: This article is reproduced from serverfault.com, please click

How would I 'listen' to/decorate a setter from an imported class

发布于 2020-11-06 16:33:43

I'm not sure whether this is a great approach to be using, but I'm not hugely experienced with Python so please accept my apologies. I've tried to do some research on this but other related questions have been given alternative problem-specific solutions - none of which apply to my specific case.

I have a class that handles the training/querying of my specific machine learning model. This algorithm is running on a remote sensor, various values are fed into the object which returns None if the algorithm isn't trained. Once trained, it returns either True or False depending on the classification assigned to new inputs. Occasionally, the class updates a couple of threshold parameters and I need to know when this occurs.

I am using sockets to pass messages from the remote sensor to my main server. I didn't want to complicate the ML algorithm class by filling it up with message passing code and so instead I've been handling this in a Main class that imports the "algorithm" class. I want the Main class to be able to determine when the threshold parameters are updated and report this back to the server.

class MyAlgorithmClass:

    def feed_value(self):
         ....


class Main:

    def __init__(self):
        self._algorithm_data = MyAlgorithmClass()
        self._sensor_data_queue = Queue()

    def process_data(self):
        while True:
            sensor_value = self._sensor_data_queue.get()
            result, value = self._algorithm_data.feed_value(sensor_value)
            if result is None:
                # value represents % training complete
                self._socket.emit('training', value)
            elif result is True:
                # value represents % chance that input is categoryA
                self._socket.emit('categoryA', value)
            elif result is False:
                ...

My initial idea was to add a property to MyAlgorithmClass with a setter. I could then decorate this in my Main class so that every time the setter is called, I can use the value... for example:

class MyAlgorithmClass:
    
    @property
    def param1(self):
        return self._param1

    @param1.setter
    def param1(self, value):
        self._param1 = value


class Main:

    def __init__(self):
        self._algorithm_data = MyAlgorithmClass()
        self._sensor_data_queue = Queue()    

        def watch_param1(func):
            def inner(*args):
                self._socket.emit('param1_updated', *args)
            func(*args)

My problem now, is how do I decorate the self._algorithm_data.param1 setter with watch_param1? If I simply set self._algorithm_data.param1 = watch_param1 then I will just end up setting self._algorithm_data._param1 equal to my function which isn't what I want to do.

I could use getter/setter methods instead of a property, but this isn't very pythonic and as multiple people are modifying this code, I don't want the methods to be replaced/changed for properties by somebody else later on.

What is the best approach here? This is a small example but I will have slightly more complex examples of this later on and I don't want something that will cause overcomplication of the algorithm class. Obviously, another option is the Observer pattern but I'm not sure how appropriate it is here where I only have a single variable to monitor in some cases.

I'm really struggling to get a good solution put together so any advice would be much appreciated.

Thanks in advance,

Tom

Questioner
Tom Bailey
Viewed
0
Tom Bailey 2020-11-30 23:20:56

The solution I went for is as follows, using a 'Proxy' subclass which overrides the properties. Eventually, once I have a better understanding of the watched parameters, I won't need to watch them anymore. At this point I will be able to swap out the Proxy for the base class and continue using the code as normal.

class MyAlgorithmClassProxy(MyAlgorithmClass):

    @property
    def watch_param1(self):
        return MyAlgorithmClass.watch_param1.fget(self)

    @watch_param1.setter
    def watch_param1(self, value):
        self._socket.emit('param1_updated', *args)
        MyAlgorithmClass.watch_param1.fset(self, value)