hat.drivers.upnp.description

  1"""WIP UPnP description"""
  2
  3import collections
  4import io
  5import typing
  6import xml.sax.handler
  7
  8import aiohttp
  9
 10
 11class Icon(typing.NamedTuple):
 12    mimetype: str
 13    width: int
 14    height: int
 15    depth: int
 16    url: str
 17
 18
 19class DeviceDescription(typing.NamedTuple):
 20    url: str | None
 21    dev_type: str
 22    dev_name: str
 23    manufacturer: str
 24    manufacturer_url: str | None
 25    model_desc: str | None
 26    model_name: str
 27    model_number: str | None
 28    model_url: str | None
 29    serial_number: str | None
 30    unique_dev_name: str
 31    icons: list[Icon]
 32
 33
 34async def get_description(location: str) -> DeviceDescription:
 35    async with aiohttp.ClientSession() as session:
 36        async with session.get(location) as res:
 37            text = await res.text()
 38        data = _decode_xml(text)
 39        version = data['root']['specVersion']
 40        if version['major'] != '1':
 41            raise Exception(f"unsupported version {version}")
 42        device = data['root']['device']
 43        device_icons = device.get('iconList', {}).get('icon', [])
 44        if not isinstance(device_icons, list):
 45            device_icons = [device_icons]
 46        icons = []
 47        for i in device_icons:
 48            icons.append(Icon(mimetype=i['mimetype'],
 49                              width=i['width'],
 50                              height=i['height'],
 51                              depth=i['depth'],
 52                              url=i['url']))
 53        return DeviceDescription(
 54            url=data['root'].get('URLBase'),
 55            dev_type=device['deviceType'],
 56            dev_name=device['friendlyName'],
 57            manufacturer=device['deviceType'],
 58            manufacturer_url=device.get('manufacturerURL'),
 59            model_desc=device.get('modelDescription'),
 60            model_name=device['modelName'],
 61            model_number=device.get('modelNumber'),
 62            model_url=device.get('modelURL'),
 63            serial_number=device.get('serialNumber'),
 64            unique_dev_name=device['UDN'],
 65            icons=icons)
 66
 67
 68def _decode_xml(text):
 69    handler = _XmlContentHandler()
 70    parser = xml.sax.make_parser()
 71    parser.setContentHandler(handler)
 72    parser.setFeature(xml.sax.handler.feature_external_ges, False)
 73    parser.setFeature(xml.sax.handler.feature_external_pes, False)
 74    parser.parse(io.StringIO(text))
 75    return handler.root
 76
 77
 78class _XmlContentHandler(xml.sax.ContentHandler):
 79
 80    def __init__(self):
 81        self._root = {}
 82        self._stack = collections.deque([self._root])
 83
 84    @property
 85    def root(self):
 86        return self._root
 87
 88    def startElement(self, name, attrs):
 89        self._stack.append({})
 90
 91    def endElement(self, name):
 92        el = self._stack.pop()
 93        text = el.get('', '')
 94        children = {k: v for k, v in el.items() if k}
 95        top = self._stack[-1]
 96        if name in top:
 97            if not isinstance(top[name], list):
 98                top[name] = [top[name]]
 99            top[name].append(children or text)
100        else:
101            top[name] = children or text
102
103    def characters(self, content):
104        el = self._stack[-1]
105        el[''] = el.get('', '') + content