Dynamically updating Subscribers in Python based on changes in nested dataclasses

28 views Asked by At

I am currently developing scientific modeling software in Python based on PyQt and VTK libraries. The software comprises both 3D and 2D views with various control elements.

I have dynamically added 3D and 2D custom classes widgets, each with numerous parameters including data descriptions and the state of view widgets. The data consists of dataset arrays and description dataclasses, which are nested.

I am seeking advice on the most rational approach to dynamically update these widgets based on changes to the parameters within the model dataclasses. Currently, I have implemented and tested a solution using Python dicts where there is a listeners list, and every time the 'update' method is called on a specific key within the dict, all listeners subscribed to that key are informed. However, this method only works for single-level dicts and not for nested dicts.

@dataclass
class OurListner:
    obj_names : list[str]
    callbacks : list[callable]

class VSDsimModel:

    def __init__(self)  -> None:
        self.model_objects : dict[str, np.ndarray | int | float] | None = dict() 
        self.listners : list[OurListner]  | None = list() # [[names, functions]]
  
    def update_objects( self, objects_to_update : dict[str, np.ndarray]):
        for key, val in objects_to_update.items():
            self.model_objects[key] = val
            listners_to_call = filter(lambda listner: key in listner.obj_names , self.listners)
            for listner in listners_to_call:  
                for callback in listner.callbacks:
                    callback()

usage:

        self.our_model.update_objects({
            'points0': points0,
            'vm_points_scalar0': vm_points_scalar0,
            'vm_voxels0': vm_voxels0,
        })        

        self.our_model.listners.append(Model.OurListner( 
            obj_names=['isoval_slider0','i_t', 'i_xslice', 'alphaslider0', 'alphaslider1', 'alphaslider2', 'fluence_voxels0'],
            callbacks=[
                       self.vm_points_view.refresh,
                       self.vm_voxels_view.refresh,
                       self.fluence_voxels_view.refresh,
            ]
        ))

Now, I am considering converting the dict to dataclasses as they are better for type checking. I am also contemplating the implementation of some sort of Observer pattern where subscribers are automatically informed of changes to attributes within the nested model dataclasses.

I am wondering if there are any existing libraries or if you could suggest a better approach to tackle this issue. I have looked into the RxPy library, but it seems more suited for complex problems involving pipelined events and multiprocessing, and may not simplify this specific problem significantly.

Thank you in advance for any insights or suggestions you can provide.

0

There are 0 answers