PedalPi - PluginsManager - Observers¶

An observer is a class that receives notifications of changes in model classes (Bank, Pedalboard, Effect, Param …).

Implementations¶

Some useful UpdatesObserver classes have been implemented. They are:

Autosaver: Allows save the changes automatically in json data files.
ModHost: Allows use mod-host, a LV2 host for Jack controllable via socket or command line

Using¶

For use a observer, it’s necessary register it in BanksManager:

>>> saver = Autosaver()  # Autosaver is a UpdatesObserver
>>> banks_manager = BanksManager()
>>> banks_manager.register(saver)

For access all observers registered, use BanksManager.observers:

>>> saver in banks_manager.observers
True

For remove a observer:

>>> banks_manager.unregister(saver)

Creating a observer¶

It is possible to create observers! Some ideas are:

Allow the use of other hosts (such as Carla);

Automatically persist changes;

Automatically update a human-machine interface (such as LEDs and displays that inform the state of the effects).

For create a observer, is necessary create a class that extends UpdatesObserver:

class AwesomeObserver(UpdatesObserver):
   ...

UpdatesObserver contains a number of methods that must be implemented in the created class. These methods will be called when changes occur:

class AwesomeObserver(UpdatesObserver):

    def on_bank_updated(self, bank, update_type, index, origin, **kwargs):
        pass

    def on_pedalboard_updated(self, pedalboard, update_type, index, origin, **kwargs):
        pass

    def on_effect_status_toggled(self, effect, **kwargs):
        pass

    def on_effect_updated(self, effect, update_type, index, origin, **kwargs):
        pass

    def on_param_value_changed(self, param, **kwargs):
        pass

    def on_connection_updated(self, connection, update_type, pedalboard, **kwargs):
        pass

Use the update_type attribute to check what type of change occurred:

class AwesomeObserver(UpdatesObserver):
    """Registers all pedalboards that have been deleted"""

    def __init__(self):
        super(AwesomeObserver, self).__init__()
        self.pedalboards_removed = []

    ...

    def on_pedalboard_updated(self, pedalboard, update_type, index, origin, **kwargs):
        if update_type == UpdateType.DELETED:
            self.pedalboards_removed.append(update_type)

    ...

Scope¶

Notification problem¶

There are cases where it makes no sense for an observer to be notified of a change. Usually this occurs in interfaces for control, where through them actions can be performed (activate an effect when pressing on a footswitch). Control interfaces need to know of changes that occur so that their display mechanisms are updated when some change occurs through another control interface.

Note that it does not make sense for an interface to be notified of the occurrence of any change if it was the one that performed the action.

A classic example would be an interface for control containing footswitch and a led. The footswitch changes the state of an effect and the led indicates whether it is active or not. If another interface to control (a mobile application, for example) changes the state of the effect to off, the led should reverse its state:

class MyControllerObserver(UpdatesObserver):

    ...

    def on_effect_status_toggled(self, effect, **kwargs):
        # Using gpiozero
        # https://gpiozero.readthedocs.io/en/stable/recipes.html#led
        self.led.toggle()

However, in this situation, when the footswitch changes the effect state, it is notified of the change itself. What can lead to inconsistency in the led:

def pressed():
    effect.toggle()
    led.toggle()

# footswitch is a button
# https://gpiozero.readthedocs.io/en/stable/recipes.html#button
footswitch.when_pressed = pressed

In this example, pressing the button:

pressed() is called;
The effect has its changed state (effect.toggle());
on_effect_status_toggled(self, effect, ** kwargs) is called and the led is changed state (self.led.toggle());
Finally, in pressed() is called led.toggle().

That is, led.toggle() will be called twice instead of one.

Scope solution¶

Using with keyword, you can indicate which observer is performing the action, allowing the observer not to be notified of the updates that occur in the with scope:

>>> with observer1:
>>>     del manager.banks[0]

Example¶

Note

The complete example can be obtained from the examples folder of the repository. observer_scope.py

Consider an Observer who only prints actions taken on a bank:

class MyAwesomeObserver(UpdatesObserver):

    def __init__(self, message):
        super(MyAwesomeObserver, self).__init__()
        self.message = message

    def on_bank_updated(self, bank, update_type, **kwargs):
        print(self.message)

    ...

