Skip to content

Python usage

miklosduma edited this page Aug 21, 2018 · 1 revision

All SPARKL CLI commands have their Python equivalent too.

You can run them through the sparkl(cmd_name, *args, **kwargs) function.

See the available commands below.

sparkl('connect', 'http://localhost:8000', alias='default')
sparkl('close', all=False, alias='default')
sparkl('session', alias='default')
sparkl('login', '', 'password', register=False, token=None, alias='default')
sparkl('logout', alias='default')
sparkl('node', alias='default')
sparkl('ls', 'Scratch', alias='default')
sparkl('source', 'Scratch/Primes', output=None, json=False, xml=False, render=False, file=False, alias='default')
sparkl('put', 'examples/primes/primes.xml', 'Scratch', alias='default')
sparkl('rm', 'Scratch/Primes_expr', alias='default')
sparkl('object', 'Scratch/Primes_expr', alias='default')
sparkl('mkdir', 'Scratch/Foo', alias='default')
sparkl('undo', alias='default')
sparkl('vars', literal=[('n', 13)], read=[('audio', '~/Downloads/sample.wav')], clear=False, alias='default')
sparkl('call', 'Scratch/Primes_expr/Mix/Frontend/CheckPrime', alias='default')
sparkl('listen', 'Scratch/Primes_expr/Sequencer', alias='default')
sparkl('active', '/', alias='default')
sparkl('stop', '', alias='default')
sparkl('start', '', alias='default')

Getting started

You can combine multiple commands into a single function, for example to connect to and log into a SPARKL instance with a single function call.

from sparkl_cli.main import sparkl

def login_sparkl(alias, url, username):
    Uses SPARKL CLI to connect to SPARKL and log 
    into existing user account using a token.
    sparkl('connect', url, alias=alias)
    sparkl('login', username, alias=alias)
    # Return alias for other funs to use
    return alias	

Reusing scripts

You can use your custom functions - for example, the login_sparkl function from the module - in other modules.

import sys
from sparkl_cli.main import sparkl
from login_script import login_sparkl

def import_mix_to_sparkl(sparkl_config, url, username):
    Imports a SPARKL mix into a selected user tree
    from the file system.
    # Log into SPARKL with import_alias
    alias = login_sparkl('import_alias', url, username)
    # Import specified SPARKL mix into Scratch folder
    dest_dir = 'Scratch'
    sparkl('put', sparkl_config, dest_dir, alias=alias)
    print 'Importing {} into {}'.format(sparkl_config, dest_dir)

# Call fun with file system path to SPARKL mix
if __name__ == '__main__':
  [sparkl_config, url, username] = sys.argv[1:]
  import_mix_to_sparkl(sparkl_config, url, username)

The sample script imports a selected mix into your SPARKL user tree.

$ python /Examples/Phidgets1/Phidgets1.xml http://localhost:8000
Password: ********
Importing /Examples/Phidgets1/Phidgets1.xml into Scratch

Running a transaction

You can run transactions and write conditions based on their outcomes.

import sys
from sparkl_cli.main import sparkl
from login import login_sparkl

if sys.version_info.major == 3:
  PROMPT = input
  PROMPT = raw_input

def is_prime(value):
    Calls a SPARKL mix to decide whether 
    a number is prime.
    # Connect and log in
    alias = PROMPT('Enter your alias: ')
    url = PROMPT('Enter your instance url: ')
    user = PROMPT('Enter your username: ')
    login_sparkl(alias, url, user)
    # Set value of fields (n) the Primes mix needs
    sparkl('vars', alias=alias, literal=[('n', value)])	
    # Fire solicit operation to run Primes transaction
    operation_path = 'Scratch/Primes_expr/Mix/Frontend/CheckPrime'
    sparkl_response_struct = sparkl('call', operation_path, alias=alias)   

    # Get name of response from response struct
    # Response is either Yes or No
    sparkl_response = sparkl_response_struct['attr']['name']

    # Close connection
    sparkl('close', alias=alias) 
    # If response is Yes, the number is prime
    if sparkl_response == 'Yes':
        print '{} is prime.'.format(value)
        return True
    # Otherwise, it is not prime
    print '{} is not prime.'.format(value)
    return False 	

if __name__ == '__main__':
  number = sys.argv[1]

Structs and testing

If successful, every SPARKL Python command returns a struct, that has the following format:

# Struct is an object
    'tag':'', # Tag is a string. For example, 'pong' or 'connections'
    'attr': {}, # Attr is an embedded object, which can have multiple key-value pairs.
    'content': [] # Content is optional. It is a list of objects.

For example, checking the live connections returns the following:

>>> from sparkl_cli.main import sparkl
>>> sparkl("connect")
    'content': [
            'tag': 'connection', 
            'attr': {
                'url': u'http://localhost:8000', 
                'alias': u'default'
            'tag': 'connection', 
            'attr': {
                'url': u'', 
                'alias': u'cloud'
    'tag': 'connections', 
    'attr': {
        'count': 2

The struct returned on running a SPARKL transaction looks like this:

See Running transactions.

  'content': [
      'tag': 'datum', 
      'attr': {
        'field': 'B-11FT-ASW-682', 
        'type': None, 
        'name': 'YES'
  'tag': 'response', 
  'attr': {
    'id': 'J-11FT-ASW-6A2', 
    'name': 'Yes'

The set format of structs allows you to write tests and conditions. See below for some examples.

Testing if connection is up

from sparkl_cli.main import sparkl

def test_connect():
    Checks if list of live connections contains "my_connection" 
    {'content':[{'attr': {'url': URL,...}},...],...}
    my_connection = 'http://localhost:8000'
    connections_struct = sparkl('connect')
    live_connections = connections_struct['content']
    assert any( x['attr']['url'] == my_connection for x in live_connections)

Testing if service is up

from sparkl_cli.main import sparkl

def test_service():
    Checks if list of running services contains the given service 
    {'content':[{'attr': {'path': PATH, ...}},...],...}
    path_to_service = ''
    services_struct = sparkl('active')
    active_services = services_struct['content']

    assert any( x['attr']['path'] == path_to_service for x in active_services )

Testing a configuration

from sparkl_cli.main import sparkl

def test_config():
    Test Primes configuration with prime number.
    Expects Yes response to CheckPrime solicit.

    # Set value of n field
    field_vals = [('n', 13)]
    sparkl('vars', literal=field_vals)
    # Call CheckPrime solicit and get name of response
    operation = ''
    response_struct = sparkl('call', operation)
    response = response_struct['attr']['name']
    expected_response = 'Yes'
    test_fail_message = 'Expected Yes response, not {}.'.format(response)
    assert response == expected_response, test_fail_message


See sample Python scripts for working examples and our SPARKL test framework.

You can’t perform that action at this time.