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
|
"""Define an object to interact with the REST API."""
from __future__ import annotations
import logging
from datetime import date
from typing import Any, cast
from aiohttp import ClientSession
from .api_request_handler import ApiRequestHandler
from .const import DEFAULT_API_VERSION, LOGGER
REST_API_BASE = "https://rt.ambientweather.net"
DEFAULT_LIMIT = 288
class API(ApiRequestHandler):
"""Define the API object."""
def __init__( # pylint: disable=too-many-arguments
self,
application_key: str | None,
api_key: str | None,
*,
api_version: int = DEFAULT_API_VERSION,
logger: logging.Logger = LOGGER,
session: ClientSession | None = None,
) -> None:
"""Initialize.
Args:
application_key: An Ambient Weather application key.
api_key: An Ambient Weather API key.
api_version: The version of the API to query.
logger: The logger to use.
session: An optional aiohttp ClientSession.
"""
super().__init__(
f"{REST_API_BASE}/v{api_version}", logger=logger, session=session
)
self._api_key = api_key
self._application_key = application_key
async def get_devices(self) -> list[dict[str, Any]]:
"""Get all devices associated with an API key.
Returns:
An API response payload.
"""
params: dict[str, Any] = {
"apiKey": self._api_key,
"applicationKey": self._application_key,
}
# This endpoint returns a list of device dicts.
return cast(
list[dict[str, Any]], await self._request("get", "devices", params=params)
)
async def get_device_details(
self,
mac_address: str,
*,
end_date: date | None = None,
limit: int = DEFAULT_LIMIT,
) -> list[dict[str, Any]]:
"""Get details of a device by MAC address.
Args:
mac_address: The MAC address of an Ambient Weather station.
end_date: An optional end date to limit data.
limit: An optional limit.
Returns:
An API response payload.
"""
params: dict[str, Any] = {
"apiKey": self._api_key,
"applicationKey": self._application_key,
"limit": limit,
}
if end_date:
params["endDate"] = end_date.isoformat()
# This endpoint returns a list device data dicts.
return cast(
list[dict[str, Any]],
await self._request("get", f"devices/{mac_address}", params=params),
)
|
