hat.drivers.ssdp

 1"""Simple Service Discovery Protocol"""
 2
 3import logging
 4import typing
 5
 6from hat import aio
 7
 8from hat.drivers import net
 9
10
11mlog: logging.Logger = logging.getLogger(__name__)
12"""Module logger"""
13
14
15class DeviceInfo(typing.NamedTuple):
16    addr: net.UdpAddress
17    location: str
18    server: str
19    service: str
20
21
22DeviceInfoCb: typing.TypeAlias = aio.AsyncCallable[[DeviceInfo], None]
23"""Device info callback"""
24
25
26default_multicast_addr = net.UdpAddress('239.255.255.250', 1900)
27
28
29async def discover(device_info_cb: DeviceInfoCb,
30                   multicast_addr: net.UdpAddress = default_multicast_addr,
31                   local_name: str = 'hat'
32                   ) -> 'DiscoveryServer':
33    """Create discovery server"""
34    endpoint = await net.create_endpoint(
35        local_addr=net.UdpAddress('0.0.0.0', multicast_addr.port))
36
37    srv = DiscoveryServer()
38    srv._endpoint = endpoint
39    srv._device_info_cb = device_info_cb
40    srv._multicast_addr = multicast_addr
41    srv._local_name = local_name
42    srv._async_group = aio.Group()
43    srv._async_group.spawn(aio.call_on_cancel, endpoint.async_close)
44    srv._async_group.spawn(srv._discovery_loop)
45    return srv
46
47
48class DiscoveryServer(aio.Resource):
49    """Discovery server"""
50
51    @property
52    def async_group(self) -> aio.Group:
53        """Async group"""
54        return self._async_group
55
56    async def _discovery_loop(self):
57        try:
58            req = _encode_search_req(self._local_name)
59            self._endpoint.send(req, self._multicast_addr)
60
61            while True:
62                res, addr = await self._endpoint.receive()
63                try:
64                    info = _decode_search_res(addr, res)
65                except Exception:
66                    continue
67                await aio.call(self._device_info_cb, info)
68
69        finally:
70            self._async_group.close()
71
72
73def _encode_search_req(local_name):
74    return (f'M-SEARCH * HTTP/1.1\r\n'
75            f'HOST: 239.255.255.250:1900\r\n'
76            f'MAN: "ssdp:discover"\r\n'
77            f'MX: 1\r\n'
78            f'ST: ssdp:all\r\n'
79            f'CPFN.UPNP.ORG: {local_name}\r\n').encode('utf-8')
80
81
82def _decode_search_res(addr, data):
83    lines = str(data, encoding='utf-8').strip().split('\r\n')
84    if lines[0].strip() != 'HTTP/1.1 200 OK':
85        raise Exception('invalid response')
86    entries = {}
87    for line in lines[1:]:
88        line = line.strip()
89        if not line:
90            continue
91        segments = [i.strip() for i in line.split(':', 1)]
92        entries[segments[0].upper()] = segments[1]
93    return DeviceInfo(addr=addr,
94                      location=entries['LOCATION'],
95                      server=entries['SERVER'],
96                      service=entries['USN'])