bosesoundtouchapi.ws.soundtouchwebsocket

@export

class SoundTouchWebSocket:

A wrapper class to use the notification system provided by SoundTouch devices.

In order to react to a message, there is a listener system. You can add functions as listener objects. The connection url is defined as follows: 'ws://{Host}:{Port}/'. Port 8080 is used by default.

This class can be used in two ways. First, the object can open a connection through the StartNotification method and secondly, the with-statement can be used to create an instance.

You must first check the device capabilities prior to using the support notifications functionality; if the device does not support it, then it's not going to work! To do this, fetch the device's capabilities and check for a IsWebSocketApiProxyCapable = True value. If true, then the device supports sending notifications; if false, it does not.

For more information and code samples refer to bosesoundtouchapi.soundtouchclient.GetCapabilities method.

Sample Code

# external package imports.
import time
from xml.etree.ElementTree import Element
from xml.etree import ElementTree

# our package imports.
from bosesoundtouchapi import *
from bosesoundtouchapi.models import Capabilities, Volume, NowPlayingStatus
from bosesoundtouchapi.ws import *


class EventHandlerClass:

    def OnSoundTouchInfoEvent(client:SoundTouchClient, args:Element) -> None:
        if (args != None):
            ElementTree.indent(args)  # for pretty printing
            argsEncoded = ElementTree.tostring(args, encoding="unicode")
            print("\n'%s' status update:\n%s" % (client.Device.DeviceName, argsEncoded))


    def OnSoundTouchWebSocketCloseEvent(client:SoundTouchClient, ex:Exception) -> None:
        if (ex != None):
            # args will be an Exception object for websocket close events.
            print("\n'%s' websocket close event:\n%s" % (client.Device.DeviceName, str(ex)))


    def OnSoundTouchWebSocketErrorEvent(client:SoundTouchClient, ex:Exception) -> None:
        if (ex != None):
            # args will be an Exception object for websocket errors.
            print("\n'%s' websocket error:\n%s" % (client.Device.DeviceName, str(ex)))


    def OnSoundTouchUpdateEvent(client:SoundTouchClient, args:Element) -> None:
        if (args != None):
            ElementTree.indent(args)  # for pretty printing
            argsEncoded = ElementTree.tostring(args, encoding="unicode")
            print("\n'%s' status update:\n%s" % (client.Device.DeviceName, argsEncoded))


    def OnSoundTouchUpdateEvent_Volume(client:SoundTouchClient, args:Element) -> None:
        if (args != None):
            ElementTree.indent(args)  # for pretty printing
            argsEncoded = ElementTree.tostring(args, encoding="unicode")
            print("\n'%s' status update:\n%s" % (client.Device.DeviceName, argsEncoded))
            # create Volume configuration model from update event argument.
            config:Volume = Volume(root=args[0])
            print(str(config))


    def OnSoundTouchUpdateEvent_NowPlaying(client:SoundTouchClient, args:Element) -> None:
        if (args != None):
            ElementTree.indent(args)  # for pretty printing
            argsEncoded = ElementTree.tostring(args, encoding="unicode")
            print("\n'%s' status update:\n%s" % (client.Device.DeviceName, argsEncoded))
            # create NowPlayingStatus configuration model from update event argument.
            config:NowPlayingStatus = NowPlayingStatus(root=args[0])
            print(str(config))


try:

    socket:SoundTouchWebSocket = None

    # create SoundTouch device instance.
    device:SoundTouchDevice = SoundTouchDevice("192.168.1.81") # Bose SoundTouch 10
    #device:SoundTouchDevice = SoundTouchDevice("192.168.1.80") # Bose SoundTouch 300

    # create SoundTouch client instance from device.
    client:SoundTouchClient = SoundTouchClient(device)

    # activate recent list cache settings.
    client.UpdateRecentListCacheStatus(True, maxItems=100)

    # get device capabilities - must have IsWebSocketApiProxyCapable=True 
    # in order to support notifications.
    capabilities:Capabilities = client.GetCapabilities()
    if capabilities.IsWebSocketApiProxyCapable:

        # create and start a websocket to receive notifications from the device.
        socket = SoundTouchWebSocket(client)

        # add our listener(s) that will handle SoundTouch device status updates.
        #socket.AddListener(SoundTouchNotifyCategorys.ALL, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.connectionStateUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.criticalErrorUpdate, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.errorNotification, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.errorUpdate, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.groupUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.languageUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.LowPowerStandbyUpdate, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.nameUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.nowPlayingUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.nowSelectionUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.presetsUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.recentsUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.soundTouchConfigurationUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.sourcesUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.swUpdateStatusUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.volumeUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)
        socket.AddListener(SoundTouchNotifyCategorys.zoneUpdated, EventHandlerClass.OnSoundTouchUpdateEvent)

        # you can also add methods for individual models, to make handling them easier.
        # add our listener(s) that will handle SoundTouch device status specific updates.
        socket.AddListener(SoundTouchNotifyCategorys.nowPlayingUpdated, EventHandlerClass.OnSoundTouchUpdateEvent_NowPlaying)
        socket.AddListener(SoundTouchNotifyCategorys.volumeUpdated, EventHandlerClass.OnSoundTouchUpdateEvent_Volume)

        # add our listener(s) that will handle SoundTouch device informational events.
        socket.AddListener(SoundTouchNotifyCategorys.SoundTouchSdkInfo, EventHandlerClass.OnSoundTouchInfoEvent)
        socket.AddListener(SoundTouchNotifyCategorys.userActivityUpdate, EventHandlerClass.OnSoundTouchInfoEvent)

        # add our listener(s) that will handle SoundTouch websocket related events.
        socket.AddListener(SoundTouchNotifyCategorys.WebSocketClose, EventHandlerClass.OnSoundTouchWebSocketCloseEvent)
        socket.AddListener(SoundTouchNotifyCategorys.WebSocketError, EventHandlerClass.OnSoundTouchWebSocketErrorEvent)

        # start receiving updates.
        socket.StartNotification()

        print("** Notification event loop has started.")
        print("** Try pressing some buttons on your SoundTouch remote or device ...")

        # for testing status notifications.
        maxcnt:int = 1200  # 1200=20 mins, 300=5 mins
        for i in range(maxcnt):

            # wait 1 second.
            time.sleep(1)

            # did we lose the connection to the SoundTouch device?
            # if so, then stop / restart the notification event thread.
            if socket.IsThreadRunForeverActive == False:
                print("socket.IsThreadRunForeverActive = %s - restarting notification thread ..." % (str(socket.IsThreadRunForeverActive)))
                socket.StopNotification()
                socket.StartNotification()

    else:

        print("SoundTouch device '%s' does not support Bose WebSocket API.")

