import machine
import onewire
import ds18x20
import network
import socket
import time
import ntptime
import json
import os
import gc
# ── Configuration ─────────────────────────────────────────────────────────────
def load_config():
with open('config.json', 'r') as f:
return json.load(f)
CONFIG = load_config()
SSID = CONFIG['wifi_ssid']
PASSWORD = CONFIG['wifi_password']
LOG_INTERVAL = CONFIG['log_interval_seconds']
MAX_DAYS = CONFIG['max_days']
MAX_READINGS = (MAX_DAYS * 24 * 60 * 60) // 300 # based on 5-min production interval
DATA_FILE = 'data.csv'
# ── Hardware ──────────────────────────────────────────────────────────────────
# XIAO ESP32-C3: D0 = GPIO2, D1 = GPIO3
ds_pin = machine.Pin(2)
ds_sensor = ds18x20.DS18X20(onewire.OneWire(ds_pin))
led = machine.Pin(3, machine.Pin.OUT)
def read_temperature():
roms = ds_sensor.scan()
if not roms:
return None
ds_sensor.convert_temp()
time.sleep_ms(750)
temp = ds_sensor.read_temp(roms[0])
return round(temp, 2)
# ── WiFi ──────────────────────────────────────────────────────────────────────
def connect_wifi():
wlan = network.WLAN(network.STA_IF)
wlan.active(False)
time.sleep(1)
wlan.active(True)
time.sleep(1)
print('Connecting to WiFi...')
wlan.connect(SSID, PASSWORD)
timeout = 20
while not wlan.isconnected() and timeout > 0:
time.sleep(1)
timeout -= 1
print('.', end='')
if wlan.isconnected():
print('\nConnected:', wlan.ifconfig()[0])
else:
print('\nWiFi connection failed — running without network')
return wlan
# ── NTP / Time ────────────────────────────────────────────────────────────────
def sync_time():
try:
ntptime.settime()
print('NTP synced')
except Exception as e:
print('NTP sync failed:', e)
def timestamp():
"""Return current time as ISO-ish string: YYYY-MM-DD HH:MM:SS (local time)"""
offset_secs = CONFIG.get('utc_offset_hours', 0) * 3600
t = time.localtime(time.time() + offset_secs)
return '{:04d}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}'.format(
t[0], t[1], t[2], t[3], t[4], t[5])
def today_str():
offset_secs = CONFIG.get('utc_offset_hours', 0) * 3600
t = time.localtime(time.time() + offset_secs)
return '{:04d}-{:02d}-{:02d}'.format(t[0], t[1], t[2])
# ── Data Logging ──────────────────────────────────────────────────────────────
def init_data_file():
try:
os.stat(DATA_FILE)
except OSError:
with open(DATA_FILE, 'w') as f:
f.write('timestamp,temp_c,temp_f\n')
def count_lines():
count = 0
try:
with open(DATA_FILE, 'r') as f:
for _ in f:
count += 1
except OSError:
pass
return count
def trim_oldest(keep_lines):
"""Drop oldest readings, keeping header + keep_lines data rows."""
lines = []
with open(DATA_FILE, 'r') as f:
lines = f.readlines()
if len(lines) <= keep_lines + 1: # +1 for header
return
lines = [lines[0]] + lines[-(keep_lines):] # header + newest N lines
with open(DATA_FILE, 'w') as f:
f.writelines(lines)
def log_reading(temp_c):
ts = timestamp()
temp_f = round(temp_c * 9 / 5 + 32, 2)
with open(DATA_FILE, 'a') as f:
f.write('{},{},{}\n'.format(ts, temp_c, temp_f))
total = count_lines() - 1 # subtract header
if total > MAX_READINGS:
trim_oldest(MAX_READINGS)
return ts, temp_c, temp_f
# ── Data Helpers ──────────────────────────────────────────────────────────────
def load_chart_data(max_points=288):
"""Load up to max_points readings from CSV. Returns (labels, temps_c)."""
labels = []
temps = []
try:
with open(DATA_FILE, 'r') as f:
next(f) # skip header
for line in f:
line = line.strip()
if not line:
continue
parts = line.split(',')
if len(parts) >= 2:
labels.append(parts[0])
try:
temps.append(float(parts[1]))
except ValueError:
pass
except OSError:
pass
return labels[-max_points:], temps[-max_points:]
def get_today_minmax():
today = today_str()
readings = []
try:
with open(DATA_FILE, 'r') as f:
next(f)
for line in f:
line = line.strip()
if line.startswith(today):
parts = line.split(',')
if len(parts) >= 2:
try:
readings.append(float(parts[1]))
except ValueError:
pass
except OSError:
pass
if readings:
return min(readings), max(readings)
return None, None
# ── JSON array helpers (no full-list allocation) ──────────────────────────────
def send_json_strings(conn, items):
"""Send a JSON string array element by element — never allocates the full string."""
conn.send(b'[')
for i, item in enumerate(items):
if i > 0:
conn.send(b',')
conn.send(b'"')
conn.send(item.encode('utf-8'))
conn.send(b'"')
conn.send(b']')
def send_json_numbers(conn, items):
"""Send a JSON number array element by element — never allocates the full string."""
conn.send(b'[')
for i, item in enumerate(items):
if i > 0:
conn.send(b',')
conn.send(str(item).encode('utf-8'))
conn.send(b']')
# ── Dashboard (streaming) ─────────────────────────────────────────────────────
def stream_dashboard(conn, current_temp, current_ts, min_today, max_today,
chart_labels, chart_data, unit='F'):
"""Stream dashboard HTML to conn in small chunks.
HTTP header must be sent before calling this function."""
if unit == 'F':
display_temp = str(round(current_temp * 9 / 5 + 32, 1)) if current_temp else '--'
display_min = str(round(min_today * 9 / 5 + 32, 1)) if min_today is not None else '--'
display_max = str(round(max_today * 9 / 5 + 32, 1)) if max_today is not None else '--'
unit_label = 'F'
chart_values = [round(v * 9 / 5 + 32, 1) for v in chart_data]
else:
display_temp = str(current_temp) if current_temp else '--'
display_min = str(min_today) if min_today is not None else '--'
display_max = str(max_today) if max_today is not None else '--'
unit_label = 'C'
chart_values = chart_data
unit_f_active = 'active' if unit == 'F' else ''
unit_c_active = 'active' if unit == 'C' else ''
ts_str = current_ts or '--'
def send(s):
conn.send(s.encode('utf-8') if isinstance(s, str) else s)
gc.collect()
# ── & styles ───────────────────────────────────────────────────
send(''
''
''
'Freezer Monitor'
''
'')
gc.collect()
# ── Body ──────────────────────────────────────────────────────────────
send('Freezer Monitor
'
'' + display_temp + '°' + unit_label + '
'
'Last reading: ' + ts_str + '
'
''
''
''
'
'
''
'
Today Low
'
'
' + display_min + '°' + unit_label + '
'
'
Today High
'
'
' + display_max + '°' + unit_label + '
'
'
'
'
'
''
''
''
'
'
'