1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
"""Discovery module.
This module uses async_upnp_client to discover devices supporting the
urn:schemas-sony-com:service:ScalarWebAPI:1 service used by this library.
"""
import logging
from xml import etree
import attr
from async_upnp_client.aiohttp import AiohttpRequester
from async_upnp_client.client_factory import UpnpFactory
from async_upnp_client.search import async_search
_LOGGER = logging.getLogger(__name__)
@attr.s
class DiscoveredDevice:
"""Container for discovered device information."""
name = attr.ib()
model_number = attr.ib()
udn = attr.ib()
services = attr.ib()
upnp_location = attr.ib()
endpoint = attr.ib()
version = attr.ib()
upnp_services = attr.ib()
class Discover:
"""Implementation of UPnP discoverer for supported devices."""
@staticmethod
async def discover(timeout, debug=0, callback=None, source_address=None):
"""Discover supported devices."""
ST = "urn:schemas-sony-com:service:ScalarWebAPI:1"
_LOGGER.info("Discovering for %s seconds" % timeout)
async def parse_device(device):
requester = AiohttpRequester()
factory = UpnpFactory(requester)
url = device["location"]
try:
device = await factory.async_create_device(url)
except Exception as ex:
_LOGGER.error(
"Unable to download the device description file from %s: %s",
url,
ex,
)
return
if debug > 0:
print(etree.ElementTree.tostring(device.xml).decode())
NS = {"av": "urn:schemas-sony-com:av"}
info = device.xml.find(".//av:X_ScalarWebAPI_DeviceInfo", NS)
if not info:
_LOGGER.error("Unable to find X_ScalaerWebAPI_DeviceInfo")
return
endpoint = info.find(".//av:X_ScalarWebAPI_BaseURL", NS).text
version = info.find(".//av:X_ScalarWebAPI_Version", NS).text
services = [
x.text for x in info.findall(".//av:X_ScalarWebAPI_ServiceType", NS)
]
dev = DiscoveredDevice(
name=device.name,
model_number=device.model_number,
udn=device.udn,
endpoint=endpoint,
version=version,
services=services,
upnp_services=list(device.services.keys()),
upnp_location=url,
)
_LOGGER.debug("Discovered: %s" % dev)
if callback is not None:
await callback(dev)
if source_address is not None:
source_address = (source_address, 0)
await async_search(
timeout=timeout,
search_target=ST,
async_callback=parse_device,
source=source_address,
)
|
