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
|
"""Python asyncio client for the Awair REST API.
This module is an object-oriented wrapper around the Awair_
REST API_. It requires an access token (which can be obtained
from the `developer console`_) and implements read-only access
to the "user" portions of the API.
.. _Awair: https://getawair.com
.. _API: https://docs.developer.getawair.com/?version=latest
.. _`developer console`: https://developer.getawair.com
"""
from asyncio import gather
from typing import List, Optional
from aiohttp import ClientSession
from python_awair import const
from python_awair.auth import AccessTokenAuth, AwairAuth
from python_awair.client import AwairClient
from python_awair.devices import AwairLocalDevice
from python_awair.exceptions import AwairError
from python_awair.user import AwairUser
class Awair:
"""Entry class for the Awair API.
Args:
session: An aiohttp session that will be used to query
the Awair API.
access_token: An optional access token, obtained from
the Awair developer console, used to authenticate
to the Awair API.
authenticator: An optional instance of an AwairAuth class,
which can provide an HTTP Bearer token for
authentication. Most users will simply provide an
access_token, instead.
"""
client: AwairClient
"""AwairClient: The instantiated AwairClient
that will be used to fetch API responses and
check for HTTP errors.
"""
def __init__(
self,
session: ClientSession,
access_token: Optional[str] = None,
authenticator: Optional[AwairAuth] = None,
) -> None:
"""Initialize the Awair API wrapper."""
if authenticator:
self.client = AwairClient(authenticator, session)
elif access_token:
self.client = AwairClient(AccessTokenAuth(access_token), session)
else:
raise AwairError("No authentication supplied!")
async def user(self) -> AwairUser:
"""Return the current AwairUser.
The Awair "user" API does not provide a way to query for
a specific user, so this method always returns the user
that is associated with the authentication that is in-use.
This is *typically* the user that owns the access_token
that was provided at class instantiaton, unless you have
provided an authenticator class implementing some other
stategy (perhaps OAuth).
"""
response = await self.client.query(const.USER_URL)
return AwairUser(client=self.client, attributes=response)
class AwairLocal:
"""Entry class for the local sensors Awair API."""
client: AwairClient
"""AwairClient: The instantiated AwairClient
that will be used to fetch API responses and
check for HTTP errors.
"""
_device_addrs: List[str]
"""IP or DNS addresses of Awair devices with the local sensors API enabled."""
def __init__(self, session: ClientSession, device_addrs: List[str]) -> None:
"""Initialize the Awair local sensors API wrapper."""
self._device_addrs = device_addrs
if len(device_addrs) > 0:
self.client = AwairClient(AccessTokenAuth(""), session)
else:
raise AwairError("No local Awair device addresses supplied!")
async def devices(self) -> List[AwairLocalDevice]:
"""Return a list of local awair devices."""
responses = await gather(
*(
self.client.query(f"http://{addr}/settings/config/data")
for addr in self._device_addrs
)
)
return [
AwairLocalDevice(
client=self.client, device_addr=self._device_addrs[i], attributes=device
)
for i, device in enumerate(responses)
]
|
