API

This document describes the Python API to connect to Mixer’s Interactive service. We assume you’re already somewhat familiar with Interactive. If you’re not, check our reference guide.

To connect to Interactive, you first need to have an Interactive project and some means of authentication. You’ll want to create an Interactive project on Mixer and register an OAuth client while you’re there. For most integrations, you can leave the OAuth “redirect URLs” empty, and you should not request a secret key.

You can now use the OAuthClient class with your OAuth client ID to get an access token.

OAuth

class interactive_python.OAuthClient(client_id, client_secret=None, host='https://mixer.com', scopes=['interactive:robot:self'], loop=None)

The OAuth client implements our shortcode OAuth flow. From a user’s perspective, your app or game displays a 6-digit code to them and prompts them to go to mixer.com/go. This library will resolve back into your application once they enter that code.

Here’s a full example of what a full usage might look like, with error handling:

import interactive_python as interactive
import asyncio

async def get_access_token(client):
    code = await client.get_code()
    print("Go to mixer.com/go and enter {}".format(code.code))

    try:
        return await code.accepted()
    except interactive.ShortCodeAccessDeniedError:
        print("The user denied access to our client")
    except interactive.ShortCodeTimeoutError:
        print("Yo, you're too slow! Let's try again...")
        return await get_access_token(client)

async def run():
    with interactive.OAuthClient(my_client_id) as client:
        token = await get_access_token(client)
        print("Access token: {}".format(token.access))

asyncio.get_event_loop().run_until_complete(run())

Parameters:

• client_id (string) – Your OAuth client ID
• client_secret (string) – Your OAuth client secret, if any
• host (string) – Base address of the Mixer servers
• scopes (list[string]) – A list of scopes to request. For Interactive, you only need the ‘interactive:robot:self’ scope
• loop (asyncio.Loop) – asyncio event loop to attach to

coroutine get_code()

Requests a shortcode from the Mixer servers and returns an OAuthShortCode handle.

Return type:OAuthShortCode

class interactive_python.OAuthShortCode(grant, client, body, check_interval=1)

OAuthShortCode is the shortcode handle returned by the OAuthClient. See documentation on that class for more information and usage examples.

code

The short six-digit code to be displayed to the user. They should be prompted to enter it on mixer.com/go.

coroutine accepted()

Waits until the user enters the shortcode on the Mixer website. Throws if they deny access or don’t enter the code in time.

Raises:

• ShortCodeAccessDeniedError – if the user denies access
• ShortCodeTimeoutError – if the user doesn’t enter

Return type:

OAuthTokens

class interactive_python.OAuthTokens(body)

OAuthTokens is a bearer from an OAuth access and refresh token retrieved via the accepted() method.

access

The OAuth access token to use in connect().

refresh

The OAuth refresh token that can be used to re-grant the access token.

expires_at

The datetime at which the access token expires.

State

class interactive_python.State(connection)

Bases: pyee.EventEmitter

State is the state container for a single interactive session. It should usually be created via the static connect() method:

connection = State.connect(
    project_version_id=my_version_id,
    authorization="Bearer " + oauth_token)

The Scene is a pyee.EventEmitter. When calls come down, they’re always emitted on the State by their method name. So, for instance, you can listen to “onSceneCreate” or “onParticipantJoin” on the scene:

def greet(call):
    for participant in call.data['participants']:
        print('Welcome {}!', participant['username'])

scene.on('onParticipantJoin', greet)

The state can work in two modes for handling delivery of events and updates. You can use pump() calls synchronously within your game loop to apply updates that have been queued. Alternately, you can call pump_async() to signal to that state that you want updates delivered asynchronously, as soon as they come in. For example:

# Async delivery. `giveInput` is emitted as soon as any input comes in.
state.on('giveInput', lambda call: do_the_thing(call))
state.pump_async()

# Sync delivery. `giveInput` is emitted only during calls to pump()
state.on('giveInput', lambda call: do_the_thing(call))
while True:
    my_game_loop.tick()
    state.pump()

    # You can also read queues of changes from pump(), if you prefer
    # to dispatch changes manually:
    # for call in pump(): ...

In both modes, all incoming call are emitted as events on the State instance.

Parameters:connection (Connection) – The websocket connection to interactive.

connection

The underlying Connection to the Interactive service. You should not need to deal with this most of the time.

coroutine connect(discovery=<interactive_python.discovery.Discovery object>, **kwargs)

Creates a new interactive connection. Most arguments will be passed through into the Connection constructor.

