Extra materiály k 9. cvičení

Waveshare Enviroment Sensors Module

bme280.py

b47XAhOjJ6GgykH_S5D3ZVWb0KYHyw6jEq7xDdKToy29W9GNfb7F9GsV08N_73klELhhYl5KESu8zaK4mXdOFA==

CSV_PATH = "./lab09/pc/influx_data.csv"
 
# ---- Parse CSV ----
f = io.StringIO(response.text) #init buffer
reader = csv.DictReader(f) #pristup pomoci klicu k prvkum
 
data = {} #struktura slovniku parsed dat jednotlivych radku
 
for row in reader:
    field = row.get("_field") #najdi co je v _field (temp/humi)
 
    if field in ("temperature", "humidity"): #z fieldu vytahni temp nebo humi
        time = row["_time"] #ziskej timestamp
        value = float(row["_value"]) #ziskej hodnotu
 
        machine = row.get("_measurement", "") #ziskej machine id
        login = row.get("login", "") #ziskej login
 
        if time not in data: #vytvor zaznam pro timestamp, pokud neni
            data[time] = {
                "time": time,
                "temperature": None,
                "humidity": None,
                "login": login,
                "machine": machine
            }
 
        data[time][field] = value #zapis hodnotu k timestampu
 
# ---- Write clean CSV ----
with open(f"{CSV_PATH}", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(
        f,
        fieldnames=["time", "machine", "temperature", "humidity", "login"]
    )
    writer.writeheader()
 
    for row in data.values():
        writer.writerow(row)