Procedures, event hooks and cronjobs as Python scripts

As an alternative to the procedure and hooks defined in config.yaml, you can also declare them as native Python entities. This is especially useful if your custom logic is more complex than a sequence of a few actions.

By default, you can place your custom scripts under ~/.config/platypush/scripts. You can change the folder by setting the scripts_dir option in your config.yaml:

scripts_dir: /path/to/my/scripts

N.B. It is quite important that the scripts folder, as well as its subfolder that are supposed to contain loadable script, contains an __init__.py file, even if it's empty. That serves to Python to identify a folder as a loadable module. The __init__.py in scripts_dir will be created automatically, but if you're planning to add more subfolders with your scripts then you'll have to manually create the file in them.

Defining a procedure in a custom script is just a matter of using the decorator @procedure on any function:

# ~/.config/platypush/scripts/test.py

from platypush.procedure import procedure
from platypush.utils import run

@procedure
def test_procedure():
    # Run some action
    return run('music.mpd.pause')

You can then call the procedure either through any of the available APIs - e.g. over cURL:

curl -XPOST -H 'Content-Type: application/json' -d '
{
  "type": "request",
  "action":"procedure.test.test_procedure"
}' http://localhost:8008/execute

You can access a wide range of function directly from the platypush module in your code. Among those:

• platypush.context.get_plugin to load a plugin or get its loaded instance
• platypush.context.get_backend to get the loaded instance of a backend
• platypush.context.get_bus to access the internal bus directly (useful to dispatch any types of messages directly to the application)
• platypush.utils.run to run an action by full name (plugin.method) and arguments

Similarly, you can also define custom hooks:

# ~/.config/platypush/scripts/test.py

from platypush.event.hook import hook
from platypush.message.event.assistant import SpeechRecognizedEvent

@hook(SpeechRecognizedEvent)
def test_procedure(event, **content):
    print('Recognized speech: {}'.format(event.args['phrase']))

As well as cronjobs:

# ~/.config/platypush/scripts/test.py

from platypush.cron import cron

# Run it every minute
@cron('* * * * *')
def test_cron(event, **content):
    import os
    import datetime

    with open(os.path.expanduser('~/test.txt'), 'a') as f:
        f.write('The time is: {}\n'.format(datetime.datetime.now().isoformat()))