# A minimalistic data logger

With Python it's simple and fun to build a minimalistic data logger in just a few lines of code. In my case I wanted to take temperature and humidity measurements with an ESP8266 and a DHT22 sensor module.

# The API

I called this thing dumpster since I did not want to create a hug IoT bloat ware, I wanted to keep it simple. No models, nothing, just a place to dump the raw data. You can do all the fancy analytics afterwards. I used FastAPI which I fell in love with recently.

This is some

from fastapi import FastAPI, Header, Depends, HTTPException
from environs import Env
from pathlib import Path
from typing import List

from pydantic import BaseModel
from shortuuid import uuid
import json
import os

api = FastAPI()
env = Env()


class Device(BaseModel):
    id: str
    name: str
    tokens: List[str]
    category: str

devices = [
    Device(id="esp-01", name="ESP Terrasse", category="Klima", tokens=["sometoken", "someothertoken"])
]

class FileStroage():
    def __init__(self, base_dir):
        self.base_dir = base_dir

    def store(self, item: dict, device: Device):
        target_dir = self.base_dir / device.category / device.id
        os.makedirs(target_dir, exist_ok=True)
        with open(target_dir / uuid(), 'w') as fh:
            json.dump(item, fh)


def storage():
    base_dir = env.path("BASE_DIR")
    return FileStroage(base_dir)


def device(authorization = Header(None)):
    try:
        return next(device for device in devices if authorization in device.tokens)
    except StopIteration:
        raise HTTPException(
    status_code=403, detail="Token invalid.")


@api.post("/dumpster/")
def post_measurement(measurement: dict, device : Device =Depends(device), storage = Depends(storage)):
    storage.store(measurement, device)
    return measurement

I'd like to walk you through the key points of this code. First, it's important to understand that FastAPI uses decorators (the @foo stuff) to register the end points you provide at the application. The next important thing is that FastAPI does some magic with the parameter list of your function.

@api.post("/dumpster/")
def post_measurement(measurement: dict, device : Device =Depends(device), storage = Depends(storage)):
    storage.store(measurement, device)
    return measurement

Here we register a route called /dumpster which accepts POST requests. In the function body we can access a measurement, a device and a storage parameter. Where does this come from?

The measurement is the payload of our request. As there is no default value, FastAPI expects to find this somewhere in the request, either in the query or, as a last resort, it takes the body. That is what happens in our case.

For device and storage I used the dependency injection system of FastAPI. This is pretty clever, but also tricky. Let's start with the storage. This a very simple case where Depends() points to a callable, and the result of that call is then available under the name storage. In this case, it's a new instance of the FileStorage class. This storage has a method that dumps something to the file system.

def device(authorization = Header(None)):
    try:
        return next(device for device in devices if authorization in device.tokens)
    except StopIteration:
        raise HTTPException(
    status_code=403, detail="Token invalid.")

For device it's one more level of indirection. You see that the device() function itself takes a parameter, authorization=Header(None). With this information, FastAPI searches in the request headers for a header called Authorization (it's case insensitive I think, and it replaces - with _) and provides it in the parameter. If you would expect a header X-Foo: bar, you would instead write x_foo=Header(None) and x_foo would have the value bar for this particular request.

What happend next is that the code looks in the list of the known devices for one with that particular token. If one is found, it is returned. If none is found, a 403 type exception is raised.

# The sending part

This is code for the ESP to feed the API with values. It's up to you to send actual values, though.

This is boot.py, the file that gets executed after a hard reset:

import network
from urequests import post
import time

# "station"
sta_if = network.WLAN(network.STA_IF)

# "acess point"
ap_if = network.WLAN(network.AP_IF)

ap_if.active(False)
sta_if.active(True)

sta_if.connect("ssid", "password")

while not sta_if.isconnected():
    time.sleep(0.5)

This will connect to your desired network, make sure you exchange ssid and the other placeholders.

Now to do the real work you need simething like these lines:

from urequests import post
import time

token = "sometoken"
url = "http://hostname:8090/dumpster/"


def measurement():
    return {
    "test": 123
}

headers = {
    "Authorization": token
}

for _ in range(3):
    post(url, json=measurement(), headers=headers)
    time.sleep(1)

This will measure something (to be implemented) and send it, three times, with a sleep(1) in between the individual calls.

A token can be generated e.g. with openssl (openssl rand -base64 32).

Be aware that the connection in this example is not encrypted (it's a simple HTTP connection, no TLS applied). In my case I need to find out whether the ESP8622 can communicate via HTTPS at all.