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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
"""Various helper utility for Hue bridge discovery."""
from __future__ import annotations
import contextlib
from dataclasses import dataclass
from aiohttp import ClientConnectionError, ClientError, ClientSession
from .util import normalize_bridge_id
URL_NUPNP = "https://discovery.meethue.com/"
@dataclass(frozen=True)
class DiscoveredHueBridge:
"""Model for a discovered Hue bridge."""
host: str
id: str
supports_v2: bool
async def discover_bridge(
host: str,
websession: ClientSession | None = None,
) -> DiscoveredHueBridge:
"""
Discover bridge details from given hostname/ip.
Raises exception from aiohttp if there is no bridge alive on given ip/host.
"""
websession_provided = websession is not None
if websession is None:
websession = ClientSession()
try:
bridge_id = await is_hue_bridge(host, websession)
supports_v2 = await is_v2_bridge(host, websession)
return DiscoveredHueBridge(host, bridge_id, supports_v2)
finally:
if not websession_provided:
await websession.close()
async def discover_nupnp(
websession: ClientSession | None = None,
) -> list[DiscoveredHueBridge]:
"""Discover bridges via NUPNP."""
result = []
websession_provided = websession is not None
if websession is None:
websession = ClientSession()
try:
async with websession.get(URL_NUPNP, timeout=30) as res:
for item in await res.json():
host = item["internalipaddress"]
# the nupnp discovery might return items that are not in local network
# connect to each bridge to find out if it's alive.
with contextlib.suppress(Exception):
result.append(await discover_bridge(host, websession))
return result
except ClientError:
return result
finally:
if not websession_provided:
await websession.close()
async def is_hue_bridge(host: str, websession: ClientSession | None = None) -> str:
"""
Check if there is a bridge alive on given ip and return bridge ID.
Raises exception from aiohttp if the bridge can not be reached on given hostname.
"""
websession_provided = websession is not None
if websession is None:
websession = ClientSession()
try:
# every hue bridge returns discovery info on this endpoint
url = f"http://{host}/api/config"
async with websession.get(url, timeout=30) as res:
res.raise_for_status()
data = await res.json()
if "bridgeid" not in data:
# there are some emulator projects out there that emulate a Hue bridge
# in a sloppy way, ignore them.
# https://github.com/home-assistant-libs/aiohue/issues/134
raise ClientConnectionError(
"Invalid API response, not a real Hue bridge?"
)
return normalize_bridge_id(data["bridgeid"])
finally:
if not websession_provided:
await websession.close()
async def is_v2_bridge(host: str, websession: ClientSession | None = None) -> bool:
"""Check if the bridge has support for the new V2 api."""
websession_provided = websession is not None
if websession is None:
websession = ClientSession()
try:
# v2 api is https only and returns a 403 forbidden when no key provided
url = f"https://{host}/clip/v2/resource"
async with websession.get(
url, ssl=False, raise_for_status=False, timeout=30
) as res:
return res.status == 403
except Exception: # pylint: disable=broad-except
# all other status/exceptions means the bridge is not v2 or not reachable at this time
return False
finally:
if not websession_provided:
await websession.close()
|
