mirror of
https://github.com/lynchaos/ashvale-station.git
synced 2026-09-12 12:47:49 +00:00
kemal
bda42a0468
The board carries two independent thermometers and the code averaged them into temp_raw without ever recording either. Measured over 12 samples on a real station: HTS221 30.973 C at sd 0.060, LPS25HB 29.810 C at sd 0.443, a standing gradient of 1.163 C with the SoC at 44.55 C. Two things follow from that and neither is possible without the raw channels. A plain average of a quiet sensor and one seven times noisier lands at sd 0.223 where inverse-variance weighting reaches 0.060, and the gradient between two chips at different distances from the SoC is a second observation of self-heating that could identify the compensator's k with no reference thermometer. Both need history, and history cannot be backfilled, so the columns land on their own ahead of the work that consumes them. CREATE TABLE IF NOT EXISTS is a no-op against a table that already exists, so adding to COLUMNS would have reached a fresh install and silently missed every station already running, then surfaced as an OperationalError inside insert_telemetry. That sits on the sample loop, so it takes a station down rather than leaving a gap. Store now reconciles the table against COLUMNS on open, which makes every future column addition safe rather than just this one. The simulator gains the same two channels, with couplings solved so their forward models average to exactly the k = 0.55 the compensator is tuned against. Aggregate behaviour is unchanged; only the per-channel detail is new. Simulated temp_raw noise does rise from 0.05 to 0.223, which is not a regression but the end of an over-optimistic figure: it was modelling the quiet sensor and calling it the average.
362 lines
14 KiB
Python
362 lines
14 KiB
Python
# Copyright 2026 Kemal Yaylali
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
"""Hardware access, with a simulator so the suite runs on your laptop too.
|
|
|
|
`SenseBoard` is the only place that touches `sense_hat` or `smbus2`. If
|
|
either import fails (which it will on any machine that is not a Pi), the
|
|
board falls back to `SimulatedBoard`: a small stochastic-differential
|
|
weather model that produces plausible diurnal cycles, synoptic pressure
|
|
waves and sensor noise. Train on it, develop against it, then move the
|
|
same code to the Pi unchanged.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
import numpy as np
|
|
|
|
from .physics import dew_point, sea_level_pressure, solar_position
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
TCS3400_ENABLE = 0x80
|
|
TCS3400_ATIME = 0x81
|
|
TCS3400_CONTROL = 0x8F
|
|
TCS3400_CDATA = 0x94
|
|
|
|
|
|
def read_cpu_temperature() -> float:
|
|
"""Core temperature in C. This is the single most important nuisance
|
|
variable on a Sense HAT: the HTS221 and LPS25HB sit millimetres above a
|
|
SoC that runs 30 C hotter than the room."""
|
|
try:
|
|
with open("/sys/class/thermal/thermal_zone0/temp", "r") as fh:
|
|
return float(fh.read().strip()) / 1000.0
|
|
except Exception:
|
|
return float("nan")
|
|
|
|
|
|
# Per-chip thermal coupling to the SoC, and per-chip noise.
|
|
#
|
|
# The Sense HAT carries two independent thermometers at different distances
|
|
# from the SoC, and they are not equally good. Measured over 12 samples on a
|
|
# real board: HTS221 30.973 C at sd 0.060, LPS25HB 29.810 C at sd 0.443, a
|
|
# standing gradient of 1.163 C with the SoC at 44.55 C.
|
|
#
|
|
# These two couplings are chosen so their forward models average to exactly the
|
|
# k = 0.55 the compensator is tuned against. The aggregate behaviour is
|
|
# therefore unchanged and only the per-channel detail is new, which matters
|
|
# because that gradient is a second observation of self-heating.
|
|
K_HTS221, K_LPS25HB = 0.6164, 0.4889
|
|
SD_HTS221, SD_LPS25HB = 0.060, 0.443
|
|
|
|
|
|
class SimulatedBoard:
|
|
"""Ornstein-Uhlenbeck weather with a diurnal driver. Good enough to
|
|
exercise every code path and to sanity-check a model's skill score."""
|
|
|
|
def __init__(self, latitude: float = 52.2, longitude: float = 0.12, seed: int = 7):
|
|
self.rng = np.random.default_rng(seed)
|
|
self.lat, self.lon = latitude, longitude
|
|
self.t0 = time.time()
|
|
self.press_anom = 0.0
|
|
self.temp_anom = 0.0
|
|
self.hum_anom = 0.0
|
|
self.last = self.t0
|
|
self.available = False
|
|
|
|
def _step(self, now: float) -> None:
|
|
dt = max(min(now - self.last, 600.0), 0.0)
|
|
self.last = now
|
|
# synoptic pressure: slow OU process, tau ~ 30 h, sigma ~ 9 hPa
|
|
self.press_anom += (-self.press_anom / (30 * 3600) * dt
|
|
+ 9.0 * math.sqrt(2 * dt / (30 * 3600)) * self.rng.normal())
|
|
self.temp_anom += (-self.temp_anom / (6 * 3600) * dt
|
|
+ 1.8 * math.sqrt(2 * dt / (6 * 3600)) * self.rng.normal())
|
|
self.hum_anom += (-self.hum_anom / (4 * 3600) * dt
|
|
+ 6.0 * math.sqrt(2 * dt / (4 * 3600)) * self.rng.normal())
|
|
|
|
def read(self) -> Dict[str, Any]:
|
|
now = time.time()
|
|
self._step(now)
|
|
elev, _ = solar_position(now, self.lat, self.lon)
|
|
doy = time.gmtime(now).tm_yday
|
|
seasonal = 6.5 * math.sin(2 * math.pi * (doy - 105) / 365.25)
|
|
solar_gain = 5.0 * max(elev, 0.0) / 60.0
|
|
temp = 12.0 + seasonal + solar_gain + self.temp_anom
|
|
rh = float(np.clip(78.0 - 1.9 * (temp - 12.0) + self.hum_anom, 12.0, 99.0))
|
|
press = 1013.0 + self.press_anom
|
|
lux = max(0.0, 60000.0 * max(math.sin(math.radians(max(elev, 0.0))), 0.0)) + 8.0
|
|
cpu = temp + 22.0 + 1.5 * self.rng.normal()
|
|
# forward model must invert the compensator exactly, see scripts/simulate.py
|
|
t_h = (temp + K_HTS221 * cpu) / (1.0 + K_HTS221) + SD_HTS221 * self.rng.normal()
|
|
t_p = (temp + K_LPS25HB * cpu) / (1.0 + K_LPS25HB) + SD_LPS25HB * self.rng.normal()
|
|
return {
|
|
"temp_raw": (t_h + t_p) / 2.0,
|
|
"temp_h": t_h,
|
|
"temp_p": t_p,
|
|
"hum": rh + 0.4 * self.rng.normal(),
|
|
"press": press + 0.05 * self.rng.normal(),
|
|
"cpu_temp": cpu,
|
|
"lux": lux * (0.35 + 0.65 * self.rng.random()),
|
|
"r": int(lux * 0.30), "g": int(lux * 0.34), "b": int(lux * 0.28),
|
|
"pitch": 0.4 * self.rng.normal(), "roll": 0.4 * self.rng.normal(),
|
|
"yaw": 180.0 + self.rng.normal(), "compass": 180.0 + 2 * self.rng.normal(),
|
|
"ax": 0.0, "ay": 0.0, "az": 1.0,
|
|
"gx": 0.0, "gy": 0.0, "gz": 0.0,
|
|
}
|
|
|
|
def clear(self, *_a, **_k): # LED no-op
|
|
pass
|
|
|
|
|
|
class OutdoorProbe:
|
|
"""Optional DS18B20 on the 1-Wire bus, read through the kernel's w1 driver.
|
|
|
|
Why this matters more than any model change: indoors the station forecasts
|
|
a room. Pressure passes through walls, temperature and humidity do not. One
|
|
three-pound sensor on a metre of cable outside the window removes the single
|
|
largest caveat in the project.
|
|
|
|
No new dependency. The kernel exposes each probe as a text file under
|
|
/sys/bus/w1/devices/28-*/w1_slave, so this is a file read and two string
|
|
splits. Enable with `dtoverlay=w1-gpio` in /boot/firmware/config.txt.
|
|
|
|
How it fails: the DS18B20 takes up to 750 ms to convert, and the driver
|
|
blocks for that whole time. Reading it on the 2 s sample loop would eat a
|
|
third of the budget on a single-issue core, so it is polled on its own
|
|
slower cadence and the last good value is reused in between. A probe that
|
|
goes missing (cable pulled, bad CRC) returns None rather than a stale value
|
|
forever: `age_s` lets the caller decide when to stop trusting it.
|
|
"""
|
|
|
|
ROOT = "/sys/bus/w1/devices"
|
|
|
|
def __init__(self, min_period_s: float = 20.0) -> None:
|
|
self.min_period_s = float(min_period_s)
|
|
self.device: Optional[str] = None
|
|
self.available = False
|
|
self.last_value: Optional[float] = None
|
|
self.last_ts: Optional[float] = None
|
|
self.errors = 0
|
|
self._discover()
|
|
|
|
def _discover(self) -> None:
|
|
try:
|
|
root = Path(self.ROOT)
|
|
if not root.is_dir():
|
|
return
|
|
probes = sorted(p for p in root.glob("28-*") if (p / "w1_slave").exists())
|
|
if probes:
|
|
self.device = str(probes[0] / "w1_slave")
|
|
self.available = True
|
|
log.info("outdoor probe found at %s", self.device)
|
|
except OSError as exc:
|
|
log.warning("1-wire scan failed: %r", exc)
|
|
|
|
def read(self) -> Optional[float]:
|
|
"""Celsius, or None. Cached between polls so the sample loop never blocks."""
|
|
if not self.available or self.device is None:
|
|
return None
|
|
now = time.time()
|
|
if self.last_ts is not None and (now - self.last_ts) < self.min_period_s:
|
|
return self.last_value
|
|
try:
|
|
with open(self.device, "r") as fh:
|
|
text = fh.read()
|
|
except OSError as exc:
|
|
self.errors += 1
|
|
log.warning("outdoor probe read failed: %r", exc)
|
|
return self.last_value
|
|
# Two lines: the first ends in YES only when the CRC checked out.
|
|
if "YES" not in text.split("\n")[0]:
|
|
self.errors += 1
|
|
return self.last_value
|
|
marker = text.find("t=")
|
|
if marker < 0:
|
|
self.errors += 1
|
|
return self.last_value
|
|
try:
|
|
milli = int(text[marker + 2:].strip())
|
|
except ValueError:
|
|
self.errors += 1
|
|
return self.last_value
|
|
# 85000 is the DS18B20 power-on default and means "never converted".
|
|
if milli == 85000:
|
|
self.errors += 1
|
|
return self.last_value
|
|
value = milli / 1000.0
|
|
if not (-55.0 <= value <= 125.0):
|
|
self.errors += 1
|
|
return self.last_value
|
|
self.last_value = value
|
|
self.last_ts = now
|
|
return value
|
|
|
|
def status(self) -> Dict[str, Any]:
|
|
age = None if self.last_ts is None else round(time.time() - self.last_ts, 1)
|
|
return {"available": self.available, "device": self.device,
|
|
"value_c": self.last_value, "age_s": age, "errors": self.errors}
|
|
|
|
|
|
class SenseBoard:
|
|
"""Real hardware wrapper. Attribute `available` tells you which world
|
|
you are in without try/except at every call site."""
|
|
|
|
def __init__(self, rotation: int = 90, low_light: bool = True,
|
|
tcs_addr: int = 0x39, latitude: float = 52.2, longitude: float = 0.12):
|
|
self.available = False
|
|
self.has_colour = False
|
|
self.sense = None
|
|
self.bus = None
|
|
self.tcs_addr = tcs_addr
|
|
self._sim = SimulatedBoard(latitude, longitude)
|
|
|
|
try:
|
|
from sense_hat import SenseHat # type: ignore
|
|
self.sense = SenseHat()
|
|
self.sense.low_light = low_light
|
|
self.sense.set_rotation(rotation)
|
|
self.available = True
|
|
except Exception:
|
|
self.sense = None
|
|
|
|
if self.available:
|
|
try:
|
|
import smbus2 # type: ignore
|
|
self.bus = smbus2.SMBus(1)
|
|
self.bus.write_byte_data(self.tcs_addr, TCS3400_ENABLE, 0x03) # power + RGBC
|
|
self.bus.write_byte_data(self.tcs_addr, TCS3400_ATIME, 0xD5) # 100 ms
|
|
self.bus.write_byte_data(self.tcs_addr, TCS3400_CONTROL, 0x00) # 1x gain
|
|
self.has_colour = True
|
|
except Exception:
|
|
self.has_colour = False
|
|
|
|
# ---------------------------------------------------------------- IO
|
|
|
|
def colour(self) -> Dict[str, Any]:
|
|
if not self.has_colour:
|
|
return {"clear": 0, "red": 0, "green": 0, "blue": 0, "hex": "#334155", "cct": None}
|
|
try:
|
|
data = self.bus.read_i2c_block_data(self.tcs_addr, TCS3400_CDATA | 0x80, 8)
|
|
c = data[0] | (data[1] << 8)
|
|
r = data[2] | (data[3] << 8)
|
|
g = data[4] | (data[5] << 8)
|
|
b = data[6] | (data[7] << 8)
|
|
return _colour_payload(c, r, g, b)
|
|
except Exception:
|
|
return {"clear": 0, "red": 0, "green": 0, "blue": 0, "hex": "#334155", "cct": None}
|
|
|
|
def read(self) -> Dict[str, Any]:
|
|
"""One full multi-sensor sample. Raw, uncompensated, untouched."""
|
|
if not self.available:
|
|
row = self._sim.read()
|
|
col = _colour_payload(int(row["lux"]), row["r"], row["g"], row["b"])
|
|
row.update({"lux": col["clear"], "r": col["red"], "g": col["green"],
|
|
"b": col["blue"], "colour": col, "simulated": True})
|
|
return row
|
|
|
|
s = self.sense
|
|
t_h = s.get_temperature_from_humidity()
|
|
t_p = s.get_temperature_from_pressure()
|
|
orientation = s.get_orientation_degrees()
|
|
accel = s.get_accelerometer_raw()
|
|
gyro = s.get_gyroscope_raw()
|
|
col = self.colour()
|
|
|
|
def wrap(v):
|
|
return v - 360.0 if v > 180.0 else v
|
|
|
|
return {
|
|
"temp_raw": (t_h + t_p) / 2.0,
|
|
"temp_h": t_h,
|
|
"temp_p": t_p,
|
|
"hum": s.get_humidity(),
|
|
"press": s.get_pressure(),
|
|
"cpu_temp": read_cpu_temperature(),
|
|
"lux": col["clear"], "r": col["red"], "g": col["green"], "b": col["blue"],
|
|
"colour": col,
|
|
"pitch": wrap(orientation["pitch"]),
|
|
"roll": wrap(orientation["roll"]),
|
|
"yaw": orientation["yaw"],
|
|
"compass": s.get_compass(),
|
|
"ax": accel["x"], "ay": accel["y"], "az": accel["z"],
|
|
"gx": gyro["x"], "gy": gyro["y"], "gz": gyro["z"],
|
|
"simulated": False,
|
|
}
|
|
|
|
# --------------------------------------------------------------- LED
|
|
|
|
def clear(self, *args):
|
|
if self.sense is not None:
|
|
self.sense.clear(*args)
|
|
|
|
def show_message(self, text: str, scroll_speed: float = 0.065, text_colour=None):
|
|
if self.sense is not None:
|
|
self.sense.show_message(text, scroll_speed=scroll_speed,
|
|
text_colour=text_colour or [255, 255, 255])
|
|
|
|
def set_pixels(self, pixels):
|
|
if self.sense is not None:
|
|
self.sense.set_pixels(pixels)
|
|
|
|
|
|
def _colour_payload(c: int, r: int, g: int, b: int) -> Dict[str, Any]:
|
|
denom = max(int(c), 1)
|
|
nr = min(int((r / denom) * 255), 255)
|
|
ng = min(int((g / denom) * 255), 255)
|
|
nb = min(int((b / denom) * 255), 255)
|
|
return {
|
|
"clear": int(c), "red": int(r), "green": int(g), "blue": int(b),
|
|
"hex": f"#{nr:02x}{ng:02x}{nb:02x}",
|
|
"cct": correlated_colour_temperature(r, g, b),
|
|
}
|
|
|
|
|
|
def correlated_colour_temperature(r: float, g: float, b: float) -> Optional[float]:
|
|
"""McCamy's approximation, in kelvin. Distinguishes a tungsten desk lamp
|
|
(~2700 K) from overcast daylight (~6500 K), which turns the colour sensor
|
|
into a crude `is anyone home` and `is it cloudy` detector."""
|
|
if (r + g + b) <= 0:
|
|
return None
|
|
X = -0.14282 * r + 1.54924 * g + -0.95641 * b
|
|
Y = -0.32466 * r + 1.57837 * g + -0.73191 * b
|
|
Z = -0.68202 * r + 0.77073 * g + 0.56332 * b
|
|
denom = X + Y + Z
|
|
if abs(denom) < 1e-9:
|
|
return None
|
|
x, y = X / denom, Y / denom
|
|
if abs(y - 0.1858) < 1e-9:
|
|
return None
|
|
n = (x - 0.3320) / (0.1858 - y)
|
|
cct = 449 * n ** 3 + 3525 * n ** 2 + 6823.3 * n + 5520.33
|
|
return float(cct) if 800 < cct < 25000 else None
|
|
|
|
|
|
def enrich(raw: Dict[str, Any], altitude_m: float) -> Dict[str, Any]:
|
|
"""Add derived quantities that do not need any model state."""
|
|
out = dict(raw)
|
|
temp = raw.get("temp_raw", float("nan"))
|
|
hum = raw.get("hum", float("nan"))
|
|
press = raw.get("press", float("nan"))
|
|
out["dew_c"] = float(dew_point(temp, hum))
|
|
out["press_slp"] = float(sea_level_pressure(press, temp, altitude_m))
|
|
return out
|