Content:
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:
from twisted.internet import reactor from coherence.base import Coherence from coherence.upnp.devices.control_point import ControlPoint from coherence.upnp.core import DIDLLite 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) 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) d.addErrback(lamda error: print error) 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') reactor.callWhenRunning(start) reactor.run()