Parameters:

• discovery (Discovery) –
• kwargs –

Return type:

State

coroutine create_scenes(*scenes)

Can be called with one or more Scenes to add them to Interactive. :param scenes: list of scenes to create :type scenes: Scene

pump()

pump causes the state to read any updates it has queued up. This should usually be called at the start of any game loop where you’re going to be doing processing of Interactive events.

Any events that have not been read when pump() is called are discarded.

Alternately, you can call pump_async() to have delivery handled for you without manual input.

Return type:Iterator of Calls

pump_async(loop=<_UnixSelectorEventLoop running=False closed=False debug=False>)

Starts a pump() process working in the background. Events will be dispatched asynchronously.

Returns a future that can be used for cancelling the pump, if desired. Otherwise the pump will automatically stop once the connection is closed.

Return type:asyncio.Future

scene(name)

Looks up an existing scene by ID. It returns None if the scene does not exist.

Parameters:name (str) – The name of the scene to look up Returns: Return type:Scene

coroutine set_ready(is_ready=True)

Marks the interactive integration as being ready-to-go. Must be called before controls will appear.

Parameters:is_ready – True or False to allow input Return type:Reply

class interactive_python.Discovery(host='https://mixer.com', path='/api/v1/interactive/hosts', loop=None, timeout=10)

Bases: object

Discovery is a simple service discovery class which retrieves an Interactive host to connect to. This is passed into the State by default; you usually should not need to override it.

coroutine find()

Returns the websocket address of an interactive server to connect to, or raises a NoServersAvailableError.

Scenes

class interactive_python.Scene(scene_id, **kwargs)

Bases: interactive_python._util.Resource

Scene is a container for controls in interactive. Groups can be assigned to scenes. It emits:

• A delete event when the scene is deleted, with the Call from “onSceneDelete”.
• An update event when the scene is updated, with the Call from “onSceneUpdate”.

controls

A dict of control IDs to :class:`~interactive_python.Control`s in the scene.

coroutine create_controls(*controls)

Can be called with one or more Controls to add them to to the scene. :param controls: list of controls to create :type controls: List[Control]

coroutine delete(reassign_scene_id='default')

Deletes the scene from Interactive. Takes the id of the scene to reassign any groups who are on that scene to. :param reassign_scene_id: :type reassign_scene_id: str :rtype: None

coroutine update(priority=0)

Saves all changes updates made to the scene.

class interactive_python.Control(control_id, **kwargs)

Bases: interactive_python._util.Resource

Control is a structure on which participants in interactive provide input. It emits:

• A delete event when the control is deleted, with the Call from “onControlDelete”.
• An update event when the control is updated, with the Call from “onControlUpdate”.
• Other kinds events are fired when input is given on the control, like mousedown for buttons or move for joysticks. They’re fired with the Call that triggered them.

Here’s an example of creating a button and listening for clicks:

btn = Button(
    control_id='click_me',
    text='Click Me!',
    keycode=keycode.space,
    position=[
        {'size': 'large', 'width': 5, 'height': 5, 'x': 0, 'y': 0},
    ],
)

# Logs the call to the console whenever the button is clicked
btn.on('mousedown', lambda call: print(call))

await state.scene('default').create_controls(btn)

coroutine delete()

Deletes the control :return: None

coroutine update()

Saves all changes updates made to the control.

class interactive_python.Button(control_id, **kwargs)

Bases: interactive_python.scene.Control

class interactive_python.Joystick(control_id, **kwargs)

Bases: interactive_python.scene.Control

Utilities

interactive_python.keycode

Keycode is an instance of a class that helps translate keycodes from their textual representation to their corresponding numeric code, as represented on the protocol. For example:

from interactive_python import keycode

print(keycode.up)     # => 38
print(keycode.a)      # => 65
getattr(keycode, '⌘') # => 91

class interactive_python._util.ChangeTracker(data={}, intrinsic_properties=[])

Bases: object

ChangeTracker is a simple structure that keeps track of changes made to properties of the object.

assign(**kwargs)

Assigns data in-bulk to the resource, returns the resource. :rtype: Resource

has_changed()

Returns whether any metadata properties have changed. :rtype: bool

to_json()

to_json will be called when serializing the resource to JSON. It returns the raw data. :return:

class interactive_python._util.Resource(id, id_property)

Bases: pyee.EventEmitter, interactive_python._util.ChangeTracker

