Commit c9b40c2b authored by Yakubov, Sergey's avatar Yakubov, Sergey
Browse files

Merge branch '4-panel-binding' into 'main' 

Resolve "Panel binding"

Closes #4

See merge request ndip/public-packages/py-mvvm!3
parents 8febe009 8879fcf7

poetry.lock

+1088 −128

File changed.

Preview size limit exceeded, changes collapsed.

py_mvvm/panel_binding/__init__.py

0 → 100644
+1 −0
Original line number Diff line number Diff line 
from .binding import *
 
 No newline at end of file

py_mvvm/panel_binding/binding.py

0 → 100644
+132 −0
Original line number Diff line number Diff line 
import inspect

import param



from ..interface import BindingInterface, rsetattr, rgetattr



def is_parameterized(var):

    return isinstance(var, param.Parameterized)



def is_callable(var):

    return inspect.isfunction(var) or inspect.ismethod(var)

class Communicator:

    def __init__(self, viewmodel_linked_object=None, linked_object_attributes=None, callback_after_update=None):

        self.viewmodel_linked_object = viewmodel_linked_object

        self.linked_object_attributes = linked_object_attributes

        self.callback_after_update = callback_after_update



        self._set_linked_object_attributes(linked_object_attributes, viewmodel_linked_object)



        self.connection = None

        self.param_connect = None

        self.linked_object_parameterized = None



    def _set_linked_object_attributes(self, linked_object_attributes, viewmodel_linked_object):

        self.linked_object_attributes = None

        if (viewmodel_linked_object and

                not is_callable(viewmodel_linked_object)):

            if not linked_object_attributes and not isinstance(viewmodel_linked_object, dict):

                self.linked_object_attributes = {k: v for k, v in viewmodel_linked_object.__dict__.items() if

                                                 not k.startswith("_")}

            elif not linked_object_attributes and isinstance(viewmodel_linked_object, dict):

                self.linked_object_attributes = viewmodel_linked_object

            else:

                self.linked_object_attributes = linked_object_attributes



    # connector can be a dictionary, function, or parameterized object

    def connect(self, connector=None, param_connect=None):

        if is_parameterized(connector):

            self.connection = connector

            self.param_connect = param_connect

        

        elif is_callable(connector):

            self.connection = connector

            return self.get_callback()



        # Register an observer on a parameterized object with specified parameters to watch and call the update function on a single parameter

        if self.viewmodel_linked_object:

            # Connection on the View side should be a dictionary which has a key with string specifying the attribute name in the viewmodel linked

            # the value should be a tuple with this format (parameterized_object, 'parameter', [optional,observers])

            if self.linked_object_attributes and isinstance(connector, dict):

                for attribute_name, connection in connector.items():

                    if attribute_name in self.linked_object_attributes:

                        if not isinstance(connection, tuple) or len(connection) < 2:

                            raise ValueError(f"Expected tuple with at least two elements for {attribute_name}")

                        # creates an watcher on the parameterized object, and uses a specific parameter of the object to get and set values from

                        if len(connection) > 2:

                            param_observable = connection[2]

                        else:

                            param_observable = connection[1]

                        try:

                            self.connection = connector

                            parameterized = connection[0]

                            param_connector = connection[1]

                            

                            if is_parameterized(parameterized):

                                parameterized.param.watch(lambda event, key=attribute_name, parameter=param_connector:self._update_in_viewmodel(events=event, key=key, parameter=parameter), param_observable)

                            else:

                                raise Exception(f"Cannot create observer for attribute: {attribute_name} and parameter {param_connector}")

                        except:

                            raise Exception("Cannot connect", attribute_name)



    # Update the viewmodel based on the event triggered or the provided value

    # event parameter is expected but will default to the value parameter if not

    def _update_in_viewmodel(self, events=None, key=None, value=None, parameter=None):



        # Checks to see if the event triggered is the correct event that was specified for the connection

        if not events == None:

            if events.name == parameter:

                value = events.new

            else:

                if self.callback_after_update:

                    self.callback_after_update(key)

                return



        if value == None:

            raise Exception("Could not update viewmodel due to invalid value")



        if self.viewmodel_linked_object:

            if self.linked_object_attributes and key in self.linked_object_attributes:

                if not isinstance(self.viewmodel_linked_object, dict):

                    rsetattr(self.viewmodel_linked_object, key, value)

                else:

                    self.viewmodel_linked_object.update({key: value})

            elif is_callable(self.viewmodel_linked_object):

                self.viewmodel_linked_object(value)

            else:

                raise Exception("cannot update", self.viewmodel_linked_object)



        if self.callback_after_update:

            self.callback_after_update(key)

    

    # Return the update function as a callback

    def get_callback(self):

        return self._update_in_viewmodel



    # Update the view based on the provided value

    def update_in_view(self, value):

        if is_callable(self.connection):

            self.connection(value)

        elif self.viewmodel_linked_object:

            if self.linked_object_attributes:

                for attribute_name in self.linked_object_attributes:

                    if not isinstance(self.viewmodel_linked_object, dict):

                        value_to_change = rgetattr(value, attribute_name)

                    else:

                        value_to_change = self.viewmodel_linked_object[attribute_name]

                    widget, param = self.connection.get(attribute_name, (None,None))[:2]

                    if widget and param:

                        rsetattr(widget, param, value_to_change)                 

            elif is_callable(self.viewmodel_linked_object):

                self.viewmodel_linked_object(value)

        else:

            rsetattr(self.connection, self.param_connect, value)



        return value



class PanelBinding(BindingInterface):



    def new_bind(self, linked_object=None, linked_object_arguments=None, callback_after_update=None):

        # each new_bind returns an object that can be used to bind a ViewModel/Model variable

        # with a corresponding GUI framework element

        # for Trame we use state to trigger GUI update and linked_object to trigger ViewModel/Model update

        return Communicator(linked_object, linked_object_arguments, callback_after_update)
 
 No newline at end of file

pyproject.toml

+2 −0
Original line number Diff line number Diff line
@@ -11,10 +11,12 @@ packages = [{include = "py_mvvm"}] 
python = "^3.10"

pyqt6 = { version = "^6.7.0", optional = true }

trame = { version = "^3.5.1", optional = true }

panel = { version = "^1.4.4", optional = true }



[tool.poetry.extras]

pyqt6 = ["pyqt6"]

trame = ["trame"]

panel = ["panel"]



[build-system]

requires = ["poetry-core"]