We will create two instances of this observer and perform some actions to see how the notification will occur:

>>> observer1 = MyAwesomeObserver("Hi! I am observer1")
>>> observer2 = MyAwesomeObserver("Hi! I am observer2")

>>> manager = BanksManager()
>>> manager.register(observer1)
>>> manager.register(observer1)

When notification occurs outside a with scope, all observers are informed of the change:

>>> bank = Bank('Bank 1')
>>> manager.banks.append(bank)
"Hi! I am observer1"
"Hi! I am observer2"

We’ll now limit the notification by telling you who performed the actions:

>>> with observer1:
>>> with observer1:
...     del manager.banks[0]
"Hi! I am observer2"
>>> with observer2:
...     manager.banks.append(bank)
"Hi! I am observer1"

If there is with inside a with block, the behavior will not change, ie it will not be cumulative

 with observer1:
     manager.banks.remove(bank)
     with observer2:
         manager.banks.append(bank)

Line 2 will result in Hi! I am observer2 and line 4 in Hi! I am observer1

Base API¶

UpdateType¶

class pluginsmanager.observer.update_type.UpdateType[source]¶

Enumeration for informs the change type.

See UpdatesObserver for more details

CREATED = 0¶: Informs that the change is caused by the creation of an object

DELETED = 2¶: Informs that the change is caused by the removal of an object

UPDATED = 1¶: Informs that the change is caused by the update of an object

UpdatesObserver¶

class pluginsmanager.observer.updates_observer.UpdatesObserver[source]¶

The UpdatesObserver is an abstract class definition for treatment of changes in some class model. Your methods are called when occurs any change in Bank, Pedalboard, Effect, etc.

To do this, it is necessary that the UpdatesObserver objects be registered in some manager, so that it reports the changes. An example of a manager is BanksManager.

__init__()[source]¶: Initialize self. See help(type(self)) for accurate signature.

on_bank_updated(bank, update_type, index, origin, **kwargs)[source]¶

Called when changes occurs in any Bank

Parameters:	bank (Bank) – Bank changed. update_type (UpdateType) – Change type index (int) – Bank index (or old index if update_type == UpdateType.DELETED) origin (BanksManager) – BanksManager that the bank is (or has) contained Bank – Contains the old bank occurs a UpdateType.UPDATED

on_connection_updated(connection, update_type, pedalboard, **kwargs)[source]¶

Called when changes occurs in any pluginsmanager.model.connection.Connection of Pedalboard (adding, updating or removing connections)

Parameters:	connection (pluginsmanager.model.connection.Connection) – Connection changed update_type (UpdateType) – Change type pedalboard (Pedalboard) – Pedalboard that the connection is (or has) contained

on_custom_change(identifier, *args, **kwargs)[source]¶

Called in specific changes that do not fit the other methods. See CustomChange for more details.

Also, called when any changes not officially supported by the PluginsManager library are made.

Developers can implement libraries on PluginsManager to control different equipment. For example, you can implement a host that will communicate with a GT100 or Zoom G3 boss pedal. If the device has any features not officially supported by the PluginsManager library, this method is useful.

For example: Zoom G3 pedalboards have level. If someone changes the level value, this method can be used to communicate this information to all other observers.

Parameters:	identifier – Unique identifier to informs that which informs the type of change in which it is being informed

on_effect_status_toggled(effect, **kwargs)[source]¶

Called when any Effect status is toggled

Parameters:	effect (Effect) – Effect when status has been toggled

on_effect_updated(effect, update_type, index, origin, **kwargs)[source]¶

Called when changes occurs in any Effect

Parameters:	effect (Effect) – Effect changed update_type (UpdateType) – Change type index (int) – Effect index (or old index if update_type == UpdateType.DELETED) origin (Pedalboard) – Pedalboard that the effect is (or has) contained

on_param_value_changed(param, **kwargs)[source]¶

Called when a param value change

Parameters:	param (Param) – Param with value changed

on_pedalboard_updated(pedalboard, update_type, index, origin, **kwargs)[source]¶

Called when changes occurs in any Pedalboard

