Source code for

# pylint: disable=fixme, invalid-name

# Disable while we have Python 2.x compatability
# pylint: disable=useless-object-inheritance

"""Classes representing Sonos UPnP services.

>>> import soco
>>> device = soco.SoCo('')
>>> print(RenderingControl(device).GetMute([('InstanceID', 0),
...     ('Channel', 'Master')]))
{'CurrentMute': '0'}
>>> r = ContentDirectory(device).Browse([
...    ('ObjectID', 'Q:0'),
...    ('BrowseFlag', 'BrowseDirectChildren'),
...    ('Filter', '*'),
...    ('StartingIndex', '0'),
...    ('RequestedCount', '100'),
...    ('SortCriteria', '')
...    ])
>>> print(r['Result'])
<?xml version="1.0" ?><DIDL-Lite xmlns="urn:schemas-upnp-org:metadata ...
>>> for action, in_args, out_args in AlarmClock(device).iter_actions():
...    print(action, in_args, out_args)
SetFormat [Argument(name='DesiredTimeFormat', vartype='string'), Argument(
name='DesiredDateFormat', vartype='string')] []
GetFormat [] [Argument(name='CurrentTimeFormat', vartype='string'),
Argument(name='CurrentDateFormat', vartype='string')] ...

# UPnP Spec at

import logging
from collections import namedtuple
from xml.sax.saxutils import escape

import requests

from .cache import Cache
from . import events
from . import config
from .exceptions import NotSupportedException, SoCoUPnPException, UnknownSoCoException
from .utils import prettify
from .xml import XML, illegal_xml_re

# UPnP requires all XML to be transmitted/received with utf-8 encoding. All
# strings used in this module are unicode. The Requests library should take
# care of all of the necessary encoding (on sending) and decoding (on
# receiving) for us, provided that we specify the correct encoding headers
# (which, hopefully, we do).
# But since ElementTree seems to prefer being fed bytes to unicode, at least
# for Python 2.x, we have to encode strings specifically before using it. see
# TODO: Keep an eye on this when it comes to
# Python 3 compatibility

log = logging.getLogger(__name__)  # pylint: disable=C0103
# logging.basicConfig()
# log.setLevel(logging.INFO)

if config.EVENTS_MODULE is None:
    config.EVENTS_MODULE = events

[docs]class Action(namedtuple("ActionBase", "name, in_args, out_args")): """A UPnP Action and its arguments.""" def __str__(self): args = ", ".join(str(arg) for arg in self.in_args) returns = ", ".join(str(arg) for arg in self.out_args) return "{0}({1}) -> {{{2}}}".format(, args, returns)
[docs]class Argument(namedtuple("ArgumentBase", "name, vartype")): """A UPnP Argument and its type.""" def __str__(self): argument = if self.vartype.default: argument = "{}={}".format(, self.vartype.default) return "{}: {}".format(argument, str(self.vartype))
[docs]class Vartype(namedtuple("VartypeBase", "datatype, default, list, range")): """An argument type with default value and range.""" def __str__(self): if self.list: return "[{}]".format(", ".join(self.list)) if self.range: return "[{}..{}]".format(self.range[0], self.range[1]) return self.datatype
[docs]class Service: """A class representing a UPnP service. This is the base class for all Sonos Service classes. This class has a dynamic method dispatcher. Calls to methods which are not explicitly defined here are dispatched automatically to the service action with the same name. """ # pylint: disable=bad-continuation soap_body_template = ( '<?xml version="1.0"?>' '<s:Envelope xmlns:s=""' ' s:encodingStyle="">' "<s:Body>" '<u:{action} xmlns:u="urn:schemas-upnp-org:service:' '{service_type}:{version}">' "{arguments}" "</u:{action}>" "</s:Body>" "</s:Envelope>" ) # noqa PEP8 def __init__(self, soco): """ Args: soco (SoCo): A `SoCo` instance to which the UPnP Actions will be sent """ #: `SoCo`: The `SoCo` instance to which UPnP Actions are sent self.soco = soco # Some defaults. Some or all these will need to be overridden # specifically in a sub-class. There is other information we could # record, but this will do for the moment. Info about a Sonos device is # available at <IP_address>/xml/device_description.xml in the # <service> tags #: str: The UPnP service type. self.service_type = self.__class__.__name__ #: str: The UPnP service version. self.version = 1 self.service_id = self.service_type #: str: The base URL for sending UPnP Actions. self.base_url = "http://{}:1400".format(self.soco.ip_address) #: str: The UPnP Control URL. self.control_url = "/{}/Control".format(self.service_type) #: str: The service control protocol description URL. self.scpd_url = "/xml/{}{}.xml".format(self.service_type, self.version) #: str: The service eventing subscription URL. self.event_subscription_url = "/{}/Event".format(self.service_type) #: A cache for storing the result of network calls. By default, this is #: a `TimedCache` with a default timeout=0. self.cache = Cache(default_timeout=0) # Caching variables for actions and event_vars, will be filled when # they are requested for the first time self._actions = None self._event_vars = None # From table 3.3 in # # This list may not be complete, but should be good enough to be going # on with. Error codes between 700-799 are defined for particular # services, and may be overriden in subclasses. Error codes >800 # are generally SONOS specific. NB It may well be that SONOS does not # use some of these error codes. # pylint: disable=invalid-name self.UPNP_ERRORS = { 400: "Bad Request", 401: "Invalid Action", 402: "Invalid Args", 404: "Invalid Var", 412: "Precondition Failed", 501: "Action Failed", 600: "Argument Value Invalid", 601: "Argument Value Out of Range", 602: "Optional Action Not Implemented", 603: "Out Of Memory", 604: "Human Intervention Required", 605: "String Argument Too Long", 606: "Action Not Authorized", 607: "Signature Failure", 608: "Signature Missing", 609: "Not Encrypted", 610: "Invalid Sequence", 611: "Invalid Control URL", 612: "No Such Session", } self.DEFAULT_ARGS = {} def __getattr__(self, action): """Called when a method on the instance cannot be found. Causes an action to be sent to UPnP server. See also `object.__getattr__`. Args: action (str): The name of the unknown method. Returns: callable: The callable to be invoked. . """ # Define a function to be invoked as the method, which calls # send_command. def _dispatcher(self, *args, **kwargs): """Dispatch to send_command.""" return self.send_command(action, *args, **kwargs) # rename the function so it appears to be the called method. We # probably don't need this, but it doesn't harm _dispatcher.__name__ = action # _dispatcher is now an unbound menthod, but we need a bound method. # This turns an unbound method into a bound method (i.e. one that # takes self - an instance of the class - as the first parameter) # pylint: disable=no-member method = _dispatcher.__get__(self, self.__class__) # Now we have a bound method, we cache it on this instance, so that # next time we don't have to go through this again setattr(self, action, method) log.debug("Dispatching method %s", action) # return our new bound method, which will be called by Python return method
[docs] @staticmethod def wrap_arguments(args=None): """Wrap a list of tuples in xml ready to pass into a SOAP request. Args: args (list): a list of (name, value) tuples specifying the name of each argument and its value, eg ``[('InstanceID', 0), ('Speed', 1)]``. The value can be a string or something with a string representation. The arguments are escaped and wrapped in <name> and <value> tags. Example: >>> from soco import SoCo >>> device = SoCo('') >>> s = Service(device) >>> print(s.wrap_arguments([('InstanceID', 0), ('Speed', 1)])) <InstanceID>0</InstanceID><Speed>1</Speed>' """ if args is None: args = [] tags = [] for name, value in args: tag = "<{name}>{value}</{name}>".format( name=name, value=escape("%s" % value, {'"': "&quot;"}) ) # % converts to unicode because we are using unicode literals. # Avoids use of 'unicode' function which does not exist in python 3 tags.append(tag) xml = "".join(tags) return xml
[docs] @staticmethod def unwrap_arguments(xml_response): """Extract arguments and their values from a SOAP response. Args: xml_response (str): SOAP/xml response text (unicode, not utf-8). Returns: dict: a dict of ``{argument_name: value}`` items. """ # A UPnP SOAP response (including headers) looks like this: # HTTP/1.1 200 OK # CONTENT-LENGTH: bytes in body # CONTENT-TYPE: text/xml; charset="utf-8" DATE: when response was # generated # EXT: # SERVER: OS/version UPnP/1.0 product/version # # <?xml version="1.0"?> # <s:Envelope # xmlns:s="" # s:encodingStyle=""> # <s:Body> # <u:actionNameResponse # xmlns:u="urn:schemas-upnp-org:service:serviceType:v"> # <argumentName>out arg value</argumentName> # ... other out args and their values go here, if any # </u:actionNameResponse> # </s:Body> # </s:Envelope> # Get all tags in order. Elementree (in python 2.x) seems to prefer to # be fed bytes, rather than unicode xml_response = xml_response.encode("utf-8") try: tree = XML.fromstring(xml_response) except XML.ParseError: # Try to filter illegal xml chars (as unicode), in case that is # the reason for the parse error filtered = illegal_xml_re.sub("", xml_response.decode("utf-8")).encode( "utf-8" ) tree = XML.fromstring(filtered) # Get the first child of the <Body> tag which will be # <{actionNameResponse}> (depends on what actionName is). Turn the # children of this into a {tagname, content} dict. XML unescaping # is carried out for us by elementree. action_response = tree.find("{}Body")[ 0 ] return {i.tag: i.text or "" for i in action_response}
[docs] def compose_args(self, action_name, in_argdict): """Compose the argument list from an argument dictionary, with respect for default values. Args: action_name (str): The name of the action to be performed. in_argdict (dict): Arguments as a dict, e.g. ``{'InstanceID': 0, 'Speed': 1}``. The values can be a string or something with a string representation. Returns: list: a list of ``(name, value)`` tuples. Raises: AttributeError: If this service does not support the action. ValueError: If the argument lists do not match the action signature. """ for action in self.actions: if == action_name: # The found 'action' will be visible from outside the loop break else: raise AttributeError("Unknown Action: {}".format(action_name)) # Check for given argument names which do not occur in the expected # argument list # pylint: disable=undefined-loop-variable unexpected = set(in_argdict) - { for argument in action.in_args} if unexpected: raise ValueError( "Unexpected argument '{}'. Method signature: {}".format( next(iter(unexpected)), str(action) ) ) # List the (name, value) tuples for each argument in the argument list composed = [] for argument in action.in_args: name = if name in in_argdict: composed.append((name, in_argdict[name])) continue if name in self.DEFAULT_ARGS: composed.append((name, self.DEFAULT_ARGS[name])) continue if argument.vartype.default is not None: composed.append((name, argument.vartype.default)) raise ValueError( "Missing argument '{}'. Method signature: {}".format(, str(action) ) ) return composed
[docs] def build_command(self, action, args=None): """Build a SOAP request. Args: action (str): the name of an action (a string as specified in the service description XML file) to be sent. args (list, optional): Relevant arguments as a list of (name, value) tuples. Returns: tuple: a tuple containing the POST headers (as a dict) and a string containing the relevant SOAP body. Does not set content-length, or host headers, which are completed upon sending. """ # A complete request should look something like this: # POST path of control URL HTTP/1.1 # HOST: host of control URL:port of control URL # CONTENT-LENGTH: bytes in body # CONTENT-TYPE: text/xml; charset="utf-8" # SOAPACTION: "urn:schemas-upnp-org:service:serviceType:v#actionName" # # <?xml version="1.0"?> # <s:Envelope # xmlns:s="" # s:encodingStyle=""> # <s:Body> # <u:actionName # xmlns:u="urn:schemas-upnp-org:service:serviceType:v"> # <argumentName>in arg value</argumentName> # ... other in args and their values go here, if any # </u:actionName> # </s:Body> # </s:Envelope> arguments = self.wrap_arguments(args) body = self.soap_body_template.format( arguments=arguments, action=action, service_type=self.service_type, version=self.version, ) soap_action_template = ( "urn:schemas-upnp-org:service:{service_type}:{version}#{action}" ) soap_action = soap_action_template.format( service_type=self.service_type, version=self.version, action=action ) headers = { "Content-Type": 'text/xml; charset="utf-8"', "SOAPACTION": soap_action, } # Note that although we set the charset to utf-8 here, in fact the # body is still unicode. It will only be converted to bytes when it # is set over the network return (headers, body)
[docs] def send_command(self, action, args=None, cache=None, cache_timeout=None, **kwargs): """Send a command to a Sonos device. Args: action (str): the name of an action (a string as specified in the service description XML file) to be sent. args (list, optional): Relevant arguments as a list of (name, value) tuples, as an alternative to ``kwargs``. cache (Cache): A cache is operated so that the result will be stored for up to ``cache_timeout`` seconds, and a subsequent call with the same arguments within that period will be returned from the cache, saving a further network call. The cache may be invalidated or even primed from another thread (for example if a UPnP event is received to indicate that the state of the Sonos device has changed). If ``cache_timeout`` is missing or `None`, the cache will use a default value (which may be 0 - see :attr:``). By default, the cache identified by the service's :attr:`` attribute will be used, but a different cache object may be specified in the ``cache`` parameter. kwargs: Relevant arguments for the command. Returns: dict: a dict of ``{argument_name, value}`` items. Raises: AttributeError: If this service does not support the action. ValueError: If the argument lists do not match the action signature. `SoCoUPnPException`: if a SOAP error occurs. `UnknownSoCoException`: if an unknown UPnP error occurs. `requests.exceptions.HTTPError`: if an http error occurs. """ # Determine the timeout for the request: use the value of # config.REQUEST_TIMEOUT unless overridden by 'timeout' # being provided as a kwarg by the caller, in which case # use this and remove it from kwargs. timeout = kwargs.pop("timeout", config.REQUEST_TIMEOUT) log.debug("Request timeout set to %s", timeout) if args is None: args = self.compose_args(action, kwargs) if cache is None: cache = self.cache result = cache.get(action, args) if result is not None: log.debug("Cache hit") return result # Cache miss, so go ahead and make a network call headers, body = self.build_command(action, args) log.debug("Sending %s %s to %s", action, args, self.soco.ip_address) log.debug("Sending %s, %s", headers, prettify(body)) # Convert the body to bytes, and send it. response = self.base_url + self.control_url, headers=headers, data=body.encode("utf-8"), timeout=timeout, ) log.debug("Received %s, %s", response.headers, response.text) status = response.status_code log.debug("Received status %s from %s", status, self.soco.ip_address) if status == 200: # The response is good. Get the output params, and return them. # NB an empty dict is a valid result. It just means that no # params are returned. By using response.text, we rely upon # the requests library to convert to unicode for us. result = self.unwrap_arguments(response.text) or True # Store in the cache. There is no need to do this if there was an # error, since we would want to try a network call again. cache.put(result, action, args, timeout=cache_timeout) return result elif status == 405: raise NotSupportedException( "{} not supported on {}".format(action, self.soco.ip_address) ) elif status == 500: # Internal server error. UPnP requires this to be returned if the # device does not like the action for some reason. The returned # content will be a SOAP Fault. Parse it and raise an error. self.handle_upnp_error(response.text) else: # Something else has gone wrong. Probably a network error. Let # Requests handle it response.raise_for_status() return None
[docs] def handle_upnp_error(self, xml_error): """Disect a UPnP error, and raise an appropriate exception. Args: xml_error (str): a unicode string containing the body of the UPnP/SOAP Fault response. Raises an exception containing the error code. """ # An error code looks something like this: # HTTP/1.1 500 Internal Server Error # CONTENT-LENGTH: bytes in body # CONTENT-TYPE: text/xml; charset="utf-8" # DATE: when response was generated # EXT: # SERVER: OS/version UPnP/1.0 product/version # <?xml version="1.0"?> # <s:Envelope # xmlns:s="" # s:encodingStyle=""> # <s:Body> # <s:Fault> # <faultcode>s:Client</faultcode> # <faultstring>UPnPError</faultstring> # <detail> # <UPnPError xmlns="urn:schemas-upnp-org:control-1-0"> # <errorCode>error code</errorCode> # <errorDescription>error string</errorDescription> # </UPnPError> # </detail> # </s:Fault> # </s:Body> # </s:Envelope> # # All that matters for our purposes is the errorCode. # errorDescription is not required, and Sonos does not seem to use it. # NB need to encode unicode strings before passing to ElementTree xml_error = xml_error.encode("utf-8") error = XML.fromstring(xml_error) log.debug("Error %s", xml_error) error_code = error.findtext(".//{urn:schemas-upnp-org:control-1-0}errorCode") if error_code is not None: description = self.UPNP_ERRORS.get(int(error_code), "") raise SoCoUPnPException( message="UPnP Error {} received: {} from {}".format( error_code, description, self.soco.ip_address ), error_code=error_code, error_description=description, error_xml=xml_error, ) # Unknown error, so just return the entire response log.error("Unknown error received from %s", self.soco.ip_address) raise UnknownSoCoException(xml_error)
[docs] def subscribe( self, requested_timeout=None, auto_renew=False, event_queue=None, strict=True ): """Subscribe to the service's events. Args: requested_timeout (int, optional): If requested_timeout is provided, a subscription valid for that number of seconds will be requested, but not guaranteed. Check :attr:`` on return to find out what period of validity is actually allocated. auto_renew (bool): If auto_renew is `True`, the subscription will automatically be renewed just before it expires, if possible. Default is `False`. event_queue (:class:`~queue.Queue`): a thread-safe queue object on which received events will be put. If not specified, a (:class:`~queue.Queue`) will be created and used. strict (bool, optional): If True and an Exception occurs during execution, the Exception will be raised or, if False, the Exception will be logged and the Subscription instance will be returned. Default `True`. Returns: :class:``: an instance of :class:``, representing the new subscription. If config.EVENTS_MODULE has been set to refer to :py:mod:`events_twisted`, a deferred will be returned with the Subscription as its result and deferred.subscription will be set to refer to the Subscription. To unsubscribe, call the :meth:`` method on the returned object. """ subscription = config.EVENTS_MODULE.Subscription(self, event_queue) return subscription.subscribe( requested_timeout=requested_timeout, auto_renew=auto_renew, strict=strict )
def _update_cache_on_event(self, event): """Update the cache when an event is received. This will be called before an event is put onto the event queue. Events will often indicate that the Sonos device's state has changed, so this opportunity is made available for the service to update its cache. The event will be put onto the event queue once this method returns. `event` is an Event namedtuple: ('sid', 'seq', 'service', 'variables') .. warning:: This method will not be called from the main thread but by one or more threads, which handle the events as they come in. You *must not* access any class, instance or global variables without appropriate locks. Treat all parameters passed to this method as read only. """ @property def actions(self): """The service's actions with their arguments. Returns: list(`Action`): A list of Action namedtuples, consisting of action_name (str), in_args (list of Argument namedtuples, consisting of name and argtype), and out_args (ditto). The return value looks like this: .. code-block:: python [ Action( name='GetMute', in_args=[ Argument(name='InstanceID', ...), Argument( name='Channel', vartype='string', list=['Master', 'LF', 'RF', 'SpeakerOnly'], range=None ) ], out_args=[ Argument(name='CurrentMute, ...) ] ) Action(...) ] Its string representation will look like this: .. code-block:: text GetMute(InstanceID: ui4, Channel: [Master, LF, RF, SpeakerOnly])\n -> {CurrentMute: boolean} """ if self._actions is None: self._actions = list(self.iter_actions()) return self._actions
[docs] def iter_actions(self): """Yield the service's actions with their arguments. Yields: `Action`: the next action. Each action is an Action namedtuple, consisting of action_name (a string), in_args (a list of Argument namedtuples consisting of name and argtype), and out_args (ditto), eg:: Action( name='SetFormat', in_args=[ Argument(name='DesiredTimeFormat', vartype=<Vartype>), Argument(name='DesiredDateFormat', vartype=<Vartype>)], out_args=[] ) """ # pylint: disable=invalid-name ns = "{urn:schemas-upnp-org:service-1-0}" # get the scpd body as bytes, and feed directly to elementtree # which likes to receive bytes scpd_body = requests.get(self.base_url + self.scpd_url, timeout=10).content tree = XML.fromstring(scpd_body) # parse the state variables to get the relevant variable types vartypes = {} srvStateTables = tree.findall("{}serviceStateTable".format(ns)) for srvStateTable in srvStateTables: statevars = srvStateTable.findall("{}stateVariable".format(ns)) for state in statevars: name = state.findtext("{}name".format(ns)) datatype = state.findtext("{}dataType".format(ns)) default = state.findtext("{}defaultValue".format(ns)) value_list_elt = state.find("{}allowedValueList".format(ns)) or () value_list = [item.text for item in value_list_elt] or None value_range_elt = state.find("{}allowedValueRange".format(ns)) or () value_range = [item.text for item in value_range_elt] or None vartypes[name] = Vartype(datatype, default, value_list, value_range) # find all the actions actionLists = tree.findall("{}actionList".format(ns)) for actionList in actionLists: actions = actionList.findall("{}action".format(ns)) for i in actions: action_name = i.findtext("{}name".format(ns)) argLists = i.findall("{}argumentList".format(ns)) for argList in argLists: args_iter = argList.findall("{}argument".format(ns)) in_args = [] out_args = [] for arg in args_iter: arg_name = arg.findtext("{}name".format(ns)) direction = arg.findtext("{}direction".format(ns)) related_variable = arg.findtext( "{}relatedStateVariable".format(ns) ) vartype = vartypes[related_variable] if direction == "in": in_args.append(Argument(arg_name, vartype)) else: out_args.append(Argument(arg_name, vartype)) yield Action(action_name, in_args, out_args)
@property def event_vars(self): """The service's eventable variables. Returns: list(tuple): A list of (variable name, data type) tuples. """ if self._event_vars is None: self._event_vars = list(self.iter_event_vars()) return self._event_vars
[docs] def iter_event_vars(self): """Yield the services eventable variables. Yields: `tuple`: a tuple of (variable name, data type). """ # pylint: disable=invalid-name ns = "{urn:schemas-upnp-org:service-1-0}" scpd_body = requests.get(self.base_url + self.scpd_url, timeout=10).text tree = XML.fromstring(scpd_body.encode("utf-8")) # parse the state variables to get the relevant variable types statevars = tree.findall("{}stateVariable".format(ns)) for state in statevars: # We are only interested if 'sendEvents' is 'yes', i.e this # is an eventable variable if state.attrib["sendEvents"] == "yes": name = state.findtext("{}name".format(ns)) vartype = state.findtext("{}dataType".format(ns)) yield (name, vartype)
[docs]class AlarmClock(Service): """Sonos alarm service, for setting and getting time and alarms.""" def __init__(self, soco): super().__init__(soco) self.UPNP_ERRORS.update( { 801: "Already an alarm for this time", } )
[docs]class MusicServices(Service): """Sonos music services service, for functions related to 3rd party music services."""
[docs]class AudioIn(Service): """Sonos audio in service, for functions related to RCA audio input."""
[docs]class DeviceProperties(Service): """Sonos device properties service, for functions relating to zones, LED state, stereo pairs etc."""
[docs]class SystemProperties(Service): """Sonos system properties service, for functions relating to authentication etc."""
[docs]class ZoneGroupTopology(Service): """Sonos zone group topology service, for functions relating to network topology, diagnostics and updates."""
[docs]class GroupManagement(Service): """Sonos group management service, for services relating to groups."""
[docs]class QPlay(Service): """Sonos Tencent QPlay service (a Chinese music service)"""
[docs]class ContentDirectory(Service): """UPnP standard Content Directory service, for functions relating to browsing, searching and listing available music.""" def __init__(self, soco): super().__init__(soco) self.control_url = "/MediaServer/ContentDirectory/Control" self.event_subscription_url = "/MediaServer/ContentDirectory/Event" # For error codes, see table 2.7.16 in # self.UPNP_ERRORS.update( { 701: "No such object", 702: "Invalid CurrentTagValue", 703: "Invalid NewTagValue", 704: "Required tag", 705: "Read only tag", 706: "Parameter Mismatch", 708: "Unsupported or invalid search criteria", 709: "Unsupported or invalid sort criteria", 710: "No such container", 711: "Restricted object", 712: "Bad metadata", 713: "Restricted parent object", 714: "No such source resource", 715: "Resource access denied", 716: "Transfer busy", 717: "No such file transfer", 718: "No such destination resource", 719: "Destination resource access denied", 720: "Cannot process the request", } )
[docs]class MS_ConnectionManager(Service): # pylint: disable=invalid-name """UPnP standard connection manager service for the media server.""" def __init__(self, soco): super().__init__(soco) self.service_type = "ConnectionManager" self.control_url = "/MediaServer/ConnectionManager/Control" self.event_subscription_url = "/MediaServer/ConnectionManager/Event"
[docs]class RenderingControl(Service): """UPnP standard rendering control service, for functions relating to playback rendering, eg bass, treble, volume and EQ.""" def __init__(self, soco): super().__init__(soco) self.control_url = "/MediaRenderer/RenderingControl/Control" self.event_subscription_url = "/MediaRenderer/RenderingControl/Event" self.DEFAULT_ARGS.update({"InstanceID": 0})
[docs]class MR_ConnectionManager(Service): # pylint: disable=invalid-name """UPnP standard connection manager service for the media renderer.""" def __init__(self, soco): super().__init__(soco) self.service_type = "ConnectionManager" self.control_url = "/MediaRenderer/ConnectionManager/Control" self.event_subscription_url = "/MediaRenderer/ConnectionManager/Event"
[docs]class AVTransport(Service): """UPnP standard AV Transport service, for functions relating to transport management, eg play, stop, seek, playlists etc.""" def __init__(self, soco): super().__init__(soco) self.control_url = "/MediaRenderer/AVTransport/Control" self.event_subscription_url = "/MediaRenderer/AVTransport/Event" # For error codes, see # self.UPNP_ERRORS.update( { 701: "Transition not available", 702: "No contents", 703: "Read error", 704: "Format not supported for playback", 705: "Transport is locked", 706: "Write error", 707: "Media is protected or not writeable", 708: "Format not supported for recording", 709: "Media is full", 710: "Seek mode not supported", 711: "Illegal seek target", 712: "Play mode not supported", 713: "Record quality not supported", 714: "Illegal MIME-Type", 715: 'Content "BUSY"', 716: "Resource Not found", 717: "Play speed not supported", 718: "Invalid InstanceID", 737: "No DNS Server", 738: "Bad Domain Name", 739: "Server Error", } ) self.DEFAULT_ARGS.update({"InstanceID": 0})
[docs]class Queue(Service): """Sonos queue service, for functions relating to queue management, saving queues etc.""" def __init__(self, soco): super().__init__(soco) self.control_url = "/MediaRenderer/Queue/Control" self.event_subscription_url = "/MediaRenderer/Queue/Event"
[docs]class GroupRenderingControl(Service): """Sonos group rendering control service, for functions relating to group volume etc.""" def __init__(self, soco): super().__init__(soco) self.control_url = "/MediaRenderer/GroupRenderingControl/Control" self.event_subscription_url = "/MediaRenderer/GroupRenderingControl/Event" self.DEFAULT_ARGS.update({"InstanceID": 0})