66 lines
1.7 KiB
Python
66 lines
1.7 KiB
Python
import urequests
|
|
|
|
from . import logging
|
|
from .config import config
|
|
from .config import secrets
|
|
|
|
|
|
class AdafruitIO:
|
|
def __init__(self):
|
|
self.hostname = config['hostname']
|
|
|
|
self.headers = {
|
|
'x-aio-key': secrets['adafruit_io_key'],
|
|
'content-type': 'application/json'
|
|
}
|
|
|
|
group = config['adafruit_io_group']
|
|
username = secrets['adafruit_io_username']
|
|
self.url = f'https://io.adafruit.com/api/v2/{username}/groups/{group}/data'
|
|
|
|
def upload(self, reading):
|
|
result = self.uploadReading(reading)
|
|
|
|
try:
|
|
if result.status_code != 200:
|
|
self.logError(result)
|
|
return False
|
|
|
|
return True
|
|
finally:
|
|
result.close()
|
|
|
|
def uploadReading(self, reading):
|
|
return urequests.post(
|
|
self.url,
|
|
json=self.createPayload(reading),
|
|
headers=self.headers
|
|
)
|
|
|
|
def createPayload(self, reading):
|
|
payload = {
|
|
'created_at': reading['timestamp'],
|
|
'feeds': []
|
|
}
|
|
|
|
for key, value in reading['readings'].items():
|
|
payload['feeds'].append({
|
|
'key': f'{self.hostname}-{key}',
|
|
'value': str(value)
|
|
})
|
|
|
|
return payload
|
|
|
|
def logError(self, result):
|
|
status = result.status_code
|
|
reason = result.reason.decode('utf-8')
|
|
error_message = self.getErrorMessage(result)
|
|
|
|
logging.debug(f'Adafruit IO upload issue: {status} {reason} - "{error_message}"')
|
|
|
|
def getErrorMessage(self, result):
|
|
try:
|
|
return result.json()['error']
|
|
except (TypeError, KeyError):
|
|
return ''
|