except Exception as ex:

    print(str(ex))
    raise

finally:

    # stop listening for Bose SoundTouch status updates.
    if (socket != None):
        socket.StopNotification()
        socket.ClearListeners()

SoundTouchWebSocket( client: bosesoundtouchapi.soundtouchclient.SoundTouchClient, port: int = 8080, pingInterval: int = 0)

Initializes a new instance of the class.

Arguments:

client (SoundTouchClient): A SoundTouchClient instance to receive status notifications from.
port (int): The port that the SoundTouch WebAPI socket is posting notifications on.
Default is 8080.
pingInterval (int): Interval (in seconds) to send 'KeepAlive' ping request to the SoundTouch WebSocket, if websocket support is enabled for the SoundTouch device. Set this value to zero to disable keepalive ping requests.
Default is 0 (disabled).

Client: bosesoundtouchapi.soundtouchclient.SoundTouchClient

A SoundTouchClient instance to receive status notifications from.

IsThreadRunForeverActive: bool

True if the websocket run_forever() loop is still executing; otherwise, False.

A value of False can indicate that an exception has occured that caused the run_forever() loop to exit.

PingInterval: int

Interval (in seconds) to send 'KeepAlive' ping request to the SoundTouch WebSocket, if websocket support is enabled for the SoundTouch device.

Default is 60.
If zero, then keepalive requests are disabled.

Port: int

The port that the SoundTouch WebAPI socket is posting notifications on.

Default is 8080.

def AddListener( self, category: bosesoundtouchapi.soundtouchnotifycategorys.SoundTouchNotifyCategorys, listener) -> bool:

Adds a listener provided here to the given category.

Since there are different types of events, the category-string can be used to add a listener to a specific notification category. The listener must take only two arguments: the SoundTouchClient instance, and an xml.etree.ElementTree.Element that contains the status update itself.

Arguments:

category (SoundTouchNotifyCategorys): The category this listener should be added to.
Use one of the pre-defined SoundTouchNotifyCategorys values to receive updates for that category (e.g. "volumeUpdated", "nowPlayingUpdated", etc).
Use SoundTouchNotifyCategorys.WebSocketError to receive notifications for WebSocket errors.
Use SoundTouchNotifyCategorys.ALL to receive notifications for any type of update event.
listener (object): A simple listener method which takes the XML-Element as a passed argument.

Returns:

True if the listener was added successfully; otherwise, False.

Please refer to the bosesoundtouchapi.soundtouchnotifycategorys.SoundTouchNotifyCategorys class for more details on what events are raised and why / when they happen.

def ClearListeners(self) -> None:

Removes all listeners that were previously added.

def GetListenerGroup(self, category: str) -> list:

Searches for a specific category in the registered ones.

This method is a convenient method to return all listeners that were added to a specific context.

Arguments:

category (str): The category, which has to be one of the SoundTouchNotifyCategorys values.

Returns:

A list containing all listeners linked to the given category.

def NotifyListeners(self, category: str, event: xml.etree.ElementTree.Element) -> None:

Notifies all listeners that were stored in the given context.

The name of each context is defined to be the tag element of the update XML-Element.

Arguments:

category (str): The category of which listeners should be notified from.
event (xmltree.Element): The event represents either an XML-Element with event.tag == category, or an Exception type if category = SoundTouchNotifyCategorys.WebSocketError.

def RemoveListener( self, category: bosesoundtouchapi.soundtouchnotifycategorys.SoundTouchNotifyCategorys, listener) -> bool:

Removes a listener from the given category.

Arguments:

category (str): The category this listener should be removed from.
Use one of the pre-defined SoundTouchNotifyCategorys values to receive updates for that category (e.g. "volumeUpdated", "nowPlayingUpdated", etc).
listener (object): A simple listener method which takes the XML-Element as a passed argument.

Returns:

True if the listener was removed successfully; otherwise, False.

def StartNotification(self) -> None:

Creates and starts a web socket event loop thread that listens for notifications from the SoundTouch device.

Only one web socket event loop thread will be started for the device.

def StopNotification(self) -> None:

Stops the web socket event loop thread, if one was previously started using the StartNotification method.

This method does nothing if the event loop thread was not started.

def ToString(self) -> str:

Returns a displayable string representation of the class.