""" XIAO ESP32C3 Fridge Monitor - the device keeps its own history ============================================================== Companion to https://tendrl.com/blog/xiao-esp32c3-keeps-its-own-history Reads a DHT22 every 5 seconds, keeps half an hour of history on the chip, and publishes ONE summary a minute carrying 30-minute statistics (temperature swing and compressor cycle count). When a Strand workflow wants the detail, it asks; this device still has it. Two stores, and the split is the whole design: ring buffer (RAM, 1.4 KB) every 5 s the numbers the per-minute math runs over. An array of floats, iterated in place, so the summary allocates nothing. client database (RAM, 11.4 KB) every 30 s the durable, queryable history the cloud can ask for later, by time range. The one AI-set threshold is a plain JSON file, not a third store. client.db_put() takes no `key` argument, so there is no stable row to overwrite or read back through the client's helpers; and making the readings store persistent (client_db_in_memory=False) would put all 2,880 of the day's reading-writes on flash to save one number that changes rarely. One value, written rarely: a file. Why not query the database every minute instead of keeping the ring? Because on this board a db_query costs ~23 KB of transient heap **no matter how few rows it returns** — the scan is sized by the database, not the result. The application only has ~58 KB free once the client is connected, so a query belongs in the rare on-demand path, never in a loop. Measured on the board; see the blog post. Hardware: Seeed Studio XIAO ESP32C3, DHT22 on D3 (GPIO 5), USB power. Any ESP32 works. The local database needs the `btree` module, which is compiled into every ESP32 build and no other port we've tested. Setup: Provision the board from the Tendrl Device Console (it writes config.json and installs the client), then upload this file as main.py. Temperatures are Fahrenheit; the DHT22 reports Celsius, so it is converted once here at the edge and everything downstream is F. """ import gc import json import time from array import array from machine import Pin from dht import DHT22 from tendrl import Client SAMPLE_S = 5 # read the sensor this often STORE_EVERY = 6 # ...and write every 6th reading to the database (30 s) WINDOW_S = 1800 # keep 30 minutes of history SUMMARY_S = 60 # publish a summary this often RING_SLOTS = WINDOW_S // SAMPLE_S # 360 floats = 1.4 KB DB_RECORDS = WINDOW_S // (SAMPLE_S * STORE_EVERY) # 60 records = 11.4 KB sensor = DHT22(Pin(5)) # D3 on the XIAO (GPIO 5) client = Client( client_db=True, client_db_max_records=DB_RECORDS, # oldest-first eviction: a rolling window offline_storage=True, # survives a drop *and* a reboot ) LIMITS_PATH = "/limits.json" def load_limits(): """Thresholds survive a power cut because they are on flash. 7.0 F matches the service rule's upper bound on purpose. A real fridge shelf near the door settles at a ~5.6 F swing over 30 minutes, so a tighter default reports warn forever on a healthy fridge — and worse, disagrees with the dashboard, which would show "warn" next to a message that passed validation. """ try: with open(LIMITS_PATH) as f: return json.load(f) except (OSError, ValueError): return {"swing_f": 7.0} limits = load_limits() ring = array("f", bytearray(4 * RING_SLOTS)) _head = 0 # next slot to write _filled = 0 # how many slots hold real readings # Per-minute accumulators, reset after each summary. Running values rather than # a list, so a summary costs no allocation at all. _m_n = 0 _m_sum = 0.0 _m_min = None _m_max = None _read_errors = 0 # consecutive failed sensor reads def record(f): """Add one reading to the ring and to the per-minute accumulators.""" global _head, _filled, _m_n, _m_sum, _m_min, _m_max ring[_head] = f _head = (_head + 1) % RING_SLOTS if _filled < RING_SLOTS: _filled += 1 _m_n += 1 _m_sum += f if _m_min is None or f < _m_min: _m_min = f if _m_max is None or f > _m_max: _m_max = f def window_stats(): """min, max and mean-crossings over the whole ring. Allocates nothing. Crossings use a deadband: at rest a sensor jitters either side of its own mean and would otherwise report a compressor cycle every few samples. On a still bench that turned 0.2 degrees of noise into 4 phantom cycles. """ n = _filled if n == 0: return None start = (_head - n) % RING_SLOTS lo = hi = ring[start] total = 0.0 for i in range(n): v = ring[(start + i) % RING_SLOTS] total += v if v < lo: lo = v if v > hi: hi = v mid = total / n band = (hi - lo) * 0.15 if band < 0.4: band = 0.4 # never tighter than sensor noise cycles = 0 above = None for i in range(n): v = ring[(start + i) % RING_SLOTS] if v > mid + band: # `is not True` rather than `is False`, so a window that opens on a # rise counts it. Requiring a prior below-band sample dropped the # first cycle of every window and undercounted by one. if above is not True: cycles += 1 above = True elif v < mid - band: above = False return lo, hi, mid, cycles, n def status_for(swing): if swing > limits["swing_f"] * 2: return "crit" return "warn" if swing > limits["swing_f"] else "ok" # --- Inbound routes ------------------------------------------------------- # Registered BEFORE client.start(), so they are live from the first message. @client.on(tag="history-request") def send_history(message): """Answer the cloud from local storage. This is the one place a db_query is worth its ~23 KB: it runs on demand, not on a schedule, and the collect() before it makes sure that spike lands in a clean heap. Contact caps a payload at 5 KB, so the readings go as a start time, a step and a flat list of integers in tenths of a degree. """ secs = message.get("data", {}).get("seconds", WINDOW_S) gc.collect() rows = client.db_query({"_tags": "reading", "t": {"$gte": time.time() - secs}}) if not rows: return t0 = rows[0]["t"] df = [int(r["f"] * 10) for r in rows] del rows gc.collect() client.publish({ "t0": t0, "step_s": SAMPLE_S * STORE_EVERY, "df": df, }, tags=["fridge", "history-reply"]) @client.on(tag="set-limits") def set_limits(message): """Take a new threshold from the workflow and make it durable.""" limits.update(message.get("data", {})) with open(LIMITS_PATH, "w") as f: # to flash, not RAM json.dump(limits, f) client.update_state({"swing_limit_f": limits["swing_f"]}) # --- Main loop ------------------------------------------------------------ client.start() next_summary = time.time() + SUMMARY_S tick = 0 while True: # A DHT22 raises OSError on a checksum miss, which is normal and not worth # dying over: skip the sample and take the next one. try: sensor.measure() f = sensor.temperature() * 9 / 5 + 32 record(f) _read_errors = 0 tick += 1 if tick % STORE_EVERY == 0: client.db_put({"t": time.time(), "f": f}, tags=["reading"]) except OSError: # A checksum miss is routine. A run of them is a disconnected sensor, # and going quiet about it is the worst thing this device could do: # silence looks identical to a healthy fridge nobody is worried about. _read_errors += 1 if time.time() >= next_summary: stats = window_stats() if _m_n == 0: client.publish({ "status": "sensor_fault", "read_errors": _read_errors, "mean_f": None, }, tags=["fridge", "summary"], write_offline=True) next_summary = time.time() + SUMMARY_S gc.collect() continue if stats and _m_n: lo, hi, mid, cycles, n = stats swing = round(hi - lo, 2) client.publish({ "mean_f": round(_m_sum / _m_n, 2), "min_f": round(_m_min, 2), "max_f": round(_m_max, 2), "swing_30m": swing, "cycles_30m": cycles, "samples": n, "status": status_for(swing), }, tags=["fridge", "summary"], write_offline=True) _m_n = 0 _m_sum = 0.0 _m_min = None _m_max = None next_summary = time.time() + SUMMARY_S gc.collect() time.sleep(SAMPLE_S)