Example Code

1. Get notified about every device Coherence detects on the network
2. Get notified via the ControlPoint about every MediaRenderer found and …
3. React on each MediaServer and browse its "root"-container
4. React on each InternetGatewayDevice and print some information about it

Get notified about every device Coherence detects on the network

from twisted.internet import reactor
from coherence.base import Coherence

def check_device(device):
    print "found device %s of type %s - %r" %(device.get_friendly_name(),
                                              device.get_device_type(),
                                              device.client)

def start():
    config = {'logmode':'warning'}
    c = Coherence(config)
    c.connect(check_device, 'Coherence.UPnP.Device.detection_completed')

reactor.callWhenRunning(start)
reactor.run()

Get notified via the ControlPoint about every MediaRenderer found and subscribe to some of its events

from twisted.internet import reactor
from coherence.base import Coherence
from coherence.upnp.devices.control_point import ControlPoint

def state_variable_change(variable):
    if variable.name == 'CurrentTrackMetaData':

        if variable.value != None and len(variable.value)>0:
            try:
                from coherence.upnp.core import DIDLLite
                elt = DIDLLite.DIDLElement.fromString(variable.value)
                for item in elt.getItems():
                    print "now playing: %r - %r (%s/%r)" % (item.artist, item.title, item.id, item.upnp_class)
            except SyntaxError:
                #print "seems we haven't got an XML string"
                return
    elif variable.name == 'TransportState':
        print variable.name, 'changed from', variable.old_value, 'to', variable.value

def media_renderer_found(client,udn):
    print "media_renderer_found", client
    print "media_renderer_found", client.device.get_friendly_name()
    client.av_transport.subscribe_for_variable('CurrentTrackMetaData', state_variable_change)
    client.av_transport.subscribe_for_variable('TransportState', state_variable_change)

def media_renderer_removed(udn):
    print "media_renderer_removed", udn

def start():
    control_point = ControlPoint(Coherence({'logmode':'warning'}),
                                 auto_client=['MediaRenderer'])
    control_point.connect(media_renderer_found, 'Coherence.UPnP.ControlPoint.MediaRenderer.detected')
    control_point.connect(media_renderer_removed, 'Coherence.UPnP.ControlPoint.MediaRenderer.removed')


reactor.callWhenRunning(start)
reactor.run()

React on each MediaServer and browse its "root"-container

as testable in source:trunk/Coherence/misc/media_server_observer.py :

from twisted.internet import reactor

from coherence.base import Coherence
from coherence.upnp.devices.control_point import ControlPoint
from coherence.upnp.core import DIDLLite

# browse callback
def process_media_server_browse(result, client):
    print "browsing root of", client.device.get_friendly_name()
    print "result contains %d out of %d total matches" % \
            (int(result['NumberReturned']), int(result['TotalMatches']))

    elt = DIDLLite.DIDLElement.fromString(result['Result'])
    for item in elt.getItems():

        if item.upnp_class.startswith("object.container"):
            print "  container %s (%s) with %d items" % \
                    (item.title,item.id, item.childCount)

        if item.upnp_class.startswith("object.item"):
            print "  item %s (%s)" % (item.title, item.id)

# called for each media server found
def media_server_found(client, udn):
    print "media_server_found", client
    print "media_server_found", client.device.get_friendly_name()

    d = client.content_directory.browse(0,
            browse_flag='BrowseDirectChildren', process_result=False,
            backward_compatibility=False)
    d.addCallback(process_media_server_browse, client)

# sadly they sometimes get removed as well :(
def media_server_removed(udn):
    print "media_server_removed", udn

def start():
    control_point = ControlPoint(Coherence({'logmode':'warning'}),
            auto_client=['MediaServer'])
    control_point.connect(media_server_found,
            'Coherence.UPnP.ControlPoint.MediaServer.detected')
    control_point.connect(media_server_removed,
            'Coherence.UPnP.ControlPoint.MediaServer.removed')

if __name__ == "__main__":
    reactor.callWhenRunning(start)
    reactor.run()

React on each InternetGatewayDevice and print some information about it

from coherence.base import Coherence
from coherence.upnp.devices.control_point import ControlPoint

from twisted.internet import reactor

def state_variable_change(variable):
    if variable.name == 'ExternalIPAddress':
        print "our external IP address is %r" % variable.value
    elif variable.name == 'PortMappingNumberOfEntries':
        if variable.value != '':
            print "there are %d port-mappings defined" % int(variable.value)
        else:
            print "there are no port-mappings defined"


def igd_found(client,udn):
    print "IGD found", client.device.get_friendly_name()
    wan_ip_connection_service = client.wan_device.wan_connection_device.wan_ip_connection
    wan_ip_connection_service.subscribe_for_variable('PortMappingNumberOfEntries', callback=state_variable_change)
    wan_ip_connection_service.subscribe_for_variable('ExternalIPAddress', callback=state_variable_change)

def igd_removed(udn):
    print "igd_removed", udn

def start():
    config = {}
    config['logmode'] = 'warning'

    control_point = ControlPoint(Coherence(config),auto_client=['InternetGatewayDevice'])
    control_point.connect(igd_found, 'Coherence.UPnP.ControlPoint.InternetGatewayDevice.detected')
    control_point.connect(igd_removed, 'Coherence.UPnP.ControlPoint.InternetGatewayDevice.removed')

reactor.callWhenRunning(start)
reactor.run()