aquifer/net/adafruit_io.py

import urequests

from . import logging
from .config import config
from .config import secrets


class AdafruitIO:
    def __init__(self):
        self.nickname = config['adafruit_io_nickname']

        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.nickname}-{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 ''