Skip to content

Instantly share code, notes, and snippets.

@nickrobson
Created October 18, 2016 04:55
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save nickrobson/fbcd09cd755e256c5ea3ac175029e334 to your computer and use it in GitHub Desktop.
Save nickrobson/fbcd09cd755e256c5ea3ac175029e334 to your computer and use it in GitHub Desktop.
Gets your Sonos speaker's queue.
import json
import requests
import xml.etree.cElementTree as XML
SOAP_TEMPLATE = '<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/" s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/"><s:Body>{0}</s:Body></s:Envelope>'
GET_QUEUE_BODY_TEMPLATE = '<u:Browse xmlns:u="urn:schemas-upnp-org:service:ContentDirectory:1"><ObjectID>Q:0</ObjectID><BrowseFlag>BrowseDirectChildren</BrowseFlag><Filter>dc:title,res,dc:creator,upnp:artist,upnp:album,upnp:albumArtURI</Filter><StartingIndex>{0}</StartingIndex><RequestedCount>{1}</RequestedCount><SortCriteria></SortCriteria></u:Browse>'
def cmd(sonos_ip, endpoint, soap_action, body):
headers = {
'Content-Type': 'application/json',
'SOAPACTION': soap_action
}
soap = SOAP_TEMPLATE.format(body)
r = requests.post('http://' + sonos_ip + ':1400' + endpoint, data=soap, headers=headers)
return r.content
def get_queue(sonos_ip, start = 0, max = 100):
endpoint = '/MediaServer/ContentDirectory/Control'
soap = 'urn:schemas-upnp-org:service:ContentDirectory:1#Browse'
body = GET_QUEUE_BODY_TEMPLATE.format(start, max)
res = cmd(sonos_ip, endpoint, soap, body)
queue = []
try:
dom = XML.fromstring(res.encode('utf-8'))
resultText = dom.findtext('.//Result')
if not resultText:
return queue
dom = XML.fromstring(resultText.encode('utf-8'))
for element in dom.findall('.//{urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/}item'):
try:
item = {}
item['uri'] = element[0].text
item['duration'] = element[0].attrib['duration']
item['album_art'] = element[1].text
item['title'] = element[2].text
item['artist'] = element[4].text
item['album'] = element[5].text
queue.append(item)
except:
logger.warning('Could not handle item: %s', element)
except Exception as e:
print e.__name__ + ':', e
return queue
if __name__ == '__main__':
print 'To get your Sonos IP, go into the Sonos Controller app'
print 'Once there, you can go either [Sonos > About My Sonos System] or [Help > About My Sonos System]'
print json.dumps(get_queue(raw_input('Sonos IP: ')))
@nat-goodspeed
Copy link

Thank you for this! I actually have a couple Sonos devices, which naturally surfaced the fact that each potentially has a distinct queue. I was able to use python-zeroconf to discover their IP addresses. This is my edited version.

Also, at least with the data returned by my devices, I couldn't count on every returned element having at least 6 children, so I had to defend against IndexError.

I tried to account for grouping, but I think a proper solution would use the actual Sonos API. Similarly, the Sonos API would presumably get me the user-friendly names of the devices.

# From https://gist.github.com/nickrobson/fbcd09cd755e256c5ea3ac175029e334
# but extended to use the python-zeroconf package to discover Sonos IP
# addresses.

from collections import defaultdict
from contextlib import closing
import logging
from pprint import pprint
import requests
import sys
import time
import xml.etree.cElementTree as XML
from zeroconf import Zeroconf, ServiceBrowser

SOAP_TEMPLATE = """
<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"
 s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/">
  <s:Body>{0}</s:Body>
</s:Envelope>
"""
GET_QUEUE_BODY_TEMPLATE = """
<u:Browse xmlns:u="urn:schemas-upnp-org:service:ContentDirectory:1">
  <ObjectID>Q:0</ObjectID>
  <BrowseFlag>BrowseDirectChildren</BrowseFlag>
  <Filter>dc:title,res,dc:creator,upnp:artist,upnp:album,upnp:albumArtURI</Filter>
  <StartingIndex>{0}</StartingIndex>
  <RequestedCount>{1}</RequestedCount>
  <SortCriteria/>
</u:Browse>"""

class BonjourServiceListener:
    def __init__(self):
        self.hosts = []

    def add_service(self, zeroconf, type, name):
        info = zeroconf.get_service_info(type, name)
        # We only need to collect the first address from each host.
        # info.addresses only contains IPv4 addresses, as byte strings.
        self.hosts.append('.'.join(str(b) for b in info.addresses[0]))

def listen_bonjour_hosts(type='_sonos._tcp.local.', expect=100, timeout=1.0):
    """
    type:    Bonjour/Avahi/zeroconf service type for which to listen
    expect:  stop listening once we've collected this many hosts
    timeout: stop listening after this long regardless of count
    """
    zeroconf = Zeroconf()
    listener = BonjourServiceListener()
    with closing(zeroconf):
        start = time.time()
        end   = start + timeout
        browser = ServiceBrowser(zeroconf, type, listener)
        while len(listener.hosts) < expect and time.time() < end:
            time.sleep(0.05)
    return listener.hosts

def cmd(sonos_ip, endpoint, soap_action, body):

    headers = {
        'Content-Type': 'application/json',
        'SOAPACTION': soap_action
    }

    soap = SOAP_TEMPLATE.format(body)

    r = requests.post('http://' + sonos_ip + ':1400' + endpoint, data=soap, headers=headers)
    return r.content

def get_queue(sonos_ip, start = 0, max = 100):
    endpoint = '/MediaServer/ContentDirectory/Control'
    soap = 'urn:schemas-upnp-org:service:ContentDirectory:1#Browse'
    body = GET_QUEUE_BODY_TEMPLATE.format(start, max)

    res = cmd(sonos_ip, endpoint, soap, body)

    queue = []

    try:
        dom = XML.fromstring(res)
        resultText = dom.findtext('.//Result')
        if not resultText:
            return queue
        dom = XML.fromstring(resultText)
        for element in dom.findall('.//{urn:schemas-upnp-org:metadata-1-0/DIDL-Lite/}item'):
            # Element seems to have no "iterate over direct children"
            # operation, so use indexing -- but beware of short lists.
            item = dict(duration=element[0].get('duration'))
            for idx, key in enumerate(('uri', 'album_art', 'title', None, 'artist', 'album')):
                try:
                    value = element[idx].text
                except IndexError:
                    value = None
                item[key] = value
            queue.append(item)
    except Exception as e:
        logging.warning(e.__class__.__name__ + ':', e)

    return queue

if __name__ == '__main__':
    logging.basicConfig()
    # Without querying the Sonos API, I don't know which of the hosts (Sonos
    # devices) are grouped. Of course all devices in a group share the same
    # queue. So make a dict {queue: set(hosts)} to group them. But since the
    # queue is a list of dicts, and dicts are mutable, have to convert the
    # queue first to a tuple of frozensets of items.
    groups = defaultdict(set)

    for host in listen_bonjour_hosts():
        # Get the queue for this host, convert to immutable key,
        # find-or-create the host set and add this host to it.
        groups[tuple(frozenset(d.items()) for d in get_queue(host, max=2000))].add(host)

    # Now traverse the dict of groups.
    for queue, hosts in groups.items():
        print(' {} '.format(', '.join(hosts)).center(72, '='))
        for idx, songitems in enumerate(queue):
            song = dict(songitems)
            print('  %4d. %s' % ((idx+1), song['title']))

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment