julian-o / pyslobs Goto Github PK

Streamlabs Desktop (formerly StreamLabs OBS) is a live-streaming application based on a fork of Open Broadcaster Software with additional features.

It offers the Streamlabs Desktop API to allow third-party applications to interact with the application while it is running.

This includes selecting, editing and monitoring scenes, video sources and audio sources. It includes monitoring performance.

This doesn't include chat or getting notifications about donations, subscriptions and followers. You will need to look elsewhere for that.

Typically, it would be accessed from same computer that is running Streamlabs Desktop, or from a computer on the same LAN.

The API is based on WebSockets, so it can be accessed from a browser in JavaScript. (Apparently, it can also be accessed through a named pipe.)

Rather than accessing it from Javascript, this Python wrapper allows access to Streamlabs Desktop from a Python application. The details about websockets are hidden and more Pythonic interfaces are provided.

This API requires Python 3.9 or later. It is tested on Python 3.10.

Streamlabs Desktop versions from 1.2.0-1.13.1 have been tested.

The Python interface is designed to allow you to write PEP8-compliant Python code.

Camel-case items in the original API are represented in their preferred PEP8 form - i.e. underscores between words to avoid ambiguity and capitalised constants.

Enumerated types and named tuples are used in preference to JSON dictionaries and numeric constants where possible.

Identifiers that clash with reserved words used in Python are suffixed with _ - e.g. id_().

Date-times from notifications do not have a timezone associated with them. They are in the timezone of the host running the API.

The API is moderately large. In this version of the Python wrapper, not every method offered has been tested. Testing has been focussed on the areas the developer actively wanted to use first.

See PROGRESS.md for an idea of what is and isn't tested.

This Python wrapper is based on asyncio. If you have not used the asyncio features of Python before, please consult a tutorial. In particular:

• You need to await all methods defined as async. Similarly, context managers defined as async need to be called with async with.

• After creation, remember to await the SlobsConnection.background_processing() coroutine.

• To shut-down cleanly, remember to close the connection.

• To avoid exceptions being swallowed, handle them in your top-level code.

See the example code.

First, you need a SlobsConnection instance. Connections actively process received messages, so it is important that they be included in the asyncio event loop.

So your main program might look like this:

import asyncio
import logging
from pyslobs import connection

async def main():
    # Note: This assumes an INI file has been configured.
    conn = connection.SlobsConnection()  

    # Give CPU to both your task and the connection instance.
    await asyncio.gather(
            conn.background_processing(),
            do_your_thing(conn)
            )
            
logging.basicConfig(level=logging.DEBUG)
asyncio.run(main())      

Connections should be closed when complete, so the background processing can know to shut-down.

The SlobsConnection needs details about how to connect to the Streamlabs Desktop application.

It needs to know which host is running the application, which port it is listening to, and the "API Token".

The API Token is like a password. It is generated by the application and shared with any tools that want to talk to it. It should not be shared with others; if it is, a new token should be generated.

To obtain all these details, start Streamlabs Desktop, open the settings page, and select "Remote Control". A (blurry) QR code will be displayed. Do not show that QR code on your stream.

If you click on the QR code, and click "Show Details" underneath it, you will be given the following details:

• API token

• Port: This should default to 59650. If it is a different value, your Python application will need to know this.

• IP addresses: If your application is on a different host to Streamlabs Desktop, your Python application will need to know one of the IP addresses offered.

These details can be provided by directly instantiating a ConnectionConfig, and passing it to SlobsConnection().

Alternatively, if none is passed, the SlobsConnection constructor will look for their presence in INI files. The INI file should be stored in the current directory or the user's home directory. They should be called .pyslobs or pyslobs.ini.

The content of the ini file should be:

[connection]
domain=localhost
port=59650
token=<your secret token>

When running the examples or exercises, if no ini file is found, it will assume defaults and prompt for the user to type in the API token each time.

Once you have a connection, it can be used to instantiate any of nine Services:

1. AudioService
2. NotificationsService
3. PerformanceService
4. SceneCollectionsService
5. ScenesService
6. SelectionService
7. SourcesService
8. StreamingService
9. TransitionsService

Services can be used:

• subscribe to events, such as when the user selects a new active scene.
• to make some limited direct changes, such as setting an active scene by scene id.
• fetch SlobsClass instances (see below) that can be manipulated more directly.

In the original API they describe "Classes". These are represented by subclasses of SlobsClass in the Python API.

SlobsClass subclasses include:

1. AudioSource
2. Scene
3. SceneItem
4. SceneItemFolder
5. SceneNode
6. Selection
7. Source

Instances of SlobsClass should only be constructed through Services methods and methods on other instances.

Instances may have properties (such as names) that can be accessed. Be careful that the values of these properties may be out-of-date if the value was changed within the app or if it was changed through a different instance referencing the same StreamLabs Desktop resource.

Objects can be used to fetch other Objects or namedtuples describing other records in the API.

Some Services offer the ability to subscribe to events. A list is provided below.

To subscribe, create a callback function with a signature like:

 async def on_event(key, message)

(A method could also be used, in which case there should be an additional self parameter.)

Then call the event's subscribe() method, passing the callback function. e.g.

subscription = ScenesService(conn).scene_switched.subscribe(on_event)

Each time the event happens, the callback will be called. The key parameter will indicate which sort of event has triggered the callback (which is useful if you want to reuse a call-back function for several event types). The message parameter is a dictionary containing details of the event. (The StreamLabs Desktop API is unclear about what fields will be present, so this may require some experimentation to find out what data is available.)

The result of subscribe is a Subscription object that supports unsubscribe(). (It can also be used as an asynchronous context manager to ensure unsubscriptions are not forgotten.)

The subscribe method also accepts a subscribe_preferences parameter which is an instance of SubscriptionPreferences. It indicates whether the callback should also be called when the subscription is ended due to a call to unsubscribe or by the connection being closed. In these events, the value of key will be set to special values: UNSUBSCRIBED and CLOSED as appropriate.

• SceneCollectionService
  • collection_added
  • collection_removed
  • collection_switched
  • collection_updated
  • collection_will_switch
• ScenesService
  • item_added
  • item_removed
  • item_updated
  • scene_added
  • scene_removed
  • scene_switched
• Sources Service
  • source_added
  • source_removed
  • source_updated
• StreamingService
  • recording_status_change
  • replay_buffer_status_change
  • streaming_status_change
• Transitions Service
  • studio_mode_changed

The examples folder contains many small programs to demonstrate how to use the API. These are not included with the package, but can be found on the GitHub site. Install PySLOBS, copy the raw example files to your machine, start your copy of Streamlabs Desktop, and run the examples (once PySLOBS is installed).

• Sources have an additional field configurable which isn't documented.
• Sources have a method refresh which raises an Internal Server Error. This has been reported to Streamlabs.
• SceneCollectionsService methods sometimes raise Internal Server Errors if called too rapidly.
• TransitionsService methods sometimes raise Internal Server Errors if called too rapidly.
• Notifications subtype may be NEWS, which is not mentioned in the documentation.
• When a source is added, the source_updated callback may also be triggered an arbitrary number of times.
• Commands may raise an asyncio.TimeoutError after 5 seconds if a websocket fails (e.g. a network error) between the time the command was sent and a reply was received.