evo10 soil moisture sensor bl

evo10.py

#!/usr/bin/env python3
"""Connect to the EVO10 sensor over BLE and log temperature, soil-moisture,
light and DLI to evo_log.csv.  Run:  pip install bleak  &&  python3 evo_logger.py"""

import asyncio, csv, datetime as dt
from bleak import BleakScanner, BleakClient

CHAR_UUID = "00010203-0405-0607-0809-0a0b0c0d2c12"   # notify characteristic
CSV_PATH  = "evo_log.csv"


def decode(data: bytes) -> dict:
    # 20-byte packet, big-endian; last byte = sum(bytes[0:19]) & 0xFF checksum
    be = lambda i, n: int.from_bytes(data[i:i + n], "big")
    return {
        "light":         be(4, 2),
        "temperature_c": be(6, 2),
        "moisture_pct":  be(8, 2),
        "dli":           round(be(10, 4) / 1000.0, 5),
    }


async def main():
    print("Scanning for EVO10 ...")
    dev = await BleakScanner.find_device_by_filter(
        lambda d, adv: (adv.local_name or d.name or "") == "EVO10", timeout=20)
    if not dev:
        raise SystemExit("EVO10 not found — make sure it's on and nearby.")

    with open(CSV_PATH, "w", newline="") as fh:
        writer = csv.writer(fh)
        writer.writerow(["timestamp", "light", "temperature_c", "moisture_pct", "dli"])

        def on_notify(_char, data):
            if len(data) != 20:
                return
            r = decode(bytes(data))
            ts = dt.datetime.now().isoformat(timespec="seconds")
            writer.writerow([ts, r["light"], r["temperature_c"], r["moisture_pct"], r["dli"]])
            fh.flush()
            print(f"[{ts}] temp={r['temperature_c']}C  moisture={r['moisture_pct']}%  "
                  f"light={r['light']}  dli={r['dli']}", flush=True)

        print(f"Connecting to {dev.address} ...")
        async with BleakClient(dev) as client:
            await client.start_notify(CHAR_UUID, on_notify)
            print("Logging to", CSV_PATH, "— press Ctrl-C to stop.\n")
            while True:
                await asyncio.sleep(3600)


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nStopped.")