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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
|
import json
from datetime import datetime
class MeaterApi(object):
"""Meater api object"""
def __init__(self, aiohttp_session):
self._jwt = None
self._session = aiohttp_session
async def get_all_devices(self):
"""Get all the device states."""
device_states = await self.__get_raw_state_all()
devices = []
for device in device_states:
devices.append(self.__get_probe_object(device))
return devices
async def get_device(self, device_id):
device_state = await self.__get_raw_state(device_id)
return self.__get_probe_object(device_state)
async def __get_raw_state_all(self):
"""Get raw device state from the Meater API. We have to have authenticated before now."""
if not self._jwt:
raise AuthenticationError('You need to authenticate before making requests to the API.')
headers = {'Authorization': 'Bearer ' + self._jwt}
async with self._session.get('https://public-api.cloud.meater.com/v1/devices', headers=headers) as device_state_request:
if device_state_request.status == 401:
raise AuthenticationError('Unable to authenticate with the Meater API')
if device_state_request.status == 500:
raise ServiceUnavailableError('The service is currently unavailable')
if device_state_request.status == 429:
raise TooManyRequestsError('Too many requests have been made to the API')
if device_state_request.status != 200:
raise Exception('Error connecting to Meater')
device_state_body = await device_state_request.json()
if len(device_state_body) == 0:
raise Exception('The server did not return a valid response')
return device_state_body.get('data').get('devices')
async def __get_raw_state(self, device_id):
"""Get raw device state from the Meater API. We have to have authenticated before now."""
if not self._jwt:
raise AuthenticationError('You need to authenticate before making requests to the API.')
headers = {'Authorization': 'Bearer ' + self._jwt}
async with self._session.get('https://public-api.cloud.meater.com/v1/devices/' + device_id, headers=headers) as device_state_request:
if device_state_request.status == 404:
raise UnknownDeviceError('The specified device could not be found, it might not be connected to Meater Cloud')
if device_state_request.status == 401:
raise AuthenticationError('Unable to authenticate with the Meater API')
if device_state_request.status == 500:
raise ServiceUnavailableError('The service is currently unavailable')
if device_state_request.status == 429:
raise TooManyRequestsError('Too many requests have been made to the API')
if device_state_request.status != 200:
raise Exception('Error connecting to Meater')
device_state_body = await device_state_request.json()
if len(device_state_body) == 0:
raise Exception('The server did not return a valid response')
return device_state_body.get('data')
async def authenticate(self, email, password):
"""Authenticate with Meater."""
headers = {'Content-Type':'application/json'}
body = {'email':email, 'password':password}
async with self._session.post('https://public-api.cloud.meater.com/v1/login', data = json.dumps(body), headers=headers) as meater_auth_req:
if meater_auth_req.status == 401:
raise AuthenticationError('The specified credientals were incorrect')
if meater_auth_req.status == 500:
raise ServiceUnavailableError('The service is currently unavailable')
if meater_auth_req.status == 429:
raise TooManyRequestsError('Too many requests have been made to the API')
if meater_auth_req.status != 200:
raise Exception('Couldn\'t authenticate with the Meater API')
auth_body = await meater_auth_req.json()
jwt = auth_body.get('data').get('token') # The JWT is valid indefinitely...
if not jwt:
raise AuthenticationError('Unable to obtain an auth token from the Meater API')
# Set JWT local variable
self._jwt = jwt
return True
def __get_probe_object(self, device):
cook = None
if device.get('cook'):
target_temp = 0
cook = MeaterCook(device.get('cook').get('id'), device.get('cook').get('name'), device.get('cook').get('state'), device.get('cook').get('temperature').get('target'), device.get('cook').get('temperature').get('peak'), device.get('cook').get('time').get('remaining'), device.get('cook').get('time').get('elapsed'))
probe = MeaterProbe(device.get('id'), device.get('temperature').get('internal'), device.get('temperature').get('ambient'), cook, device.get('updated_at'))
return probe
class MeaterProbe(object):
def __init__(self, id, internal_temp, ambient_temp, cook, time_updated):
self.id = id
self.internal_temperature = float(internal_temp) # Always in degrees celcius
self.ambient_temperature = float(ambient_temp) # Always in degrees celcius
self.cook = cook
self.time_updated = datetime.fromtimestamp(time_updated)
class MeaterCook(object):
def __init__(self, id, name, state, target_temp, peak_temp, time_remaining, time_elapsed):
self.id = id
self.name = name
self.state = state
if target_temp:
self.target_temperature = float(target_temp) # Always in degrees celcius
if peak_temp:
self.peak_temperature = float(peak_temp) # Always in degrees celcius
if time_remaining:
self.time_remaining = int(time_remaining) # Always in seconds
if time_elapsed:
self.time_elapsed = int(time_elapsed) # Always in seconds
class UnknownDeviceError(Exception):
pass
class AuthenticationError(Exception):
pass
class ServiceUnavailableError(Exception):
pass
class TooManyRequestsError(Exception):
pass
|
