Autosaver

[SrMouraSilva]

Description

It would be incredible if there were an observer for automatic persistence.
I tried to implement but got problems:

Replace

observer = Autosaver('/home/paulo/PycharmProjects/PedalPi-Raspberry/data/test/')

manager = BanksManager()
manager.register(observer)

bank1 = Bank('Bank 1')
pedalboard = Pedalboard('Rocksmith')
bank1.append(pedalboard)

bank2 = Bank('Bank 2')

manager.append(bank1)

# Important part
manager.banks[0] = bank2

Swap

observer = Autosaver('/home/paulo/PycharmProjects/PedalPi-Raspberry/data/test/')

manager = BanksManager()
manager.register(observer)

bank1 = Bank('Bank 1')
bank2 = Bank('Bank 2')

manager.banks.append(bank1)
manager.banks.append(bank2)


manager.banks[0], manager.banks[1] = manager.banks[1], manager.banks[0]

self.validate_persisted(manager)

[SrMouraSilva]

Class

from pluginsmanager.model.updates_observer import UpdatesObserver
from pluginsmanager.model.update_type import UpdateType

from pluginsmanager.util.persistence_decoder import PersistenceDecoder

import json
import asyncio
import os
from glob import glob


class Autosaver(UpdatesObserver):
    """
    Save all plugins changes in json files in a specified path.

    :param string data_path: Path that banks (one in one file)
    """
    def __init__(self, data_path):
        self.data_path = data_path

    def load(self, system_effect):
        """
        Return a bank list present in data_path

        :param SystemEffect system_effect: SystemEffect used in pedalboards
        :return list[bank]: Banks list in data_path
        """
        persistence = PersistenceDecoder(system_effect)
        banks = []

        for file in glob(self.data_path + "*.json"):
            bank = persistence.read(self._read(file))
            bank._uuid = file.split('/')[-1].split('.json')[0]
            banks.append(bank)

        return banks

    def _read(self, path):
        with open(path) as data_file:
            return json.load(data_file)

    def _bank_path(self, bank):
        """
        :param Bank bank:
        :return string: Bank path .json
        """
        return self.data_path + bank._uuid + ".json"

    def save(self, bank):
        """
        Save the bank in your current path
        :param Bank bank: Bank that will be saved
        """
        bank.index = bank.manager.banks.index(bank)

        loop = asyncio.get_event_loop()
        loop.run_until_complete(Autosaver._save(self._bank_path(bank), bank.json))

    def delete(self, bank):
        """
        Delete the bank's file
        :param Bank bank: Bank that will be removed
        """
        loop = asyncio.get_event_loop()
        loop.run_until_complete(Autosaver._delete(self._bank_path(bank)))

    @staticmethod
    @asyncio.coroutine
    def _save(url, data):
        json_file = open(url, "w+")
        json_file.write(json.dumps(data))
        json_file.close()

    @staticmethod
    @asyncio.coroutine
    def _delete(url):
        os.remove(url)

    def on_bank_updated(self, bank, update_type, origin=None, **kwargs):
        if update_type == UpdateType.DELETED:
            return self.delete(bank)

        self.save(bank)

    def on_pedalboard_updated(self, pedalboard, update_type, origin=None, **kwargs):
        if update_type == UpdateType.DELETED:
            self.save(origin)
        else:
            self.save(pedalboard.bank)

    def on_effect_updated(self, effect, update_type, **kwargs):
        self.save(effect.pedalboard.bank)

    def on_effect_status_toggled(self, effect):
        self.save(effect.pedalboard.bank)

    def on_param_value_changed(self, param):
        self.save(param.effect.pedalboard.bank)

    def on_connection_updated(self, connection, update_type):
        self.save(connection.output.effect.pedalboard.bank)

[SrMouraSilva]

Test

import unittest
from unittest.mock import MagicMock

from pluginsmanager.banks_manager import BanksManager

from pluginsmanager.model.bank import Bank
from pluginsmanager.model.pedalboard import Pedalboard

from pluginsmanager.model.lv2.lv2_effect_builder import Lv2EffectBuilder

from pluginsmanager.observer.autosaver import Autosaver


class AutoSaverTest(unittest.TestCase):

    def test_observers(self):
        mock = MagicMock()
        observer = Autosaver('data/test/')
        observer.save = mock
        observer.delete = mock

        manager = BanksManager()
        manager.register(observer)

        bank = Bank('Bank 1')
        manager.append(bank)
        observer.save.assert_called_with(bank)

        pedalboard = Pedalboard('Rocksmith')
        bank.append(pedalboard)
        observer.save.assert_called_with(bank)

        builder = Lv2EffectBuilder()
        reverb = builder.build('http://calf.sourceforge.net/plugins/Reverb')
        fuzz = builder.build('http://guitarix.sourceforge.net/plugins/gx_fuzzfacefm_#_fuzzfacefm_')
        reverb2 = builder.build('http://calf.sourceforge.net/plugins/Reverb')

        pedalboard.append(reverb)
        observer.save.assert_called_with(bank)
        pedalboard.append(fuzz)
        observer.save.assert_called_with(bank)
        pedalboard.append(reverb2)
        observer.save.assert_called_with(bank)

        reverb.outputs[0].connect(fuzz.inputs[0])
        observer.save.assert_called_with(bank)
        reverb.outputs[1].connect(fuzz.inputs[0])
        observer.save.assert_called_with(bank)
        fuzz.outputs[0].connect(reverb2.inputs[0])
        observer.save.assert_called_with(bank)
        reverb.outputs[0].connect(reverb2.inputs[0])
        observer.save.assert_called_with(bank)

        fuzz.toggle()
        observer.save.assert_called_with(bank)

        fuzz.params[0].value = fuzz.params[0].minimum / fuzz.params[0].maximum
        observer.save.assert_called_with(bank)

        del bank.pedalboards[0]
        observer.save.assert_called_with(bank)

        bank2 = Bank('Bank 2')
        manager.banks[0] = bank2
        observer.delete.assert_called_with(bank2)
        observer.save.assert_called_with(bank2)

        manager.banks.remove(bank2)
        observer.delete.assert_called_with(bank2)

    def test_replace_bank(self):
        observer = Autosaver('/home/paulo/PycharmProjects/PedalPi-Raspberry/data/test/')

        manager = BanksManager()
        manager.register(observer)

        bank1 = Bank('Bank 1')
        bank2 = Bank('Bank 2')

        manager.append(bank1)
        pedalboard = Pedalboard('Rocksmith')
        bank1.append(pedalboard)

        manager.banks[0] = bank2

        self.validate_persisted(manager)

        while manager.banks:
            del manager.banks[0]

    def test_swap_bank(self):
        observer = Autosaver('/home/paulo/PycharmProjects/PedalPi-Raspberry/data/test/')

        manager = BanksManager()
        manager.register(observer)

        bank1 = Bank('Bank 1')
        bank2 = Bank('Bank 2')

        manager.banks.append(bank1)
        manager.banks.append(bank2)

        manager.banks[0], manager.banks[1] = manager.banks[1], manager.banks[0]

        self.validate_persisted(manager)

        while manager.banks:
            del manager.banks[0]

    def validate_persisted(self, manager):
        autosaver_validation = Autosaver('/home/paulo/PycharmProjects/PedalPi-Raspberry/data/test/')
        banks = autosaver_validation.load(None)

        self.assertEqual(len(manager.banks), len(banks))

        for bank_manager, bank_persisted in zip(manager.banks, banks):
            self.assertEqual(bank_manager.json, bank_persisted.json)

[SrMouraSilva]

Fixed by #8 (351f82c)