Resource represents some taggable, metadata-attachable construct in Interactive. Scenes, groups, and participants are resources.

id

Return type:int

Errors

class interactive_python.DiscoveryError

Bases: Exception

Raised if some error occurs during service discovery that we didn’t anticipate.

class interactive_python.NoServersAvailableError

Bases: Exception

Raised if Beam reports that no servers are available.

class interactive_python.ShortCodeError

Bases: Exception

Base exception raised when some unexpected event occurs in the shortcode OAuth flow.

class interactive_python.UnknownShortCodeError

Bases: interactive_python.errors.ShortCodeError

Exception raised when an unknown error happens while running shortcode OAuth.

class interactive_python.ShortCodeAccessDeniedError

Bases: interactive_python.errors.ShortCodeError

Exception raised when the user denies access to the client in shortcode OAuth.

class interactive_python.ShortCodeTimeoutError

Bases: interactive_python.errors.ShortCodeError

Exception raised when the shortcode expires without being accepted.

Low-Level Protocol

class interactive_python.Connection(address=None, authorization=None, project_version_id=None, project_sharecode=None, extra_headers={}, loop=<_UnixSelectorEventLoop running=False closed=False debug=False>, socket=None, protocol_version='2.0')

Bases: object

The Connection is used to connect to the Interactive server. It connects to a provided socket address and provides an interface for making RPC calls. Example usage:

connection = Connection(
    address=get_interactive_address(),
    authorization="Bearer {}".format(my_oauth_token),
    interactive_version_id=1234)

coroutine call(method, params, discard=False, timeout=10)

Sends a method call to the interactive socket. If discard is false, we’ll wait for a response before returning, up to the timeout duration in seconds, at which point it raises an asyncio.TimeoutError. If the timeout is None, we’ll wait forever.

Parameters:

• method (str) – Method name to call
• params – Parameters to insert into the method, generally a dict.
• discard (bool) – True to not request any reply to the method.
• timeout (int) – Call timeout duration, in seconds.

Returns:

The call response, or None if it was discarded.

Raises:

asyncio.TimeoutError

coroutine close()

Closes the socket connection gracefully

coroutine connect()

Connects to the Interactive server, waiting until the connection if fully established before returning. Can throw a ClosedError if something, such as authentication, fails.

get_packet()

Synchronously reads a packet from the connection. Returns None if there are no more packets in the queue. Example:

while await connection.has_packet():
    dispatch_call(connection.get_packet())

Return type:Call

coroutine has_packet()

Blocks until a packet is read. Returns true if a packet is then available, or false if the connection is subsequently closed. Example:

while await connection.has_packet():
    dispatch_call(connection.get_packet())

Return type:bool

reply(call_id, result=None, error=None)

Sends a reply for a packet id. Either the result or error should be fulfilled.

Parameters:

• call_id (int) – The ID of the call being replied to.
• result – The successful result of the call.
• error – The errorful result of the call.

coroutine set_compression(scheme)

Updates the compression used on the websocket this should be called with an instance of the Encoding class, for example:

connection.set_compression(GzipEncoding())

You can, optionally, await on the resolution of method, though doing so it not at all required. Returns True if the server agreed on and executed the switch.

Parameters:scheme (Encoding) – The compression scheme to use Returns:Whether the upgrade was successful Return type:bool

class interactive_python.Call(connection, payload)

Bases: object

data

Returns:The payload of the method being called. Return type:dict

name

Returns:The name of the method being called. Return type:str

reply_error(error)

Submits an errorful reply for the call. :param error: The error to send to tetrisd

class interactive_python.encoding.Encoding

Bases: object

Encoding is an abstract class that defines methods for decoding incoming and encoding outgoing websocket calls. Both encode() and decode() are allowed to raise EncodingExceptions, which will trigger a fallback to plain-text encoding.

decode(data)

decode takes a byte slice of data and returns it decoded, string form

encode(data)

encode takes a string of data and returns its encoded form

name()

exception interactive_python.encoding.EncodingException

Bases: Exception

An EncodingException is raised if an error occurs in an encoding or decoding algorithm. Raising this exception triggers a fallback to plain text encoding.

class interactive_python.encoding.GzipEncoding(compression_level=6)

Bases: interactive_python.encoding.Encoding

decode(data)

encode(data)

name()

class interactive_python.encoding.TextEncoding

Bases: interactive_python.encoding.Encoding

decode(data)

encode(data)

name()

interactive_python.encoding.reset_buffer(buffer, value=None)