instrumentation/next-share/BaseLib/UPnP/upnpserver/upnpdevice.py

   1 # Written by Ingar Arntzen, Norut
   2 # see LICENSE.txt for license information
   3
   4 """
   5 This module implements a UPnP device.  The implementation
   6 takes care of automatic production of xml and html descriptions
   7 of the device.
   8 """
   9 import uuid
  10 import exceptions
  11
  12 ##############################################
  13 # HTML
  14 ##############################################
  15
  16 _HTML_FMT = "<html>\n<header></header>\n<body>\n%s</body>\n</html>"
  17 _HTML_BODY_FMT = "<h1>RootDevice : %s</h1>\n%s"
  18 _HTML_SERVICE_LIST_FMT = "<h2>Services</h2><ol>\n%s</ol>\n"
  19
  20 _HTML_DEVICE_TYPE_FMT = "DeviceType : urn:%s:device:%s:%s<br>\n"
  21 _HTML_DEVICE_NAME_FMT = "Name : %s<br>\n"
  22 _HTML_MANUFACTURER_FMT = "Manufacturer : %s<br>\n"
  23 _HTML_MANUFACTURER_URL_FMT = "ManufacturerURL : %s<br>\n"
  24 _HTML_MODEL_NAME_FMT = "ModelName : %s<br>\n"
  25 _HTML_MODEL_NUMBER_FMT = "ModelNumber : %s<br>\n"
  26 _HTML_MODEL_DESCRIPTION_FMT = "ModelDescription : %s<br>\n"
  27 _HTML_MODEL_URL_FMT = "ModelURL : %s<br>\n"
  28 _HTML_SERIAL_NUMBER_FMT = "SerialNumber : %s<br>\n"
  29 _HTML_DEVICE_UUID_FMT = "UDN : uuid:%s<br>\n"
  30 _HTML_UPC_FMT = "UPC : %s<br>\n"
  31 _HTML_PRESENTATION_FMT = "PresentationURL : <a href=%s>%s</a><br>\n"
  32 _HTML_DESCRIPTION_FMT = "DescriptionURL : <a href=%s>%s</a><br>\n"
  33
  34 _HTML_SERVICE_FMT = "<li><h3>%s</h3><br>\n%s</li>\n"
  35 _HTML_SERVICE_TYPE_FMT = "ServiceType : urn:schemas-upnp-org:" + \
  36     "service:%s:%s<br>\n"
  37 _HTML_SERVICE_ID_FMT = "ServiceID : urn:upnp-org:serviceId:%s<br>\n"
  38 _HTML_SERVICE_DESCRIPTION_URL_FMT = "SCPDURL : <a href=%s>%s</a><br>\n"
  39 _HTML_SERVICE_CONTROL_URL_FMT = "ControlURL : %s<br>\n"
  40 _HTML_SERVICE_EVENT_URL_FMT = "EventSubURL : %s<br>\n"
  41
  42 def _device_entries_tohtml(device):
  43     """Produce html for all attributes of a device."""
  44     str_ = []
  45     str_.append( _HTML_DEVICE_TYPE_FMT % (device.device_domain,
  46                                        device.device_type,
  47                                        device.device_version))
  48     str_.append( _HTML_DEVICE_NAME_FMT % device.name)
  49     if device.manufacturer != None:
  50         str_.append( _HTML_MANUFACTURER_FMT % device.manufacturer)
  51     if device.manufacturer_url != None:
  52         str_.append( _HTML_MANUFACTURER_URL_FMT % device.manufacturer_url)
  53     if device.model_name != None:
  54         str_.append( _HTML_MODEL_NAME_FMT % device.model_name)
  55     if device.model_number != None:
  56         str_.append( _HTML_MODEL_NUMBER_FMT % device.model_number)
  57     if device.model_description != None:
  58         str_.append( _HTML_MODEL_DESCRIPTION_FMT % device.model_description)
  59     if device.model_url != None:
  60         str_.append( _HTML_MODEL_URL_FMT % device.model_url)
  61     if device.serial_number != None:
  62         str_.append( _HTML_SERIAL_NUMBER_FMT % device.serial_number)
  63     str_.append( _HTML_DEVICE_UUID_FMT % device.uuid)
  64     if device.upc != None:
  65         str_.append( _HTML_UPC_FMT % device.upc)
  66     url = device.get_presentation_url()
  67     str_.append( _HTML_PRESENTATION_FMT % (url, url ))
  68     url = device.get_description_url()
  69     str_.append( _HTML_DESCRIPTION_FMT % (url, url ))
  70     return "".join(str_)
  71
  72 def _service_list_tohtml(services):
  73     """Produce html for all services contained in a device."""
  74     if len(services) > 0:
  75         list_ = []
  76         for service in services:
  77             str_ = ""
  78             str_ += _HTML_SERVICE_TYPE_FMT % (service.service_type,
  79                                            service.service_version)
  80             str_ += _HTML_SERVICE_ID_FMT % service.service_id
  81             url = service.base_url + service.description_path
  82             str_ += _HTML_SERVICE_DESCRIPTION_URL_FMT % (url, url)
  83             str_ += _HTML_SERVICE_CONTROL_URL_FMT % \
  84                 (service.base_url + service.control_path)
  85             str_ += _HTML_SERVICE_EVENT_URL_FMT % \
  86                 (service.base_url + service.event_path)
  87             list_.append(_HTML_SERVICE_FMT % (service.service_id, str_))
  88         return _HTML_SERVICE_LIST_FMT % "".join(list_)
  89     else : return ""
  90
  91 def _device_tohtml(device):
  92     """Produce html description for a device. """
  93     entries = _device_entries_tohtml(device)
  94     service_list = _service_list_tohtml(device.get_services())
  95     body = _HTML_BODY_FMT % (device.name, entries + service_list)
  96     return _HTML_FMT % body
  97
  98
  99 ##############################################
 100 # XML
 101 ##############################################
 102
 103 _DEVICE_DESCRIPTION_FMT = """<?xml version="1.0"?>
 104 <root xmlns="urn:schemas-upnp-org:device-1-0">
 105 <specVersion>
 106 <major>1</major>
 107 <minor>0</minor>
 108 </specVersion>
 109 <URLBase>%s</URLBase>
 110 %s
 111 </root>"""
 112
 113 _DEVICE_FMT = "<device>\n%s</device>"
 114 _DEVICE_LIST_FMT = "<deviceList>\n%s</deviceList>\n"
 115 _SERVICE_LIST_FMT = "<serviceList>\n%s</serviceList>\n"
 116
 117 _DEVICE_TYPE_FMT = "<deviceType>urn:%s:device:%s:%s</deviceType>\n"
 118 _DEVICE_NAME_FMT = "<friendlyName>%s</friendlyName>\n"
 119 _MANUFACTURER_FMT = "<manufacturer>%s</manufacturer>\n"
 120 _MANUFACTURER_URL_FMT = "<manufacturerURL>%s</manufacturerURL>\n"
 121 _MODEL_NAME_FMT = "<modelName>%s</modelName>\n"
 122 _MODEL_NUMBER_FMT = "<modelNumber>%s</modelNumber>\n"
 123 _MODEL_DESCRIPTION_FMT = "<modelDescription>%s</modelDescription>\n"
 124 _MODEL_URL_FMT = "<modelURL>%s</modelURL>\n"
 125 _SERIAL_NUMBER_FMT = "<serialNumber>%s</serialNumber>\n"
 126 _DEVICE_UUID_FMT = "<UDN>uuid:%s</UDN>\n"
 127 _UPC_FMT = "<UPC>%s</UPC>\n"
 128 _PRESENTATION_FMT = "<presentationURL>%s</presentationURL>\n"
 129
 130
 131 _SERVICE_FMT = "<service>\n%s</service>\n"
 132 _SERVICE_TYPE_FMT = "<serviceType>urn:schemas-upnp-org:service:" + \
 133     "%s:%s</serviceType>\n"
 134 _SERVICE_ID_FMT = "<serviceId>urn:upnp-org:serviceId:%s</serviceId>\n"
 135 _SERVICE_DESCRIPTION_URL_FMT = "<SCPDURL>%s</SCPDURL>\n"
 136 _SERVICE_CONTROL_URL_FMT = "<controlURL>%s</controlURL>\n"
 137 _SERVICE_EVENT_URL_FMT = "<eventSubURL>%s</eventSubURL>\n"
 138
 139
 140 def _device_entries_toxml(device):
 141     """Produce xml description for device attributes."""
 142     str_ = []
 143     str_.append( _DEVICE_TYPE_FMT % (device.device_domain,
 144                                   device.device_type, device.device_version))
 145     str_.append( _DEVICE_NAME_FMT % device.name)
 146     if device.manufacturer != None:
 147         str_.append( _MANUFACTURER_FMT % device.manufacturer)
 148     if device.manufacturer_url != None:
 149         str_.append( _MANUFACTURER_URL_FMT % device.manufacturer_url)
 150     if device.model_name != None:
 151         str_.append( _MODEL_NAME_FMT % device.model_name)
 152     if device.model_number != None:
 153         str_.append( _MODEL_NUMBER_FMT % device.model_number)
 154     if device.model_description != None:
 155         str_.append( _MODEL_DESCRIPTION_FMT % device.model_description)
 156     if device.model_url != None:
 157         str_.append( _MODEL_URL_FMT % device.model_url)
 158     if device.serial_number != None:
 159         str_.append( _SERIAL_NUMBER_FMT % device.serial_number)
 160     str_.append( _DEVICE_UUID_FMT % device.uuid)
 161     if device.upc != None:
 162         str_.append( _UPC_FMT % device.upc)
 163     str_.append( _PRESENTATION_FMT % (device.get_presentation_url()))
 164     return "".join(str_)
 165
 166 def _device_list_toxml(devices):
 167     """Produce xml for devices contained within a device."""
 168     if len(devices) > 0:
 169         str_ = ""
 170         for device in devices:
 171             str_ += device.to_xml()
 172         return _DEVICE_LIST_FMT % str_
 173     else : return ""
 174
 175 def _service_list_toxml(services):
 176     """Produce xml for services contained within a device."""
 177     if len(services) > 0:
 178         list_ = []
 179         for service in services:
 180             str_ = ""
 181             str_ += _SERVICE_TYPE_FMT % (service.service_type,
 182                                       service.service_version)
 183             str_ += _SERVICE_ID_FMT % service.service_id
 184             str_ += _SERVICE_DESCRIPTION_URL_FMT % (service.description_path)
 185             str_ += _SERVICE_CONTROL_URL_FMT % (service.control_path)
 186             str_ += _SERVICE_EVENT_URL_FMT % (service.event_path)
 187             list_.append(_SERVICE_FMT % str_)
 188         return _SERVICE_LIST_FMT % "".join(list_)
 189     else : return ""
 190
 191
 192 ##############################################
 193 # UPNP DEVICE
 194 ##############################################
 195
 196 class UPnPDeviceError(exceptions.Exception):
 197     """Errro associated with UPnP Device."""
 198     pass
 199
 200 class UPnPDevice:
 201
 202     """
 203     This implements the internal representation of a UPNP Device.
 204     The representation is used to generate the XML description,
 205     and to add or remove services.
 206     The given service manager implements the hierarchical namespace
 207     for devices and services.
 208     """
 209
 210     def __init__(self, device_config=None):
 211         self._sm = None
 212         self._is_root = False
 213
 214         # Initialse Device from config.
 215         if device_config == None:
 216             device_config = {}
 217         self.name = device_config.get('name', None)
 218         self.device_type = device_config.get('device_type', None)
 219         self.device_version = device_config.get('device_version', 1)
 220         self.device_domain = device_config.get('device_domain',
 221                                                'schemas-upnp-org')
 222         self.manufacturer = device_config.get('manufacturer', None)
 223         self.manufacturer_url = device_config.get('manufacturer_url', None)
 224         self.model_name = device_config.get('model_name', None)
 225         self.model_number = device_config.get('model_number', None)
 226         self.model_description = device_config.get('model_description', None)
 227         self.model_url = device_config.get('model_url', None)
 228         self.serial_number = device_config.get('serial_number', None)
 229         self.uuid = uuid.uuid1()
 230         self.upc = device_config.get('upc', None)
 231
 232         self.base_url = ""
 233         self.presentation_path = ""
 234         self.description_path = ""
 235
 236
 237     def set_service_manager(self, service_manager):
 238         """Initialise device with reference to service manager."""
 239         self._sm = service_manager
 240         self.base_url = self._sm.get_base_url()
 241         self.presentation_path = "devices/%s/presentation.html" % self.name
 242         self.description_path = "devices/%s/description.xml" % self.name
 243
 244     def set_is_root(self, value):
 245         """Initialise device with is_root flag."""
 246         self._is_root = value
 247
 248     def is_root(self):
 249         """Checks if device is root."""
 250         return self._is_root
 251
 252     def is_valid(self):
 253         """Checks if device object has been properly initialised."""
 254         return (self.device_type != None and self.name != None
 255                 and self.uuid != None and self.base_url != None
 256                 and self._sm != None)
 257
 258     def get_services(self):
 259         """Returns services which are included in this device."""
 260         return self._sm.get_services_of_device(self)
 261
 262     def get_devices(self):
 263         """Returns devices which are included in this device."""
 264         return self._sm.get_devices_of_device(self)
 265
 266     def to_xml(self):
 267         """Returns the xml description of this device."""
 268         if self.is_valid():
 269             device_entries_xml = _device_entries_toxml(self)
 270             service_list_xml = _service_list_toxml(self.get_services())
 271             device_list_xml = _device_list_toxml(self.get_devices())
 272             return _DEVICE_FMT % (device_entries_xml + \
 273                                       service_list_xml + device_list_xml)
 274         else:
 275             msg = "Can not generate XML description . Invalid Device"
 276             raise UPnPDeviceError, msg
 277
 278     def get_presentation_url(self):
 279         """Returns the presentation URL (html) for this device."""
 280         if self.is_root():
 281             return self._sm.get_presentation_path()
 282         else:
 283             return self.base_url + self.presentation_path
 284
 285     def get_description_url(self):
 286         """Returns the description URL (xml) for this device."""
 287         if self.is_root():
 288             return self._sm.get_description_path()
 289         else:
 290             return self.base_url + self.description_path
 291
 292     def get_xml_description(self):
 293         """
 294         Returns xml description wrapped in a valid xml document.
 295         Should only be invoked on the root device.
 296         """
 297         if self.is_valid():
 298             return _DEVICE_DESCRIPTION_FMT % (self.base_url, self.to_xml())
 299         else:
 300             msg = "Can not generate XML description. Invalid Device"
 301             raise UPnPDeviceError, msg
 302
 303     def get_html_description(self):
 304         """Return html description of device."""
 305         return _device_tohtml(self)
 306
 307     def close(self):
 308         """Close this device."""
 309         pass
 310
 311 ##############################################
 312 # MAIN
 313 ##############################################
 314
 315 if __name__ == '__main__':
 316
 317     DEVICE_CONF = {
 318         'device_type' : "MediaServer",
 319         'device_version': 1,
 320         'device_domain': 'schemas-upnp-org',
 321         'name': "MediaServer",
 322         'manufacturer': "Manufacturer",
 323         'manufacturer_url': 'http://manufacturer.com',
 324         'model_description': 'Model description',
 325         'model_name': 'Model 1',
 326         'model_number': '123456',
 327         'model_url': 'http://manufacturer.com/model_1',
 328         'serial_number': '123456',
 329         'uuid' : "3dd705c2-1c8a-11df-80c7-00248116b859",
 330         'upc': 'universial product code',
 331         }
 332
 333
 334     class MockService:
 335         """Mock Service"""
 336         def __init__(self):
 337             self.service_type = "ContentDirectory"
 338             self.service_version = 1
 339             self.service_id =  "ContentDirectory"
 340             self.description_path = "%s/description.xml" % self.service_id
 341             self.control_path = "%s/control" % self.service_id
 342             self.event_path = "%s/events" % self.service_id
 343             self.base_url = "http://myhost:4444/"
 344
 345     class MockServiceManager:
 346         """Mock Service Manager."""
 347         def __init__(self):
 348             self._mock_service = MockService()
 349         def get_services_of_device(self, device):
 350             """Mock method."""
 351             return [self._mock_service]
 352         def get_devices_of_device(self, device):
 353             """Mock method."""
 354             return []
 355         def get_base_url(self):
 356             """Mock method."""
 357             return 'http://myhost:4444/'
 358
 359
 360     SM = MockServiceManager()
 361
 362     DEVICE = UPnPDevice(DEVICE_CONF)
 363     DEVICE.set_service_manager(SM)
 364
 365     print DEVICE.get_xml_description()
 366     print DEVICE.get_html_description()