#!/usr/bin/env python3 """Pi Zero temperature service: sample DS18B20 probes and serve latest reading.""" from __future__ import annotations import json import os import threading import time from copy import deepcopy from datetime import datetime, timezone from pathlib import Path from typing import Any from flask import Flask, jsonify WATER_PROBE_IDS = ("water_1", "water_2", "water_3", "water_4") HOT_INPUT_ID = "hot_input" ALL_PROBE_IDS = WATER_PROBE_IDS + (HOT_INPUT_ID,) ROOT = Path(__file__).resolve().parent CONFIG_PATH = Path(os.environ.get("TEMP_SENSOR_CONFIG", ROOT / "config.json")) MOCK = os.environ.get("TEMP_SENSOR_MOCK", "").lower() in {"1", "true", "yes"} _lock = threading.Lock() _latest: dict[str, Any] = { "ts": None, "age_ms": None, "average_f": None, "hot_input_f": None, "sensors": [], "failed_sensors": [], } _started_at = time.time() _config: dict[str, Any] = {} def c_to_f(temp_c: float) -> float: return temp_c * 9.0 / 5.0 + 32.0 def load_config(path: Path) -> dict[str, Any]: if not path.exists(): example = ROOT / "config.example.json" raise FileNotFoundError( f"Missing {path}. Copy {example.name} to config.json and set probe IDs." ) with path.open() as f: config = json.load(f) probes = config.get("probes") or {} missing = [pid for pid in ALL_PROBE_IDS if pid not in probes] if missing: raise ValueError(f"config probes missing keys: {', '.join(missing)}") return config def list_w1_device_ids(devices_path: Path) -> list[str]: if not devices_path.exists(): return [] return sorted( p.name for p in devices_path.iterdir() if p.name.startswith("28-") ) def read_w1_temp_c(devices_path: Path, device_id: str) -> float | None: slave = devices_path / device_id / "w1_slave" try: text = slave.read_text() except OSError: return None lines = text.strip().splitlines() if len(lines) < 2 or "YES" not in lines[0]: return None marker = "t=" if marker not in lines[1]: return None try: milli_c = int(lines[1].split(marker, 1)[1]) except ValueError: return None # Datasheet uses 85000 as the power-on / fault sentinel. if milli_c == 85000: return None return milli_c / 1000.0 def mock_temp_c(probe_id: str, now: float) -> float | None: """Deterministic fake temps for local testing without hardware.""" if probe_id == HOT_INPUT_ID: return 48.0 + 0.4 * ((now // 3) % 5) index = WATER_PROBE_IDS.index(probe_id) # Simulate one missing probe occasionally for status handling. if probe_id == "water_3" and int(now) % 30 < 3: return None return 38.5 + index * 0.15 + 0.05 * ((now // 3) % 7) def read_probe(probe_id: str, device_id: str, devices_path: Path) -> dict[str, Any]: if MOCK: temp_c = mock_temp_c(probe_id, time.time()) else: temp_c = read_w1_temp_c(devices_path, device_id) if temp_c is None: return { "id": probe_id, "device_id": device_id, "role": "hot_input" if probe_id == HOT_INPUT_ID else "water", "temp_c": None, "temp_f": None, "status": "missing", } temp_f = round(c_to_f(temp_c), 2) return { "id": probe_id, "device_id": device_id, "role": "hot_input" if probe_id == HOT_INPUT_ID else "water", "temp_c": round(temp_c, 3), "temp_f": temp_f, "status": "ok", } def build_snapshot(config: dict[str, Any]) -> dict[str, Any]: devices_path = Path(config.get("w1_devices_path", "/sys/bus/w1/devices")) probes_cfg = config["probes"] sensors = [ read_probe(probe_id, probes_cfg[probe_id], devices_path) for probe_id in ALL_PROBE_IDS ] water_ok = [ s["temp_f"] for s in sensors if s["id"] in WATER_PROBE_IDS and s["status"] == "ok" ] hot = next(s for s in sensors if s["id"] == HOT_INPUT_ID) average_f = round(sum(water_ok) / len(water_ok), 2) if water_ok else None failed = [s["id"] for s in sensors if s["status"] != "ok"] ts = datetime.now(timezone.utc).isoformat() return { "ts": ts, "average_f": average_f, "hot_input_f": hot["temp_f"], "water_ok_count": len(water_ok), "sensors": sensors, "failed_sensors": failed, } def sampler_loop(config: dict[str, Any], stop: threading.Event) -> None: interval = float(config.get("sample_interval_sec", 3)) while not stop.is_set(): started = time.time() snapshot = build_snapshot(config) with _lock: global _latest _latest = snapshot elapsed = time.time() - started stop.wait(max(0.1, interval - elapsed)) def create_app(config: dict[str, Any]) -> Flask: app = Flask(__name__) @app.get("/temps") def temps(): with _lock: payload = deepcopy(_latest) if payload.get("ts"): # age_ms helps the Pi 3 reject stale packages. try: sampled = datetime.fromisoformat(payload["ts"]) age_ms = int( (datetime.now(timezone.utc) - sampled).total_seconds() * 1000 ) except ValueError: age_ms = None payload["age_ms"] = age_ms return jsonify(payload) @app.get("/health") def health(): with _lock: latest_ts = _latest.get("ts") failed = list(_latest.get("failed_sensors") or []) return jsonify( { "ok": True, "mock": MOCK, "uptime_sec": round(time.time() - _started_at, 1), "latest_ts": latest_ts, "failed_sensors": failed, "known_w1_devices": list_w1_device_ids( Path(config.get("w1_devices_path", "/sys/bus/w1/devices")) ), } ) @app.get("/devices") def devices(): """List detected 1-Wire IDs to help fill config.json.""" return jsonify( { "devices": list_w1_device_ids( Path(config.get("w1_devices_path", "/sys/bus/w1/devices")) ) } ) return app def main() -> None: global _config _config = load_config(CONFIG_PATH) stop = threading.Event() thread = threading.Thread( target=sampler_loop, args=(_config, stop), name="temp-sampler", daemon=True ) thread.start() # Prime one sample before serving traffic. with _lock: global _latest _latest = build_snapshot(_config) app = create_app(_config) try: app.run( host=_config.get("host", "0.0.0.0"), port=int(_config.get("port", 5000)), threaded=True, use_reloader=False, ) finally: stop.set() if __name__ == "__main__": main()