Faster charts and replies, spike-free air figures, tidier logs

README.md

@@ -36,12 +36,16 @@ Under systemd see `ecobot.service` (it loads the environment file; the clone's l
 | `CHART_ALL_FEELS_LIKE=on` | optional: "weather all week" (every reading, a panel each, sent with just its title; UV is left out as it has the same shape as solar) also draws the feels-like panel; off by default, as it nearly repeats temperature. Naming either ("plot feels like", "plot solar and uv") always works |
 | `TUYA_DEVICE_ID`, `TUYA_CLIENT_ID`, `TUYA_CLIENT_SECRET` | enables the irrigation controller (a Tuya smart irrigation timer, through the Tuya cloud API): an Irrigation button in private chats (it replaces Wind) with its status and buttons to water for 5 to 30 minutes or switch the valve off, pause for 24 to 72 hours (the Pause button becomes Unpause while paused), refresh, and an Alerts menu to turn the chat's irrigation battery and irrigation pause alerts on or off (or type irrigation, water, tap or sprinkler); a daily check at 18:00 that delays it 24 hours when more than 1 mm fell in the last 24 hours or Open-Meteo gives today or tomorrow a 50% or better chance of 1 mm (needs Ecowitt and `FORECAST=on` for each half); an "irrigation pause" alert when it does, and an "irrigation battery" alert below 5%. `TUYA_BASE_URL` (default `https://openapi.tuyaeu.com`, the EU data centre) and `IRRIGATION_CHECK_HOUR` (default `18`) are optional |
 | `CHART_ALL_VPD=on` | optional: "weather all week" also draws the vapour pressure deficit panel; off by default, as it is temperature and humidity combined (the station reports it for the outdoor sensor only). Naming it ("plot vpd") always works |
+| `TELEGRAM_PRELOAD_CHAT` | optional: a chat id (negative for a group or channel) the bot silently uploads charts to ahead of need. With it set, each chart sent with period buttons has its other periods drawn and uploaded there in the background (one at a time, never while a question is being answered), so pressing a period button edits the message with Telegram's copy and uploads nothing. A chart uploaded this way is used for 30 minutes, then drawn afresh. The bot must be able to post in that chat; the uploaded messages are left there. Unset: off |
+| `ANTISPIKE_REQUIRE_CONSECUTIVE` | optional: how many readings in a row a rise of the particles (PM1, PM2.5, PM10) needs to count as the air rather than a glitch of the sensor. Default `2`: a single-reading spike (at least 2.5 times, and 5 µg/m³ above, the readings either side) is left out of PM charts and of every air figure (highs, lows, averages, days over a limit), and the unhealthy-air alert counts a check only when the sensor's last 2 readings are all over the mask level. `3` also drops two-reading spikes, and so on; `1` or `off` turns the rule off. The live "air now" reading and the "safe again" check are unchanged |
+| `LOG_WIDTH` | optional: the most characters of a log line, cut with an ellipsis (…); `0` cuts none. Default 71. Every INFO line obeys it; a `WARN` or `ERROR` entry is never cut (a warning is printed as `WARN`); a request, chat message, reply or alert is also put on one line, and the startup "Alerts:" list wraps at it (at 90 when `0`) |
 
 ## Layout
 
 ```
 ecobot.py            wiring and lifecycle
 lib/config.py           settings
+lib/logs.py             how log lines look: the module that spoke, and quoted requests and chat messages cut to LOG_WIDTH
 lib/bot.py              Telegram handlers and replies; every button press comes in through one entry point that checks it first
 lib/admins.py           who may use the bot with ADMIN_ONLY: the admins of the groups it is in, plus ADMIN_CHAT_IDS
 lib/llm.py, prompt.py   tool-calling loop; system prompt
@@ -82,7 +86,9 @@ Both are off by default. They are websites, not APIs (the pollen page is parsed)
 - The model only reasons for predictions ("will it rain?") and "describe it" questions, or when asked to think, try, reason, predict, estimate, grind or whirl.
 - Simple highs/lows, chart and "air quality now" questions are fetched by the bot first, so the
   model is called once, at the end, just to word the answer.
-- Recent readings are prefetched when a question arrives and refreshed every 4 minutes.
+- Recent readings are kept in memory, refreshed every minute (the weather station) and every 4 minutes (the air sensor); a question that may need them starts a refresh as it arrives (a chart made in code doesn't: it reads just what it needs). A question never waits for a refresh in flight when the cache and memory already hold what it needs, and its own Ecowitt requests go before the background ones.
+- The chart buttons' charts are drawn ahead every 5 minutes, one at a time and never while a question is being answered: a press is sent the one drawn in the last 5½ minutes at once (its write-up kept in the chat's memory, as for one drawn on the spot); an alert's chart is still drawn when it goes out, so it shows the reading the alert is about. A chart is uploaded once: sending it again (to the next chat an alert goes to, toggling back to a period drawn lately, the next press of a button) uses Telegram's copy.
+- Each chart reply logs where its time went: `Chart ms: data 12, calc 6, draw 112, PNG 41 (22 KB), send 812`, or `Chart ms: drawn 40s ago, send 210 (Telegram's copy)` for one drawn ahead.
 - The whole history of both sensors is kept in SQLite (`ecowitt_cache.sqlite`,
   `airgradient_cache.sqlite`): a background archive copies every Ecowitt cycle (5-minute, 30-minute,
   4-hour, daily) before Ecowitt expires it, and backfills AirGradient day by day to where the sensor's

ecobot.py

@@ -22,31 +22,25 @@
 
 from lib import intent
 from lib.airgradient import AirGradient
+from lib.airgradient import metrics as air_metrics
 from lib.alerts import (PARTICULATES_CHECK_SECONDS, AlertState, ForecastMonitor, IrrigationMonitor, Notifier, ParticulatesMonitor,
                         PollenAndAsthmaMonitor, WeatherMonitor)
 from lib.alerts.menu import LABELS, kinds_for
-from lib.bot import Bot, polling_error
+from lib.bot import PREDRAW_SECONDS, Bot, polling_error
 from lib.ecowitt import Archive, Ecowitt
 from lib.forecast import Forecast
 from lib.irrigation import Tuya
 from lib.irrigation.watchdog import CHECK_SECONDS as WATCHDOG_SECONDS
 from lib.llm import Agent
+from lib.logs import LogFormat
 from lib.pollen import Pollen
 from lib.compose import Composer
 from lib.tools import Tools
 from lib.warm import every, safely
 
 
-class LogFormat(logging.Formatter):
-    """"INFO message" for this file's own lines, "INFO lib.bot: message" for the rest: the module name says who spoke."""
-
-    def format(self, record):
-        record.where = "" if record.name == "ecobot" else f"{record.name}: "
-        return super().format(record)
-
-
 _handler = logging.StreamHandler()
-_handler.setFormatter(LogFormat("%(levelname)s %(where)s%(message)s"))
+_handler.setFormatter(LogFormat("%(levelname)s %(message)s"))
 logging.basicConfig(level=logging.INFO, handlers=[_handler])
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("telegram.ext.Application").setLevel(logging.WARNING)  # "Application started" etc.
@@ -116,6 +110,9 @@ def wrapped(head: str, items: list[str], tail: str = "", width: int = 90) -> str
 async def main():
     log.info("Starting ecobot %s", version())
     cfg = Config.from_env()
+    _handler.formatter.width = cfg.log_width   # every log line is cut to it from here on
+    WRAP = max(cfg.log_width - len("INFO "), 30) if cfg.log_width else 90   # where the lists in the log wrap
+    air_metrics.REQUIRE_CONSECUTIVE = cfg.antispike_require_consecutive
     intent.FEELS_LIKE_IN_ALL = cfg.chart_all_feels_like
     intent.VPD_IN_ALL = cfg.chart_all_vpd
     sources = await start_sources(cfg)
@@ -131,7 +128,7 @@ async def main():
     agent = Agent(AsyncOpenAI(api_key=cfg.xai_api_key, base_url=cfg.xai_base_url), cfg.xai_model, tools)
     state = AlertState(cfg.state_path)
     if cfg.admin_only:
-        log.info("ADMIN_ONLY=on the bot only answers admins of known groups; set ADMIN_CHAT_IDS for more")
+        log.info("ADMIN_ONLY=on: only admins are answered (see: ADMIN_CHAT_IDS)")
         if not cfg.admin_chat_ids and not any(chat < 0 for chat in state.chats):
             log.warning("Admin only is on but no group is known yet and ADMIN_CHAT_IDS is empty, so nobody can use the bot: "
                         "add it to a group, list your user id in ADMIN_CHAT_IDS, or set ADMIN_ONLY=off")
@@ -157,8 +154,8 @@ async def main():
             await app.updater.start_polling(allowed_updates=Update.ALL_TYPES, drop_pending_updates=False,
                                             error_callback=polling_error)   # messages sent while starting are answered (lib.bot.PENDING_MAX_SECONDS)
             try:  # any startup failure below still runs the shutdown steps (and shows the real error)
-                log.info("@%s + %s (sources: %s)", app.bot.username, cfg.xai_model,
-                         ", ".join(s.name for s in sources))
+                log.info("@%s + %s", app.bot.username, cfg.xai_model)
+                log.info("Sources: %s", ", ".join(s.name for s in sources))
                 kinds = kinds_for((s.name for s in sources), irrigation is not None)
                 notify = Notifier(app.bot, state, kinds)
                 await bot.refresh_keyboards(app.bot)  # chats whose buttons are out of date are told, with the new ones
@@ -185,7 +182,7 @@ async def main():
 
                 # Keeping warm: everything questions need, refreshed before they arrive
                 tasks.extend(asyncio.create_task(s.warmer.run()) for s in sources)
-                log.info("Keeping warm: %s", ", ".join(f"{s.name} {s.warmer.interval:.0f}s" for s in sources))
+                log.info("%s", wrapped("Keeping warm: ", [f"{s.name} {s.warmer.interval:.0f}s" for s in sources], "", WRAP))
 
                 async def startup_warmup():
                     """Fetch everything once, together, and log one summary line."""
@@ -214,7 +211,7 @@ async def air_part():
                     async def air_backfill():
                         started = time.monotonic()
                         fetched, failed, total = await air.backfill()
-                        log.info("AirGradient archive: %d day(s) cached, %d failed, %d day(s) held, %.0fs",
+                        log.info("AirGradient archive: %d days cached, %d failed, %d held, %.0fs",
                                  fetched, failed, total, time.monotonic() - started)
 
                     try:
@@ -223,8 +220,9 @@ async def air_backfill():
                         log.exception("Startup warm-up failed (the regular refreshes will catch up)")
                     log.info("Startup warm-up: %s, %.1fs", ", ".join(parts) or "nothing to fetch",
                              time.monotonic() - started)
+                    tasks.append(asyncio.create_task(every(PREDRAW_SECONDS, bot.predraw)))   # the button charts, ready before they are pressed
                     on = [k for k in LABELS if k in running]   # the line says what started, not what is configured
-                    log.info("%s", wrapped("Alerts: ", [LABELS[k] for k in on] or ["none"], f": {len(state.alert_chats())} chats"))
+                    log.info("%s", wrapped("Alerts: ", [LABELS[k] for k in on] or ["none"], f": {len(state.alert_chats())} chats", WRAP))
                     if missing := [LABELS[k] for k in kinds if k not in running]:
                         log.warning("Alerts NOT running: %s (see the error above)", ", ".join(missing))
 

lib/airgradient/days.py

@@ -10,7 +10,7 @@
 
 from ..daytable import analyse, grouped_as_asked, parameters, validate
 from ..timeutil import day_bounds, local_date, now_local
-from .metrics import ALL_METRICS, CHART_UNITS
+from .metrics import ALL_METRICS, CHART_UNITS, without_spikes
 
 log = logging.getLogger(__name__)
 
@@ -31,7 +31,7 @@
 def daily_values(rows: list[dict], shown: list[str], tz: tzinfo) -> dict[str, dict[date, float]]:
     """{field: {local day: that day's highest, lowest or mean}} from the stored readings."""
     by_day: dict[date, list[dict]] = {}
-    for r in rows:
+    for r in without_spikes(rows):   # a single-reading spike of the particles is a glitch of the sensor, not a day's highest
         by_day.setdefault(local_date(r["ts"], tz), []).append(r)
     values: dict[str, dict[date, float]] = {n: {} for n in shown}
     for name in shown:

lib/airgradient/metrics.py

@@ -37,6 +37,35 @@
 # is left out.
 AIR_PANELS = (("pm1", "pm2_5", "pm10"), ("co2",), ("voc_index",), ("nox_index",))
 
+PARTICLES = AIR_PANELS[0]
+SPIKE_RATIO = 2.5    # a particle reading that is this many times the readings either side of it ...
+SPIKE_RISE = 5.0     # ... and this far (µg/m³) above both, with nothing like it next to it, is a glitch of the sensor, not the air
+SPIKE_NEIGHBOUR = 900   # the readings either side must be within this many seconds
+REQUIRE_CONSECUTIVE = 2   # ANTISPIKE_REQUIRE_CONSECUTIVE: readings in a row a rise needs to be the air (set at startup); 1 turns the rule off
+
+
+def without_spikes(rows: list[dict], require: int | None = None) -> list[dict]:
+    """The readings with each spike of the particles taken out, for drawing and for the figures: a stretch of fewer than `require`
+    readings in a row (REQUIRE_CONSECUTIVE) of PM1, PM2.5 or PM10 that are each at least SPIKE_RATIO times, and SPIKE_RISE above, the readings
+    either side of the stretch. All the particle readings of those moments go (they spike together). A rise that lasts `require`
+    readings or more is the air, and stays; so do the first and last readings (they have one side). require 1: nothing is taken out."""
+    require = REQUIRE_CONSECUTIVE if require is None else require
+    drop: set[int] = set()
+    for name in PARTICLES:
+        have = [(i, r["ts"], r[name]) for i, r in enumerate(rows) if name in r]
+        for a in range(1, len(have) - 1):
+            for length in range(1, require):
+                b = a + length - 1                                   # the stretch is have[a..b]
+                if b + 1 >= len(have):
+                    break
+                (_, t0, before), (_, t1, after) = have[a - 1], have[b + 1]
+                near = max(before, after)
+                if have[a][1] - t0 <= SPIKE_NEIGHBOUR and t1 - have[b][1] <= SPIKE_NEIGHBOUR and all(
+                        v >= SPIKE_RATIO * near and v - near >= SPIKE_RISE for _, _, v in have[a:b + 1]):
+                    drop.update(i for i, _, _ in have[a:b + 1])
+    return [{k: v for k, v in r.items() if k not in PARTICLES} if i in drop else r for i, r in enumerate(rows)]
+
+
 # US EPA 2024 PM2.5 breakpoints: (conc low, conc high, AQI low, AQI high, band)
 PM25_AQI = [
     (0.0, 9.0, 0, 50, "good"),

lib/airgradient/source.py

@@ -31,7 +31,8 @@
 from ..tools import Tool, Turn
 from ..warm import Warmer
 from .days import DESCRIPTION as DAYS_DESCRIPTION, PARAMETERS as DAYS_PARAMETERS, air_days_tool
-from .metrics import AIR_PANELS, ALL_METRICS, MARK_LOW, LABELS, METRICS, epoch, normalise, pm25_aqi, rating, value_of
+from . import metrics
+from .metrics import AIR_PANELS, ALL_METRICS, MARK_LOW, LABELS, METRICS, epoch, normalise, pm25_aqi, rating, value_of, without_spikes
 from .store import AirStore
 
 log = logging.getLogger(__name__)
@@ -253,6 +254,7 @@ async def handle(self, args: dict, turn: Turn | None = None) -> str:
                                   ensure_ascii=False)
             reading = await self.current()
             reading.pop("_time_utc", None)
+            reading.pop("_sustained", None)
             return json.dumps(reading, ensure_ascii=False)
         except Exception as e:
             log.warning("AirGradient request failed: %s", e)
@@ -268,6 +270,17 @@ async def _previous_row(self, ts: int) -> dict | None:
         rows = [r for r in await self._day_rows(today) if r["ts"] < ts]
         return rows[-1] if rows and ts - rows[-1]["ts"] <= MAX_GAP_SECONDS else None
 
+    async def _previous_rows(self, ts: int, count: int) -> list[dict]:
+        """Up to `count` of today's cached readings just before `ts`, each within MAX_GAP_SECONDS of the next, oldest first."""
+        today = to_local(ts, self.tz).date()
+        earlier, got, after = [r for r in await self._day_rows(today) if r["ts"] < ts], [], ts
+        for r in reversed(earlier):
+            if len(got) >= count or after - r["ts"] > MAX_GAP_SECONDS:
+                break
+            got.insert(0, r)
+            after = r["ts"]
+        return got
+
     async def current(self) -> dict:
         """The latest reading. A rating (and PM2.5's AQI band) is the level the last two readings both reached, so a passing
         spike is not called poor until the next reading agrees; the value shown is always the latest one."""
@@ -289,9 +302,16 @@ async def current(self) -> dict:
         if "pm2_5" in out:
             aqi, band = pm25_aqi(out["pm2_5"]["_rated"])
             out["pm2_5"].update(aqi_us=aqi, band=band)
+        sustained = {}   # each metric's lowest value over the last REQUIRE_CONSECUTIVE readings: what the air has been for that long
+        if row.get("timestamp"):
+            earlier = await self._previous_rows(epoch(row["timestamp"]), max(metrics.REQUIRE_CONSECUTIVE - 1, 0))
+            for name, entry in out.items():
+                if isinstance(entry, dict) and "value" in entry:
+                    sustained[name] = min([entry["value"], *(r[name] for r in earlier if name in r)])
         for entry in out.values():
             if isinstance(entry, dict):
                 entry.pop("_rated", None)
+        out["_sustained"] = sustained   # for the alerts (removed before the model sees it)
         return out
 
     async def rows(self, t0: datetime, t1: datetime) -> tuple[list[dict], int, list[date]]:
@@ -322,10 +342,12 @@ async def history(self, start: str | None, end: str | None, chart: bool = False,
         if t0 >= t1:
             return {"error": "start must be before end"}
         chart = wants_chart({"chart": chart}, turn, t0, t1)
-        before = self.requests
+        before, started = self.requests, time.perf_counter()
         rows, hourly, skipped = await self.rows(t0, t1)
-        log.info("AirGradient %s to %s: %d readings, %d req", f"{t0:%Y-%m-%d}", f"{t1:%m-%d %H:%M}", len(rows),
-                 self.requests - before)
+        turn.times["data"] = turn.times.get("data", 0.0) + time.perf_counter() - started   # for the reply's timing line
+        rows = without_spikes(rows)   # a single-reading spike of the particles is a glitch of the sensor: in no figure or chart
+        log.log(logging.DEBUG if turn.ahead else logging.INFO, "AirGradient %s to %s: %d readings, %d req", f"{t0:%Y-%m-%d}",
+                f"{t1:%m-%d %H:%M}", len(rows), self.requests - before)
         out = {"sensor_type": "outdoor", "period": f"{t0:%a} {t0.day} {t0:%b %Y} - {t1:%a} {t1.day} {t1:%b %Y}",
                "readings": len(rows)}
         for name, (_, unit) in METRICS.items():
@@ -376,6 +398,7 @@ def _line(self, name: str, rows: list[dict]) -> tuple[Line, Plotted] | None:
     def _chart(self, names: list[str], rows: list[dict], period: str) -> Chart | None:
         """These metrics as one chart. One metric is drawn large, with its rating zones. Several go into panels on a
         shared time axis, grouped by AIR_PANELS (the particles together, the rest each alone)."""
+        rows = without_spikes(rows)   # (a no-op for the rows history() gives it; a caller with its own rows gets the same drawing)
         drawn = {n: got for n in names if (got := self._line(n, rows))}
         if not drawn:
             return None