Parameters:	pedalboard (Pedalboard) – Pedalboard changed update_type (UpdateType) – Change type index (int) – Pedalboard index (or old index if update_type == UpdateType.DELETED) origin (Bank) – Bank that the pedalboard is (or has) contained old (Pedalboard) – Contains the old pedalboard when occurs a UpdateType.UPDATED

pluginsmanager.observer.observable_list.ObservableList¶

class pluginsmanager.observer.observable_list.ObservableList(lista=None)[source]¶

Detects changes in list.

In append, in remove and in setter, the observer is callable with changes details

Based in https://www.pythonsheets.com/notes/python-basic.html#emulating-a-list

__contains__(item)[source]¶: See list.__contains__() method

__delitem__(sliced)[source]¶

See list.__delitem__() method

Calls observer self.observer(UpdateType.DELETED, item, index) where item is self[index]

__getitem__(index)[source]¶: See list.__getitem__() method

__init__(lista=None)[source]¶: Initialize self. See help(type(self)) for accurate signature.

__iter__()[source]¶: See list.__iter__() method

__len__()[source]¶: See list.__len__() method

__repr__()[source]¶: See list.__repr__() method

__setitem__(index, val)[source]¶

See list.__setitem__() method

Calls observer self.observer(UpdateType.UPDATED, item, index) if val != self[index]

__str__()[source]¶: See list.__repr__() method

append(item)[source]¶

See list.append() method

Calls observer self.observer(UpdateType.CREATED, item, index) where index is item position

index(x)[source]¶: See list.index() method

insert(index, x)[source]¶

See list.insert() method

Calls observer self.observer(UpdateType.CREATED, item, index)

move(item, new_position)[source]¶

Moves a item list to new position

Calls observer self.observer(UpdateType.DELETED, item, index) and observer self.observer(UpdateType.CREATED, item, index) if val != self[index]

Parameters:	item – Item that will be moved to new_position new_position – Item’s new position

pop(index=None)[source]¶

See list.pop() method

Remove the item at the given position in the list, and return it. If no index is specified, a.pop() removes and returns the last item in the list.

Parameters:	index (int) – element index that will be removed
Returns:	item removed

remove(item)[source]¶

See list.remove() method

Calls observer self.observer(UpdateType.DELETED, item, index) where index is item position

Implementations API¶

pluginsmanager.observer.autosaver.autosaver.Autosaver¶

class pluginsmanager.observer.autosaver.autosaver.Autosaver(data_path, auto_save=True)[source]¶

The UpdatesObserver Autosaver allows save any changes automatically in json data files. Save all plugins changes in json files in a specified path.

It also allows loading of saved files:

>>> system_effect = SystemEffect('system', ('capture_1', 'capture_2'), ('playback_1', 'playback_2'))
>>>
>>> autosaver = Autosaver('my/path/data/')
>>> banks_manager = autosaver.load(system_effect)

When loads data with Autosaver, the autosaver has registered in observers of the banks_manager generated:

>>> autosaver in banks_manager.observers
True

For manual registering in BanksManager uses register():

>>> banks_manager = BanksManager()
>>> autosaver = Autosaver('my/path/data/')
>>> autosaver in banks_manager.observers
False
>>> banks_manager.register(autosaver)
>>> autosaver in banks_manager.observers
True

After registered, any changes in Bank, Pedalboard, Effect, Connection or Param which belong to the structure of BanksManager instance are persisted automatically by Autosaver:

>>> banks_manager = BanksManager()
>>> banks_manager.register(autosaver)
>>> my_bank = Bank('My bank')
>>> banks_manager.append(my_bank)
>>> # The bank will be added in banksmanger
>>> # and now is observable (and persisted) by autosaver

It’s possible disables autosaver for saves manually:

>>> autosaver.auto_save = False
>>> autosaver.save(banks_manager)  # save() method saves all banks data

Parameters:	data_path (string) – Path that banks will be saved (each bank in one file) auto_save (bool) – Auto save any change?

__init__(data_path, auto_save=True)[source]¶: Initialize self. See help(type(self)) for accurate signature.

load(system_effect)[source]¶

Return a BanksManager instance contains the banks present in data_path

Return BanksManager:
Parameters:	system_effect (SystemEffect) – SystemEffect used in pedalboards
	`BanksManager` with banks persisted in `data_path`