diff --git a/README.md b/README.md index 342d0a8..d688572 100644 --- a/README.md +++ b/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 diff --git a/ecobot.py b/ecobot.py index 903eda0..aa35baf 100644 --- a/ecobot.py +++ b/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)) diff --git a/lib/airgradient/days.py b/lib/airgradient/days.py index d0bcd3d..9776b4e 100644 --- a/lib/airgradient/days.py +++ b/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: diff --git a/lib/airgradient/metrics.py b/lib/airgradient/metrics.py index 06286b8..ac082ce 100644 --- a/lib/airgradient/metrics.py +++ b/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"), diff --git a/lib/airgradient/source.py b/lib/airgradient/source.py index 9aa1a51..294cbbc 100644 --- a/lib/airgradient/source.py +++ b/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 diff --git a/lib/alerts/notify.py b/lib/alerts/notify.py index 66f8a41..437e32b 100644 --- a/lib/alerts/notify.py +++ b/lib/alerts/notify.py @@ -9,11 +9,13 @@ import json import logging import os +import time from datetime import date from telegram import MessageEntity from telegram.error import BadRequest, ChatMigrated, Forbidden, TelegramError +from ..logs import QUOTES from .menu import ALL, LABELS, keyboard log = logging.getLogger(__name__) @@ -194,13 +196,19 @@ def __init__(self, bot, state: AlertState, kinds: list[str] | None = None): """kinds: the alert types this bot sends (the settings buttons under each alert list them).""" self.bot, self.state, self.kinds = bot, state, kinds or list(LABELS) - async def __call__(self, text: str, link: tuple[str, str] | None = None, kind: str | None = None, photo: bytes | None = None): + async def __call__(self, text: str, link: tuple[str, str] | None = None, kind: str | None = None, photo: bytes | str | None = None): """Send an alert (silently) to every chat subscribed to its type; forget chats the bot can no longer post to. link = (label, url) adds a clickable label to the footer. Each carries the settings buttons. - photo: a picture the text is the caption of (the text alone if Telegram refuses the picture).""" + photo: a picture the text is the caption of (the text alone if Telegram refuses the picture): a PNG, or the file id of one + Telegram holds. It is uploaded once: the chats after the first are sent Telegram's copy.""" text, entities = with_footer(text, link) - sent = sum([await self._send(chat_id, text, entities, kind, photo) for chat_id in self.state.alert_chats(kind)]) - log.info("Alert sent to %d chat(s): %s", sent, text.replace("\n", " ")) + started, sent = time.monotonic(), 0 + for chat_id in self.state.alert_chats(kind): + ok, message = await self._send(chat_id, text, entities, kind, photo) + sent += ok + if photo and (sizes := getattr(message, "photo", None)): + photo = sizes[-1].file_id # the largest size: the picture as sent + log.info("Alert sent to %d chat(s) in %.1fs: %s", sent, time.monotonic() - started, text, extra=QUOTES) async def to_chat(self, chat_id: int, text: str, link: tuple[str, str] | None = None, kind: str | None = None) -> bool: """An alert for one chat (the one that was sent something that has since changed), if it is subscribed to the type. @@ -208,13 +216,13 @@ async def to_chat(self, chat_id: int, text: str, link: tuple[str, str] | None = if not self.state.subscribed(chat_id, kind): return False text, entities = with_footer(text, link) - sent = await self._send(chat_id, text, entities, kind) - log.info("Alert %s chat %s: %s", "sent to" if sent else "NOT sent to", chat_id, text.replace("\n", " ")) + sent, _ = await self._send(chat_id, text, entities, kind) + log.info("Alert %s chat %s: %s", "sent to" if sent else "NOT sent to", chat_id, text, extra=QUOTES) return sent - async def _post(self, chat_id: int, text: str, entities, kind: str | None, photo: bytes | None): - """The message itself: silent unless the chat turned sound on for this type, with the settings buttons (which highlight this alert's own - type). With a photo it is the picture with the text as its caption; a picture Telegram refuses falls back to the text alone.""" + async def _post(self, chat_id: int, text: str, entities, kind: str | None, photo: bytes | str | None): + """The message itself (returned): silent unless the chat turned sound on for this type, with the settings buttons (which highlight this + alert's own type). With a photo it is the picture with the text as its caption; a picture Telegram refuses falls back to the text alone.""" common = dict(disable_notification=kind not in self.state.loud(chat_id), reply_markup=keyboard(self.state.muted(chat_id), self.kinds, parent=kind, loud=self.state.loud(chat_id) if chat_id > 0 else None)) if photo: @@ -226,14 +234,13 @@ async def _post(self, chat_id: int, text: str, entities, kind: str | None, photo log.warning("Alert picture to %s failed (%s); sending the text alone", chat_id, e) return await self.bot.send_message(chat_id, text, entities=entities, disable_web_page_preview=True, **common) - async def _send(self, chat_id: int, text: str, entities, kind: str | None = None, photo: bytes | None = None) -> bool: - """Send an alert (see _post); forget a chat the bot can no longer post to.""" + async def _send(self, chat_id: int, text: str, entities, kind: str | None = None, photo: bytes | str | None = None) -> tuple[bool, object]: + """Send an alert (see _post); forget a chat the bot can no longer post to. (whether it was sent, the message sent)""" try: - await self._post(chat_id, text, entities, kind, photo) - return True + return True, await self._post(chat_id, text, entities, kind, photo) except ChatMigrated as e: # the group became a supergroup: follow it, and send again there self.state.migrate_chat(chat_id, e.new_chat_id) - return await self._send(e.new_chat_id, text, entities, kind, photo) if e.new_chat_id != chat_id else False + return await self._send(e.new_chat_id, text, entities, kind, photo) if e.new_chat_id != chat_id else (False, None) except Forbidden as e: # kicked from the group, or blocked in a private chat self.state.remove_chat(chat_id, f"can't post: {e}") except BadRequest as e: @@ -243,4 +250,4 @@ async def _send(self, chat_id: int, text: str, entities, kind: str | None = None log.warning("Alert to %s failed: %s", chat_id, e) except TelegramError as e: log.warning("Alert to %s failed: %s", chat_id, e) - return False + return False, None diff --git a/lib/alerts/particulates.py b/lib/alerts/particulates.py index 0c1059d..e75f7f7 100644 --- a/lib/alerts/particulates.py +++ b/lib/alerts/particulates.py @@ -1,7 +1,8 @@ """Particulates alerts (PM2.5 and PM10) from the AirGradient sensor, checked every 30 minutes. "Wear a mask outside" when PM2.5 or PM10 reaches US AQI 151+ ("unhealthy", the official level -at which masks are advised for everyone) in two readings in a row; "back to safe" when both are at +at which masks are advised for everyone) in two readings in a row, each check counting only when the sensor's last +ANTISPIKE_REQUIRE_CONSECUTIVE (2) readings are all that high (a spike never does); "back to safe" when both are at AQI 100 or below in two readings in a row (confirm.py: a check that brings no new reading counts for nothing). Every episode gets the pair. Particles only: masks filter particles, not gases like CO2/VOC/NOx. """ @@ -39,7 +40,9 @@ async def check(self, reading: dict | None = None, now: datetime | None = None): return if when is not None and not confirm.new_reading(m, when.timestamp()): # the sensor has said nothing new since the last check return - bad = (pm25 is not None and pm25 >= MASK_PM25) or (pm10 is not None and pm10 >= MASK_PM10) + held = reading.get("_sustained") or {} # the lowest of the last readings (ANTISPIKE_REQUIRE_CONSECUTIVE): a spike alone can't make the air unhealthy + high25, high10 = held.get("pm2_5", pm25), held.get("pm10", pm10) + bad = (high25 is not None and high25 >= MASK_PM25) or (high10 is not None and high10 >= MASK_PM10) good = (pm25 is None or pm25 <= SAFE_PM25) and (pm10 is None or pm10 <= SAFE_PM10) confirm.run(m, "above", bad) confirm.run(m, "below", good) diff --git a/lib/bot.py b/lib/bot.py index 259008a..a8ac48c 100644 --- a/lib/bot.py +++ b/lib/bot.py @@ -31,6 +31,7 @@ from .config import Config from .timeutil import now_local from .llm import Agent, shorten_old, strip_tool_turns, trim_history +from .logs import QUOTES log = logging.getLogger(__name__) @@ -47,6 +48,7 @@ STALE_BUTTON = "That button is out of date, please ask again" MAX_CALLBACK_BYTES = 64 # Telegram's own limit on a button's data WATCHDOG_SECONDS = 45 # a question still running after this many seconds logs where everything is waiting +PREDRAW_SECONDS = 300 # the button charts are drawn ahead this often (a press finds one drawn within periods.ImageCache.FRESH) HELP = ("Hi! Message me directly, or in groups @mention me or reply to me.\n" "/reset clears this chat's memory, /alerts opens the alert settings (buttons to subscribe or unsubscribe), /usage shows what I can do.\n" @@ -74,6 +76,19 @@ def _short(text: str, limit: int = 200) -> str: return text if len(text) <= limit else text[:limit] + "..." +def chart_times(times: dict, made: float | None, pngs: list[bytes], sending: float) -> str: + """A reply's chart timings for the log: "data 12, calc 3, draw 98, PNG 41 (23 KB), send 812" (ms). times is + the question's Turn.times; made is how long the tool took when the chart was made in code (the calculation is that less its + data), else None. A step that wasn't timed is left out.""" + ms = lambda seconds: f"{seconds * 1000:.0f}" + parts = [f"data {ms(times['data'])}"] if "data" in times else [] + if made is not None: + parts.append(f"calc {ms(max(0.0, made - times.get('data', 0.0)))}") + if "draw" in times: + parts += [f"draw {ms(times['draw'])}", f"PNG {ms(times['png'])} ({sum(map(len, pngs)) / 1024:.0f} KB)"] + return ", ".join([*parts, f"send {ms(sending)}"]) + + def split_message(text: str, limit: int = TG_LIMIT) -> list[str]: chunks = [] while len(text) > limit: @@ -230,7 +245,15 @@ def __init__(self, cfg: Config, agent: Agent, sources: list, state: AlertState | self.admins = Admins(state, getattr(cfg, "admin_chat_ids", ())) self.chats: dict[tuple, ChatState] = defaultdict(ChatState) self.by_name = {s.name: s for s in sources} - self.images = periods.ImageCache() # charts drawn lately, so toggling the period buttons is instant + self.images = periods.ImageCache() # charts drawn lately, so toggling the period buttons is instant (and the buttons' drawn ahead) + self.answering = 0 # questions being answered right now: charts are drawn ahead only when there are none + self.idle = asyncio.Event() + self.idle.set() + self.predrawn = False # the first round of charts drawn ahead has been logged + self.predraw_failing: set[str] = set() # button charts that couldn't be drawn ahead (said once, until they can again) + self.preloading: set[str] = set() # questions whose other periods are being uploaded (TELEGRAM_PRELOAD_CHAT) + self.preload_off = False # Telegram refused an upload to that chat: no more are tried until restart + self.background: set[asyncio.Task] = set() # tasks running on their own, kept so they aren't dropped self.charted = periods.Charted(state.charts if state else None, state.save if state else None) # the question behind each chart sent (kept across restarts) def register(self, app): @@ -590,23 +613,27 @@ async def on_period_button(self, update: Update, context: ContextTypes.DEFAULT_T return await query.answer(f"Drawing {periods.NAMES[days]}...") asked = intent.with_period(question, days, now) - log.info("Period button: %s -> %s", _short(question, 40), _short(asked, 40)) - if cached := self.images.get(self.images.key(asked, days, now)): # drawn lately: no new question - png, title = cached + log.info("Period button %s: %s", periods.NAMES[days], question, extra=QUOTES) + key = self.images.key(asked, days, now) + if cached := self.images.get(key): # drawn lately: no new question + photo, title = cached try: - await message.edit_media(InputMediaPhoto(png, caption=title or None), reply_markup=periods.keyboard(days)) + began = time.monotonic() + self.images.sent(key, await message.edit_media(InputMediaPhoto(photo, caption=title or None), reply_markup=periods.keyboard(days))) self.charted.remember(message.chat_id, message.message_id, asked) - log.info("Period button: chart from the cache") + self.preload_soon(context.bot, asked) + log.info("Period button: cached chart sent in %.0f ms%s", (time.monotonic() - began) * 1000, + " (Telegram's copy)" if isinstance(photo, str) else "") return except TelegramError as e: log.warning("Couldn't edit the chart in place (%s); drawing it again", e) await self.respond(update, context, asked, redraw=message) - async def _redraw(self, message: Message, reply: str, photo: bytes, title: str, row, question: str) -> bool: + async def _redraw(self, message: Message, reply: str, photo: bytes, title: str, row, question: str, key: str | None = None) -> bool: """Put a new chart into the message a period button was pressed under (its answer text, if any, follows as a message of - its own). False when Telegram refuses, and the chart is then sent as a new message.""" + its own). False when Telegram refuses, and the chart is then sent as a new message. key: where the image cache keeps it.""" try: - await message.edit_media(InputMediaPhoto(photo, caption=title or None), reply_markup=row) + self.images.sent(key, await message.edit_media(InputMediaPhoto(photo, caption=title or None), reply_markup=row)) except BadRequest as e: if "not modified" not in str(e).lower(): log.warning("Couldn't edit the chart in place (%s); sending a new one", e) @@ -710,7 +737,7 @@ async def on_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE): return waited = (datetime.now(UTC) - msg.date).total_seconds() if getattr(msg, "date", None) else 0 if waited > PENDING_MAX_SECONDS: - log.info("Ignored a message %d minutes old from %s: %s", waited // 60, describe_source(update), _short(msg.text, 40)) + log.info("Ignored a message %d minutes old from %s: %s", waited // 60, describe_source(update), msg.text, extra=QUOTES) return private = msg.chat.type == ChatType.PRIVATE if private and not await self._allowed(update, context): # before remember_chat: a stranger never becomes an alert chat @@ -762,7 +789,7 @@ async def _written_in_code(self, read, turn: Turn) -> str: async def _warn(self, msg: Message, tokens: int): """A heads-up before a big job: the question has pulled in a lot of data. Sent now, silently; the answer follows by itself.""" text = f"\u26a0\ufe0f That's pulling in about {tokens / 1000:.0f}k tokens of data, so it will take a little longer. Working on it..." - log.info("Heads-up sent: %s", text) + log.info("Heads-up sent: %s", text, extra=QUOTES) with contextlib.suppress(TelegramError): await msg.reply_text(text) @@ -806,7 +833,105 @@ async def _ask_model(self, working: list[dict], system: str, read, turn: Turn, d working[:] = before # the unfinished attempt's turns are dropped effort, budget, retried = lower, RETRY_SECONDS, True + async def _draw_ahead(self, text: str, now) -> tuple[str, bytes, str, str] | None: + """(image key, PNG, title, write-up) of the chart `text` asks for when it is one made in code with its own words as its key; + None otherwise. Raises when it can't be drawn. Nobody is waiting: its requests are not urgent and its logs are quiet.""" + read = intent.read(text, now, "Ecowitt" in self.by_name, "AirGradient" in self.by_name, "Pollen" in self.by_name, + "Forecast" in self.by_name) + key = self.images.canonical(text, now) + if not (read.fast and read.chart_in_code and key): # not a chart made in code ("Report"), or a source that isn't there + return None + turn = Turn(chart_asked=read.chart_asked, chart_field=read.chart_field, chart_fields=read.chart_fields, + average_asked=read.average_asked, readings=read.readings, per_day=read.per_day, text=text, ahead=True) + caption = await asyncio.wait_for(self._chart_in_code(read, turn), TURN_SECONDS) + if caption is None or len(turn.charts) != 1: + raise ValueError("no chart came of it") + png = await asyncio.to_thread(render_chart, turn.charts[0], self.cfg.tz) + return key, png, turn.charts[0].caption(now.date()), caption + + def preload_soon(self, bot, question: str): + """With TELEGRAM_PRELOAD_CHAT set, upload the other periods of this chart's question in the background.""" + chat = getattr(self.cfg, "telegram_preload_chat", None) + if chat is None or self.preload_off or question in self.preloading: + return + self.preloading.add(question) + task = asyncio.create_task(self.preload(bot, chat, question)) + self.background.add(task) + task.add_done_callback(self.background.discard) + + async def preload(self, bot, chat: int, question: str): + """Upload (silently, to `chat`) the chart for each other period of `question`, so a press of a period button edits the message by + Telegram's copy and uploads nothing. One at a time and never while a question is being answered.""" + started, done = time.monotonic(), [] + try: + for days in periods.PERIODS: + await self.idle.wait() + now = now_local(self.cfg.tz) + if days == intent.period_days(question, now): + continue + asked = intent.with_period(question, days, now) + if self.images.has_copy(self.images.key(asked, days, now)): + continue + try: + drawn = await self._draw_ahead(asked, now) + except Exception as e: + log.debug("Couldn't draw %r to preload: %s: %s", asked, type(e).__name__, e) + continue + if drawn is None: + return + key, png, title, caption = drawn + self.images.put(key, png, title, caption, ttl=self.images.PRELOAD_TTL) + try: + self.images.sent(key, await bot.send_photo(chat, png, caption=title or None, disable_notification=True)) + except TelegramError as e: + self.preload_off = True + log.warning("Couldn't upload to TELEGRAM_PRELOAD_CHAT %s (%s: %s); not preloading charts", chat, type(e).__name__, e) + return + done.append(periods.NAMES[days]) + finally: + self.preloading.discard(question) + if done: + log.info("Preloaded %s of %r to chat %s in %.1fs", ", ".join(done), question, chat, time.monotonic() - started, + extra=QUOTES) + + async def predraw(self): + """Draw the charts the buttons ask for ahead of the press: one pressed in the next ImageCache.FRESH seconds is sent at once. + One chart at a time, and never while a question is being answered (its own drawing would wait behind these). Run every + PREDRAW_SECONDS; the first round is logged, and a chart that can't be drawn once, until it can again.""" + started, done = time.monotonic(), [] + for text in dict.fromkeys(question for row in templates.rows(self.irrigation is not None) for _, question in row if question): + await self.idle.wait() + try: + drawn = await self._draw_ahead(text, now_local(self.cfg.tz)) + if drawn is None: + continue + except Exception as e: + if text not in self.predraw_failing: + self.predraw_failing.add(text) + log.warning("Can't draw %r ahead: %s: %s", text, type(e).__name__, e) + continue + key, png, title, caption = drawn + self.images.put(key, png, title, caption) + done.append(text) + if text in self.predraw_failing: + self.predraw_failing.discard(text) + log.info("Drawn ahead again: %s", text) + if not self.predrawn: + self.predrawn = True + log.info("Button charts drawn ahead every %ds: %d in %.1fs", PREDRAW_SECONDS, len(done), time.monotonic() - started) + async def respond(self, update: Update, context: ContextTypes.DEFAULT_TYPE, text: str, redraw: Message | None = None): + """Answer a question (_respond), counted while it runs: no chart is drawn ahead meanwhile.""" + self.answering += 1 + self.idle.clear() + try: + await self._respond(update, context, text, redraw) + finally: + self.answering -= 1 + if not self.answering: + self.idle.set() + + async def _respond(self, update: Update, context: ContextTypes.DEFAULT_TYPE, text: str, redraw: Message | None = None): msg = update.effective_message text = text.strip() if not text: @@ -819,9 +944,9 @@ async def respond(self, update: Update, context: ContextTypes.DEFAULT_TYPE, text read = intent.read(text, now, "Ecowitt" in self.by_name, "AirGradient" in self.by_name, "Pollen" in self.by_name, "Forecast" in self.by_name) log.info("%s %s%s", describe_source(update), f"(reasoning: {read.effort}) " if read.effort != intent.EFFORT_DEFAULT else "", - _short(text)) + text, extra=QUOTES) for source in self.sources: # fetch recent readings while the model thinks - if source.wants(text): + if source.wants(text) and not (read.fast and read.chart_in_code): # a chart made in code reads just what it needs, first in line source.poke() if read.fast: log.info("Fast path: %s", read.fast[2]) @@ -837,13 +962,19 @@ async def respond(self, update: Update, context: ContextTypes.DEFAULT_TYPE, text working = [*chat.history, {"role": "user", "content": self._content(msg, text, context.bot.id)}] new_from = len(working) ok, photos, titles, silent = True, [], [], False + made = None # how long the tool took when the chart is made in code, for the timing line turn = Turn(chart_asked=read.chart_asked, chart_field=read.chart_field, chart_fields=read.chart_fields, average_asked=read.average_asked, readings=read.readings, per_day=read.per_day, text=text) + ready = self.images.ready(text, now) if read.fast and read.chart_in_code and redraw is None else None try: reply = None - if read.fast and read.chart_in_code: # drawn by the tool; the chart goes out with just its title (the text is kept for the history) + if ready: # drawn moments ago (ahead, or for someone's same question): sent as it is, its text kept for the history + reply, photos, titles, silent = ready.caption, [ready.photo], [ready.title], True + elif read.fast and read.chart_in_code: # drawn by the tool; the chart goes out with just its title (the text is kept for the history) + began = time.monotonic() reply = await asyncio.wait_for(self._chart_in_code(read, turn), TURN_SECONDS) silent = reply is not None + made = time.monotonic() - began if silent else None elif read.lookup: reply = await asyncio.wait_for(self._lookup_in_code(read, turn), TURN_SECONDS) if reply is None: @@ -862,12 +993,12 @@ async def respond(self, update: Update, context: ContextTypes.DEFAULT_TYPE, text chat.history = shorten_old(trim_history(strip_tool_turns(working))) for spec in turn.charts[:MAX_CHARTS]: # drawn while "typing..." is still showing try: - photos.append(await asyncio.to_thread(render_chart, spec, self.cfg.tz)) + photos.append(await asyncio.to_thread(render_chart, spec, self.cfg.tz, turn.times)) titles.append(spec.caption(now.date())) except Exception: log.exception("Chart failed; sending the answer without it") except TimeoutError: - log.error("Gave up after %ds on: %s", TURN_SECONDS, _short(text, 60)) + log.error("Gave up after %ds on: %s", TURN_SECONDS, text, extra=QUOTES) ok, reply = False, "Sorry, that took too long. Please try again in a moment." except Exception as e: log.exception("Agent error") @@ -878,31 +1009,42 @@ async def respond(self, update: Update, context: ContextTypes.DEFAULT_TYPE, text stop_typing.set() await asyncio.gather(*[t for t in (typing, drafting) if t]) # wait for any in-flight "typing" or draft so none is sent after the reply + caption = reply if silent else None # a chart made in code: its text, kept with the chart for the next to ask the same if silent and photos: reply = "" if ok and not reply.strip() and not photos: # a chart with no text that could not be drawn reply = "Sorry, I couldn't draw that chart. Please try again in a moment." used = [m for m in working[new_from:] if m["role"] == "tool"] log.info("%s %s in %.1fs, tools %d, charts %d, %d chars: %s", "Replied" if ok else "Error reply", msg.chat_id, - time.monotonic() - started, len(used), len(photos), len(reply), _short(reply, 30)) + time.monotonic() - started, len(used), len(photos), len(reply), reply, extra=QUOTES) used_air = any(tc["function"]["name"] == "air_quality" for m in working[new_from:] for tc in m.get("tool_calls") or []) air = self.by_name.get("AirGradient") markup = templates.keyboard(self.irrigation is not None) if self._keyboard_stale(msg) else None refresh = ok and read.report and not photos and 0 < len(reply) <= TG_LIMIT and not markup # the report, in one message, to be refreshed in place row =periods.keyboard(intent.period_days(text, now)) if len(photos) == 1 else None # period buttons under a single chart - sent_photo = [] - if row and ok and not reply.strip(): # a chart and nothing else: keep it for the period buttons - self.images.put(self.images.key(text, intent.period_days(text, now), now), photos[0], titles[0]) + sent_photo, drawn = [], list(photos) + kept = self.images.key(text, intent.period_days(text, now), now) if row and ok and not reply.strip() else None + if kept and not ready: # a chart and nothing else: keep it for the period buttons (and, its words being its key, for the next to ask them) + self.images.put(kept, photos[0], titles[0], caption if kept == self.images.canonical(text, now) else None) try: - if redraw is not None and row and await self._redraw(redraw, reply, photos[0], titles[0], row, text): + began = time.monotonic() + if redraw is not None and row and await self._redraw(redraw, reply, photos[0], titles[0], row, text, kept): reply, photos, row = "", [], None carried_keyboard = await deliver(msg, reply, photos, link=air.link if air and used_air else None, markup=markup or (report_menu.keyboard() if refresh else None), titles=titles, period_row=row, sent_photo=sent_photo, bold_headings=ok and bool(read.report or read.lookup in ("pollen", "forecast"))) and bool(markup) # the refresh button is not the chat's keyboard + if ready: + log.info("Chart ms: drawn %.0fs ago, send %.0f%s", self.images.age(ready), (time.monotonic() - began) * 1000, + " (Telegram's copy)" if isinstance(drawn[0], str) else "") + elif drawn: + log.info("Chart ms: %s", chart_times(turn.times, made, drawn, time.monotonic() - began)) if sent_photo: self.charted.remember(msg.chat_id, sent_photo[0].message_id, text) + self.images.sent(kept, sent_photo[0]) + if kept and read.fast and read.chart_in_code: # a chart with period buttons, made in code: its other periods are uploaded ahead + self.preload_soon(context.bot, text) if self.state and ok and turn.forecast_shown: # it was sent: a later revision of these days is worth telling this chat self.state.record_forecast(msg.chat_id, turn.forecast_shown) if carried_keyboard: diff --git a/lib/charts.py b/lib/charts.py index 1098c97..934faa0 100644 --- a/lib/charts.py +++ b/lib/charts.py @@ -6,13 +6,14 @@ - one panel of lines is drawn large, with the records marked as pills, an end dot and (for wind) the compass beside it, at exactly 1280x720, the size Telegram displays photos at, so nothing is rescaled; - several panels are stacked, the figure growing a little taller with each one past two (four panels: 1280x1160). -Rain (bars) sits behind the lines of the panel it belongs to, on its own right-hand axis. +Rain (bars) sits behind the lines of the panel it belongs to, on a scale of its own read on the right. """ import glob import io import math import threading +import time from dataclasses import dataclass from datetime import tzinfo from statistics import fmean @@ -24,10 +25,13 @@ import matplotlib.pyplot as plt # noqa: E402 import numpy as np # noqa: E402 from matplotlib import font_manager # noqa: E402 +from matplotlib.collections import PatchCollection, PolyCollection # noqa: E402 from matplotlib.colors import to_rgb # noqa: E402 from matplotlib.lines import Line2D # noqa: E402 -from matplotlib.patches import Polygon # noqa: E402 +from matplotlib.patches import Polygon, Rectangle, Wedge # noqa: E402 from matplotlib.ticker import FixedLocator, FuncFormatter, MaxNLocator # noqa: E402 +from matplotlib.transforms import ScaledTranslation # noqa: E402 +from PIL import Image # noqa: E402 from .specs import Chart, Compass, Line, Panel # noqa: E402 from .timeutil import to_local # noqa: E402 @@ -150,8 +154,10 @@ def _box(colour: str, pad: float = 0.25, rounding: float = 0.6) -> dict: def _dot(ax, x: float, y: float, colour: str, size: float = 12, edge: float = 0.8, z: int = 4): - """A small dot with a white rim, marking a point on a line.""" - ax.scatter([x], [y], s=size, color=colour, edgecolors="white", linewidths=edge, zorder=z) + """A small dot with a white rim, marking a point on a line (`size` is its area in pt², as scatter takes it). It is a one-point + line's marker: drawn just as a one-point scatter is, at a fraction of the cost.""" + ax.add_line(Line2D([x], [y], marker="o", linestyle="", markersize=math.sqrt(size), color=colour, markeredgecolor="white", + markeredgewidth=edge, zorder=z)) def _to_dt(tz: tzinfo): @@ -182,12 +188,12 @@ def _style_axis(ax, size: float = 7.5, pad: float = 5, grid: float = 0.8, below: ax.set_axisbelow(True) -def _shade_zones(ax, zones, ybottom: float, ytop: float, alpha: float): +def _shade_zones(ax, zones, ybottom: float, ytop: float, alpha: float, zorder: float = 0): """Faint rating zones behind a line: good up to z1, poor up to z2, very poor above.""" z1, z2 = zones for lo, hi, colour in ((ybottom, z1, ZONE_COLOURS[0]), (z1, z2, ZONE_COLOURS[1]), (z2, ytop, ZONE_COLOURS[2])): if hi > ybottom and lo < ytop: - ax.axhspan(max(lo, ybottom), min(hi, ytop), color=colour, alpha=alpha, linewidth=0, zorder=0) + ax.axhspan(max(lo, ybottom), min(hi, ytop), color=colour, alpha=alpha, linewidth=0, zorder=zorder) def _pad_limits(ax, lo: float, hi: float, top: float = 0.2, bottom: float = 0.16, floor: float | None = None): @@ -219,42 +225,38 @@ def _draw_lines(ax, lines: list[Line], tz: tzinfo, look: Look, first: int = 0, r return drawn -def _draw_bars(ax, bx, ys, width: float, axis_top: float, zorder: int, alpha: float = 0.5): +def _columns(ax, x, heights, width: float, bottom=0.0, **style): + """Bars from each x to x + width, as one collection of rectangles: what ax.bar draws, without a patch (and its bookkeeping) + for every bar. The axes' limits are the caller's to set.""" + x = np.asarray(x, dtype=float) + low = np.broadcast_to(np.asarray(bottom, dtype=float), x.shape) + high = low + np.asarray(heights, dtype=float) + corners = np.stack([np.column_stack(xy) for xy in ((x, low), (x + width, low), (x + width, high), (x, high))], axis=1) + ax.add_collection(PolyCollection(corners, edgecolors="none", linewidths=0, **style), autolim=False) + + +def _draw_bars(ax, bx, ys, width: float, axis_top: float, zorder: int, alpha: float = 0.5, **kw): """Rain bars: a pale body (RAIN blended with white by `alpha`, so solid and nothing behind it shows through) with a - brighter cap, so each reads as a little column.""" + brighter cap, so each reads as a little column. `kw` goes to both (for rain behind a line, the transform).""" ys = np.asarray(ys, dtype=float) - ax.bar(bx, ys, width=width, align="edge", color=_mix(RAIN, 1 - alpha), linewidth=0, zorder=zorder) + _columns(ax, bx, ys, width, facecolors=_mix(RAIN, 1 - alpha), zorder=zorder, **kw) cap = axis_top * 0.014 - ax.bar(bx, np.minimum(cap, ys), bottom=np.maximum(ys - cap, 0), width=width, align="edge", color=RAIN, alpha=0.95, - linewidth=0, zorder=zorder) + _columns(ax, bx, np.minimum(cap, ys), width, np.maximum(ys - cap, 0), facecolors=RAIN, alpha=0.95, zorder=zorder, **kw) def _bars_behind(ax, bars, tz: tzinfo) -> float: - """Bars (rain) on their own right-hand axis, behind the panel's lines and never taller than BARS_SHARE of it. Returns - where the last bar ends (matplotlib date number).""" + """Bars (rain) behind the panel's lines and over its grid, on a scale of their own read on the right: the tallest reaches + BARS_SHARE of the panel's height. They are drawn in the panel itself (x on its time axis, y as a share of its height), and + the scale's two labels sit where a right-hand axis would put them. Returns where the last bar ends (matplotlib date number).""" bx = _nums(tz, bars.x) - ax2 = ax.twinx() - weight = ax.yaxis.get_gridlines()[0].get_linewidth() if ax.yaxis.get_gridlines() else 0.8 - ax.grid(False) # the grid is drawn by an axes of its own under the rain (sharing this one's scale): grid, rain, lines - under = ax.figure.add_axes(ax.get_position(), sharex=ax, sharey=ax, facecolor=ax.get_facecolor()) - for side in ("top", "right", "left", "bottom"): - under.spines[side].set_visible(False) - under.tick_params(axis="both", length=0, labelleft=False, labelbottom=False) - under.grid(axis="y", color=GRID, linewidth=weight) - under.set_axisbelow(True) - under.set_zorder(ax2.get_zorder() - 1) top = max([*bars.y, 1.0]) - _draw_bars(ax2, bx, bars.y, bars.width / 86400 * 0.85, top / BARS_SHARE, 1) - ax2.set_ylim(0, top / BARS_SHARE) - ax2.axhline(top, color=RAIN, linewidth=0.7, linestyle=(0, (1, 2)), alpha=0.8, zorder=1) # the top of the rain scale, on its own - ax2.yaxis.set_major_locator(FixedLocator([0, top])) - ax2.yaxis.set_major_formatter(FuncFormatter(lambda v, _: f"{v:g} {bars.unit}" if v else "0")) - for side in ("top", "right", "left", "bottom"): - ax2.spines[side].set_visible(False) - ax2.tick_params(axis="y", length=0, labelsize=7, labelcolor=RAIN_TEXT, pad=4) - ax.set_zorder(ax2.get_zorder() + 1) # the lines above the bars - ax.patch.set_visible(False) - ax2.patch.set_visible(False) # so the grid beneath shows between the bars (set last: the line above re-shows it) + _draw_bars(ax, bx, np.asarray(bars.y, dtype=float) * (BARS_SHARE / top), bars.width / 86400 * 0.85, 1.0, 1, + transform=ax.get_xaxis_transform()) + ax.add_line(Line2D([0, 1], [BARS_SHARE, BARS_SHARE], transform=ax.transAxes, color=RAIN, linewidth=0.7, linestyle=(0, (1, 2)), + alpha=0.8, zorder=1)) # the top of the rain scale, on its own + beside = ax.transAxes + ScaledTranslation(4 / 72, 0, ax.figure.dpi_scale_trans) # 4 pt right of the panel, as tick labels sit + for y, label in ((0, "0"), (BARS_SHARE, f"{top:g} {bars.unit}")): + ax.text(1, y, label, transform=beside, ha="left", va="center_baseline", fontsize=7, color=RAIN_TEXT) return _end_of(bx, bars.width, 0.0) @@ -361,17 +363,32 @@ def _pills(ax, lines: list[Line], drawn: list[tuple], tz: tzinfo, x0: float, x1: bbox=_box(colour, *look.pill_box), zorder=look.pill_z + 1) +def _rose_radius(outer: np.ndarray) -> float: + """The rose's radius for wedges ending at `outer` (16 compass points x 3 speed steps, out from the centre): the one matplotlib + gives bars that long, 5% beyond the longest wedge, or right at its end when that wedge ends in an empty step (the base of a + bar is a sticky edge). Worked out as matplotlib does, so the rose is the size it always was.""" + longest = float(outer[:, -1].max()) + if not longest: + return 1.0 # nothing to draw + ends = outer[:, :-1][outer[:, :-1] >= longest - 1e-5 * longest] # a step's base at the end of a longest wedge: nothing beyond it + return float(ends.min()) if ends.size else longest + longest * 0.05 + + def _render_rose(fig, compass: Compass): """The whole period as a wind rose (N up, clockwise): each of the 16 wedges is the share of readings from that - way, stacked by wind speed. It sits beside the wind speed line.""" + way, stacked by wind speed. It sits beside the wind speed line. The wedges are one collection in the polar axes' own + units (the centre at 0.5, 0.5 and the edge at radius 0.5): 48 polar bars cost several times as much to make and draw.""" counts = np.asarray(compass.rose, dtype=float) # 16 compass points x 3 speed steps ax = fig.add_axes([0.685, 0.12, 0.27, 0.62], projection="polar") - theta = np.deg2rad(np.arange(16) * 22.5) - percent = counts / max(counts.sum(), 1) * 100 - bottom = np.zeros(16) - for i, colour in enumerate(WIND_STEPS): - ax.bar(theta, percent[:, i], width=np.deg2rad(20), bottom=bottom, color=colour, edgecolor=BG, linewidth=0.4) - bottom += percent[:, i] + outer = np.cumsum(counts / max(counts.sum(), 1) * 100, axis=1) # where each step ends, out from the centre + radius = _rose_radius(outer) + ax.set_ylim(0, radius) # the grid's rings follow it + half = 0.5 / radius + steps = [(k, i) for i in range(3) for k in range(16) if outer[k, i]] # step by step, each over the one before; nothing at the centre + wedges = [Wedge((0.5, 0.5), outer[k, i] * half, 80 - 22.5 * k, 100 - 22.5 * k, width=(outer[k, i] - (outer[k, i - 1] if i else 0)) * half) + for k, i in steps] # 20° wide around each compass point, clockwise from N + ax.add_collection(PatchCollection(wedges, facecolors=[WIND_STEPS[i] for _, i in steps], edgecolors=BG, linewidths=0.4, + transform=ax.transAxes), autolim=False) ax.set_theta_zero_location("N") ax.set_theta_direction(-1) ax.set_xticks(np.deg2rad([0, 90, 180, 270])) @@ -484,7 +501,7 @@ def _draw_panel(ax, p: Panel, tz: tzinfo, first: int, x0: float, x1: float, look bx = _nums(tz, s.x) base = np.zeros(len(bx)) for column, colour in zip((s.good, s.poor, s.very_poor), ZONE_COLOURS): - ax.bar(bx, column, bottom=base, width=s.width / 86400 * 0.85, align="edge", color=colour, linewidth=0, zorder=3) + _columns(ax, bx, column, s.width / 86400 * 0.85, base, facecolors=colour, zorder=3) base += np.asarray(column, dtype=float) ax.set_ylim(0, 100) ax.legend([Line2D([], [], marker="s", linestyle="", markersize=5, color=c) for c in ZONE_COLOURS], @@ -511,8 +528,8 @@ def _draw_panel(ax, p: Panel, tz: tzinfo, first: int, x0: float, x1: float, look floor=0) ybottom, ytop = ax.get_ylim() drawn = _draw_lines(ax, lines, tz, look, first, p.reading) - if p.zones: # a rated reading: its good / poor / very poor zones behind the line - _shade_zones(ax, p.zones, ybottom, ytop, 0.07) + if p.zones: # a rated reading: its good / poor / very poor zones behind the line (and over any rain behind it, as always) + _shade_zones(ax, p.zones, ybottom, ytop, 0.07, 1.5 if p.bars else 0) ax.set_ylim(ybottom, ytop) if marked and p.peaks == "aside": _mark_highs(ax, [lines[i] for i in marked], [drawn[i] for i in marked], x0, x1, look) @@ -528,9 +545,10 @@ def _draw_panel(ax, p: Panel, tz: tzinfo, first: int, x0: float, x1: float, look return len(lines), x1, entries -def _render(chart: Chart, tz: tzinfo) -> bytes: +def _render(chart: Chart, tz: tzinfo, times: dict | None = None) -> bytes: """The panels top to bottom on one time axis, so the rain (or another reading) lines up with what the others were doing. A chart of one panel of lines is the same thing drawn BIG (and with its wind rose beside it, when it has one).""" + started = time.perf_counter() panels, n = chart.panels, len(chart.panels) look = _look_for(chart) height = H_IN if n <= 2 else H_IN + PANEL_IN * (n - 2) @@ -541,21 +559,18 @@ def _render(chart: Chart, tz: tzinfo) -> bytes: body = height - HEAD_IN - FOOT_IN gap = 0.27 if n > 2 else 0.22 each = (body - gap * (n - 1)) / n - xs = _nums(tz, [t for p in panels for t in p.xs]) + xs = _nums(tz, {t for p in panels for t in p.xs}) # each time once: only the range is wanted x0, x1 = float(xs.min()), float(xs.max()) axes, used = [], 0 for i, p in enumerate(panels): bottom = (FOOT_IN + (n - 1 - i) * (each + gap)) / height rect = [AX_RECT[0], bottom, width, each / height] - if look.card: - card = fig.add_axes(rect, facecolor=CARD, zorder=-1) # the faint tint behind the panel - card.set_xticks([]) - card.set_yticks([]) - for side in card.spines.values(): - side.set_visible(False) + if look.card: # the faint tint behind the panel: a rectangle on the figure (an axes of its own costs far more) + fig.add_artist(Rectangle(rect[:2], rect[2], rect[3], transform=fig.transFigure, facecolor=CARD, edgecolor="none", + linewidth=0, zorder=-1)) ax = fig.add_axes(rect, facecolor="none" if look.card else BG, sharex=axes[0] if axes else None) axes.append(ax) - _style_axis(ax, grid=0 if p.bars and p.lines else 0.8) # with rain behind, the grid is drawn under it (_bars_behind) + _style_axis(ax) count, x1, entries = _draw_panel(ax, p, tz, used, x0, x1, look) used += count if len(entries) >= look.key_from: # the colour key: beside the headline for a big chart, above the panel in a stack @@ -570,9 +585,9 @@ def _render(chart: Chart, tz: tzinfo) -> bytes: ax.yaxis.set_major_formatter(FuncFormatter(lambda v, _, u=tick_unit: f"{v:g}{u}")) if not look.in_headline: title = f"{p.label} ({p.unit})" if p.unit and not tick_unit else p.label - ax.set_title(title, loc="left", fontsize=8.5, fontweight=TITLE_WEIGHT, color=TEXT, pad=4) + ax.set_title(title, loc="left", fontsize=8.5, fontweight=TITLE_WEIGHT, color=TEXT, pad=4, y=1.0) # y: placed, not searched for if i < n - 1: - plt.setp(ax.get_xticklabels(), visible=False) + ax.xaxis.set_visible(False) # the time is read off the bottom panel (these ticks have no length and no grid) margin = (x1 - x0) * 0.015 # room for the end dots axes[0].set_xlim(x0 - margin, x1 + margin) _time_axis(axes[-1], x1 - x0, panels[-1].bars if not panels[-1].lines else None, tz) @@ -583,13 +598,61 @@ def _render(chart: Chart, tz: tzinfo) -> bytes: except Exception: plt.close(fig) raise - return _png(fig) - - -def _png(fig) -> bytes: + return _png(fig, times, started) + + +def _palette(rgb: np.ndarray, counts: np.ndarray, size: int = 256) -> tuple[list[int], np.ndarray]: + """Up to `size` of these colours (rows of `rgb`, each with its count of pixels) for a palette, and the nearest of them for every + colour. It starts from the commonest; each next one is the colour the palette so far draws worst (its pixels times its squared + distance to the nearest colour taken), so the colours that cover an area are kept exactly and the in-between shades of + anti-aliased edges come out within a few levels. With `size` colours or fewer, every one is kept. Whole numbers throughout, so + the picks are exact (ties go to the first colour).""" + red, green, blue = (np.ascontiguousarray(rgb[:, i], dtype=np.int64) for i in range(3)) + weight = np.asarray(counts, dtype=np.int64) + + def distance_to(k: int) -> np.ndarray: # every colour's squared distance to colour k + distance = (red - red[k]) ** 2 + distance += (green - green[k]) ** 2 + distance += (blue - blue[k]) ** 2 + return distance + taken = [int(np.argmax(weight))] + error = distance_to(taken[0]) # each colour's squared distance to its nearest taken one + score = weight * error # how badly the palette so far draws its pixels + nearest = np.zeros(len(rgb), np.uint8) + while len(taken) < min(size, len(rgb)): + pick = int(np.argmax(score)) + if not error[pick]: + break + distance = distance_to(pick) + closer = np.flatnonzero(distance < error) # only these colours' error and score change + error[closer] = distance[closer] + score[closer] = weight[closer] * distance[closer] + nearest[closer] = len(taken) + taken.append(pick) + return taken, nearest + + +def _png(fig, times: dict | None = None, started: float = 0.0) -> bytes: + """The figure as a PNG of at most 256 colours (see _palette): well under half the size of a full-colour one, and quicker to + write. A chart has a few thousand colours, nearly all of them the in-between shades on the edges of lines and letters. + times: the seconds since `started` (the clock when the figure was begun) are added to its "draw", and the PNG's to its "png".""" try: + fig.canvas.draw() + drawn = time.perf_counter() + width, height = fig.canvas.get_width_height(physical=True) + pixels = np.frombuffer(fig.canvas.buffer_rgba(), "> 8 & 255, colours >> 16], axis=1) + taken, nearest = _palette(rgb, counts) + lookup = np.empty(1 << 24, np.uint8) # each colour's place in the palette + lookup[colours] = nearest + image = Image.frombuffer("P", (width, height), lookup[pixels], "raw", "P", 0, 1) + image.putpalette(rgb[taken].astype(np.uint8).tobytes()) buf = io.BytesIO() - fig.savefig(buf, format="png", dpi=DPI, facecolor=BG) + image.save(buf, format="png", dpi=(DPI, DPI)) + if times is not None: + times["draw"] = times.get("draw", 0.0) + drawn - started + times["png"] = times.get("png", 0.0) + time.perf_counter() - drawn return buf.getvalue() finally: plt.close(fig) @@ -598,7 +661,8 @@ def _png(fig) -> bytes: _DRAWING = threading.Lock() # pyplot keeps global state (current figure, rcParams): one chart is drawn at a time -def render(chart: Chart, tz: tzinfo) -> bytes: - """PNG bytes for one chart (1280x720; a stack of more than two panels is taller). Safe to call from several threads.""" +def render(chart: Chart, tz: tzinfo, times: dict | None = None) -> bytes: + """PNG bytes for one chart (1280x720; a stack of more than two panels is taller). Safe to call from several threads. + times (a Turn's): the seconds spent drawing the figure and writing the PNG are added to its "draw" and "png".""" with _DRAWING: - return _render(chart, tz) + return _render(chart, tz, times) diff --git a/lib/config.py b/lib/config.py index a8784ec..d048e5c 100644 --- a/lib/config.py +++ b/lib/config.py @@ -65,6 +65,29 @@ def _ids(value: str) -> frozenset[int]: return frozenset(found) +def _whole(value: str, default: int) -> int: + """"100" -> 100, "0" -> 0; anything but a whole number of 0 or more -> the default.""" + return int(value) if value.isdecimal() else default + + +def _readings_in_a_row(value: str, default: int) -> int: + """"3" -> 3; "off", "0" or "1" -> 1 (no rule); anything unreadable -> the default.""" + if value.lower() in ("off", "no", "false", "none"): + return 1 + return max(1, int(value)) if value.isdecimal() else default + + +def _chat_id(value: str, name: str) -> int | None: + """A Telegram chat id (negative for a group or channel); None, with a warning, when it is not a whole number.""" + if not value: + return None + try: + return int(value) + except ValueError: + log.warning("%s: ignoring %r (not a chat id, a whole number)", name, value) + return None + + def _number(value: str) -> float | None: try: return float(value) @@ -105,6 +128,9 @@ class Config: admin_only: bool = True # ADMIN_ONLY (on by default): the bot answers nobody but the admins of the groups it is in admin_chat_ids: frozenset[int] = frozenset() # ADMIN_CHAT_IDS: more ids allowed besides those admins (a user id, or a group id for its anonymous admins) alert_cooldown_minutes: int = 30 # ALERT_COOLDOWN_MINUTES: the rain and temperature-crossing alerts: dry this long before "the rain has stopped", and this long between alerts (5 to 150) + telegram_preload_chat: int | None = None # TELEGRAM_PRELOAD_CHAT: a chat the other periods of each chart are silently uploaded to ahead of the press (none: off) + antispike_require_consecutive: int = 2 # ANTISPIKE_REQUIRE_CONSECUTIVE: readings in a row a rise of the particles needs to be the air, not a glitch of the sensor (1: off) + log_width: int = 71 # LOG_WIDTH: the most characters of a long log line (a request, a chat message), cut with an ellipsis (…); 0 cuts none @property def ecowitt(self) -> bool: @@ -158,6 +184,9 @@ def from_env(cls) -> "Config": admin_only=_flag(env("ADMIN_ONLY"), True), admin_chat_ids=_ids(env("ADMIN_CHAT_IDS")), alert_cooldown_minutes=int(min(150, max(5, _number(env("ALERT_COOLDOWN_MINUTES")) or 30))), + log_width=_whole(env("LOG_WIDTH"), 71), + antispike_require_consecutive=_readings_in_a_row(env("ANTISPIKE_REQUIRE_CONSECUTIVE"), 2), + telegram_preload_chat=_chat_id(env("TELEGRAM_PRELOAD_CHAT"), "TELEGRAM_PRELOAD_CHAT"), state_path=data / "bot_state.json", cache_path=data / "ecowitt_cache.sqlite", air_cache_path=data / "airgradient_cache.sqlite", diff --git a/lib/ecowitt/api.py b/lib/ecowitt/api.py index 11f1e87..8c9e6aa 100644 --- a/lib/ecowitt/api.py +++ b/lib/ecowitt/api.py @@ -107,7 +107,7 @@ async def acquire(self, urgent: bool = False): except asyncio.CancelledError: if waiter.done() and not waiter.cancelled(): # the turn was already handed to us: pass it on self.release() - else: + elif waiter in self._queues[urgent]: # (a release since the cancel has already dropped it) self._queues[urgent].remove(waiter) raise @@ -172,8 +172,8 @@ async def devices(self) -> list[dict]: async def realtime(self, mac: str, groups: str, urgent: bool = False) -> dict: return fix_units(await self._get("real_time", urgent=urgent, mac=mac, call_back=groups, **UNIT_IDS)) - async def history(self, mac: str, cycle: str, start: datetime, end: datetime, groups: str) -> dict: - """{group: {field: {"unit", "list": {epoch: value}}}}; empty when there is no data.""" - data = await self._get("history", mac=mac, call_back=groups, cycle_type=cycle, + async def history(self, mac: str, cycle: str, start: datetime, end: datetime, groups: str, urgent: bool = False) -> dict: + """{group: {field: {"unit", "list": {epoch: value}}}}; empty when there is no data. urgent: as for _get.""" + data = await self._get("history", urgent=urgent, mac=mac, call_back=groups, cycle_type=cycle, start_date=start.strftime(FMT), end_date=end.strftime(FMT), **UNIT_IDS) return fix_units(data) if isinstance(data, dict) else {} diff --git a/lib/ecowitt/archive.py b/lib/ecowitt/archive.py index 8602968..88fc674 100644 --- a/lib/ecowitt/archive.py +++ b/lib/ecowitt/archive.py @@ -76,7 +76,7 @@ async def run_once(self) -> tuple[int, int]: # each range takes the pause after it plus about a second for the request itself seconds = len(todo) * (max(self.pace, MIN_GAP_SECONDS) + 1) if seconds >= 60: # a short run only gets the line when it finishes - log.info("Ecowitt archive: %d req to fetch (about %s)", len(todo), + log.info("Ecowitt: %d req to fetch (about %s)", len(todo), f"{seconds:.0f} s" if seconds < 90 else f"{seconds / 60:.0f} min") fetched = failed = 0 for i, (cycle, start, end) in enumerate(todo, 1): @@ -87,7 +87,7 @@ async def run_once(self) -> tuple[int, int]: else: fetched += 1 if i % 20 == 0 and i < len(todo): - log.info("Ecowitt archive: %d of %d req done", i, len(todo)) + log.info("Ecowitt: %d of %d req done", i, len(todo)) await asyncio.sleep(self.pace) return fetched, failed @@ -114,7 +114,7 @@ def held(self) -> str: """Days of history held per resolution, e.g. "88/363/728/1456 days (5min/30min/4h/1d)" """ cycles = ("5min", "30min", "4hour", "1day") days = "/".join(str(self.station.cache.days_held(self.station.mac, c, self.groups)) for c in cycles) - return f"{days} days (5min/30min/4h/1d)" + return f"{days} days (5m/30m/4h/1d)" def _seconds_until_next_run(self) -> float: now = datetime.now(self.station.tz) @@ -129,7 +129,7 @@ async def loop(self): try: started = datetime.now() fetched, failed = await self.run_once() - log.info("Ecowitt archive: %d req, %sheld %s, %.0fs", fetched, f"{failed} failed, " if failed else "", + log.info("Ecowitt: %d req, %sheld %s, %.0fs", fetched, f"{failed} failed, " if failed else "", self.held(), (datetime.now() - started).total_seconds()) except asyncio.CancelledError: raise diff --git a/lib/ecowitt/fetch.py b/lib/ecowitt/fetch.py index 2753841..648035c 100644 --- a/lib/ecowitt/fetch.py +++ b/lib/ecowitt/fetch.py @@ -2,12 +2,14 @@ or from Ecowitt, within Ecowitt's per-request limits.""" import asyncio +import functools import logging +import time from datetime import datetime, timedelta, tzinfo, UTC from ..timeutil import local_epoch, to_local from .api import EcowittError, MAX_SPAN, RETENTION -from .store import HistoryCache, HotStore, merge as merge_intervals +from .store import HistoryCache, HotStore, merge as merge_intervals, subtract log = logging.getLogger(__name__) @@ -41,17 +43,31 @@ def merge_data(into: dict, new: dict, window: tuple[int, int] | None = None): REJECTIONS_STOP = 2 # requests Ecowitt refuses in one job before the rest are not sent +def timed(method): + """A Fetcher method whose time (waits included) is added to the fetcher's `seconds`.""" + @functools.wraps(method) + async def run(self, *args, **kwargs): + started = time.perf_counter() + try: + return await method(self, *args, **kwargs) + finally: + self.seconds += time.perf_counter() - started + return run + + class Fetcher: """History for one job (a question, a refresh, an alert check): from the disk cache where possible, recent readings from memory if fresh, the rest from Ecowitt.""" - def __init__(self, api, cache: HistoryCache, hot: HotStore, mac: str, groups: list[str], tz: tzinfo): - self.api, self.cache, self.hot, self.tz = api, cache, hot, tz + def __init__(self, api, cache: HistoryCache, hot: HotStore, mac: str, groups: list[str], tz: tzinfo, urgent: bool = False): + """urgent: someone is waiting for the answer, so its requests go before the background ones queued for Ecowitt.""" + self.api, self.cache, self.hot, self.tz, self.urgent = api, cache, hot, tz, urgent self.mac = mac.strip().upper() self.groups = [g.split(".")[0].strip() for g in groups if g.strip()] # plain names, never "outdoor.temp" self.ranges = self.from_cache = self.from_memory = self.calls = 0 self.errors: list[str] = [] self.rejected = 0 # requests Ecowitt itself refused (not network trouble or a busy server) + self.seconds = 0.0 # time spent reading (the cache, memory, Ecowitt), for the reply's timing line def epoch(self, local: datetime) -> int: return local_epoch(local, self.tz) @@ -59,35 +75,50 @@ def epoch(self, local: datetime) -> int: def local(self, epoch: int) -> datetime: return to_local(epoch, self.tz).replace(tzinfo=None) + @timed async def get(self, cycle: str, start: datetime, end: datetime, refresh: bool = False, load: bool = True) -> dict: """Readings for [start, end] (local time). refresh=True ignores the in-memory copy (used by the keep-warm refresh). load=False only makes sure the range is cached and in memory, and - skips building the result nobody will read (the refresh and the archive).""" + skips building the result nobody will read (the refresh and the archive). + When the cache and memory hold it all (the usual case) it is read at once, even while a refresh is fetching; otherwise + it waits for any fetch of this cycle in flight (which may bring it), so the same thing isn't asked of Ecowitt twice.""" self.ranges += 1 s, e = self.epoch(start), self.epoch(end) + if not refresh and (held := await self._read(cycle, s, e, load, fetch=False)) is not None: + return held async with self.hot.lock(self.mac, cycle): - missing = {g: await asyncio.to_thread(self.cache.missing, self.mac, cycle, g, s, e) for g in self.groups} - gaps = merge_intervals([iv for ivs in missing.values() for iv in ivs]) - if not gaps: - self.from_cache += 1 - fresh: dict = {} - for gap_start, gap_end in gaps: - need = [g for g in self.groups if any(a <= gap_end and b >= gap_start for a, b in missing[g])] - hot = {} if refresh else {g: self.hot.get(self.mac, cycle, g, gap_start, gap_end) for g in need} - if hot and all(v is not None for v in hot.values()): - self.from_memory += 1 + return await self._read(cycle, s, e, load, fetch=True, refresh=refresh) + + async def _read(self, cycle: str, s: int, e: int, load: bool, fetch: bool, refresh: bool = False) -> dict | None: + """[s, e] (epoch seconds) from the cache, its newest readings from memory, and (fetch) what neither holds from Ecowitt, which + is then cached and kept in memory. None, with nothing counted, when fetch is False and the cache and memory lack some of it.""" + (missing,) = await self._missing(cycle, [(s, e)]) + gaps = merge_intervals([iv for ivs in missing.values() for iv in ivs]) + fresh: dict = {} + from_memory = 0 + for gap_start, gap_end in gaps: + need = [g for g in self.groups if any(a <= gap_end and b >= gap_start for a, b in missing[g])] + hot = {} if refresh else {g: self.hot.get(self.mac, cycle, g, gap_start, gap_end) for g in need} + if hot and all(v is not None for v in hot.values()): + from_memory += 1 + if load: merge_data(fresh, hot, (gap_start, gap_end)) - continue - data = await self._fetch(cycle, self.local(gap_start), self.local(gap_end), need) - if data is None: - continue # failed: not recorded as fetched, so it's retried next time - await asyncio.to_thread(self.cache.store, self.mac, cycle, need, data, gap_start, gap_end) - for g in need: - self.hot.put(self.mac, cycle, g, gap_start, gap_end, data.get(g) or {}) + continue + if not fetch: + return None + data = await self._fetch(cycle, self.local(gap_start), self.local(gap_end), need) + if data is None: + continue # failed: not recorded as fetched, so it's retried next time + await asyncio.to_thread(self.cache.store, self.mac, cycle, need, data, gap_start, gap_end) + for g in need: + self.hot.put(self.mac, cycle, g, gap_start, gap_end, data.get(g) or {}) + if load: merge_data(fresh, data) - if not load: - return {} - result = await asyncio.to_thread(self.cache.load, self.mac, cycle, self.groups, s, e) + self.from_cache += not gaps + self.from_memory += from_memory + if not load: + return {} + result = await asyncio.to_thread(self.cache.load, self.mac, cycle, self.groups, s, e) merge_data(result, fresh, (s, e)) # recent readings aren't on disk; use the fresh ones return result @@ -97,7 +128,7 @@ async def _fetch(self, cycle: str, start: datetime, end: datetime, groups: list[ return None self.calls += 1 try: - return await self.api.history(self.mac, cycle, start, end, ",".join(groups)) + return await self.api.history(self.mac, cycle, start, end, ",".join(groups), urgent=self.urgent) except EcowittError as e: label = f"{cycle} {start:%d %b %Y} - {end:%d %b %Y}" log.warning("History request failed (%s): %s", label, e) @@ -105,12 +136,25 @@ async def _fetch(self, cycle: str, start: datetime, end: datetime, groups: list[ self.rejected += not e.transient return None + @timed + async def read(self, fn, *args): + """fn(*args), a read of the cache's own (slots, slot ranges), in a thread.""" + return await asyncio.to_thread(fn, *args) + + async def _missing(self, cycle: str, windows: list[tuple[int, int]]) -> list[dict[str, list[tuple[int, int]]]]: + """For each window (epoch seconds), the parts of it the cache lacks, by group: one read of the coverage in one thread hop + (the hop costs more than the query, and far more while a chart is being drawn).""" + held = await asyncio.to_thread(self.cache.coverage_of, self.mac, cycle, self.groups) if self.groups else {} + return [{g: subtract(w, held[g]) for g in self.groups} for w in windows] + async def covered(self, cycle: str, start: int, end: int) -> bool: """True if the cache holds every group for [start, end] (epoch seconds).""" - for g in self.groups: - if await asyncio.to_thread(self.cache.missing, self.mac, cycle, g, start, end): - return False - return bool(self.groups) + return (await self.covered_each(cycle, [(start, end)]))[0] + + @timed + async def covered_each(self, cycle: str, windows: list[tuple[int, int]]) -> list[bool]: + """covered() for each of these windows, from one read of the coverage.""" + return [bool(self.groups) and not any(missing.values()) for missing in await self._missing(cycle, windows)] async def finer_cycle(self, ts: int, current: str, window_end: int, now_utc: datetime) -> str | None: """Finest cycle finer than `current` available for this window: still kept by Ecowitt, diff --git a/lib/ecowitt/query.py b/lib/ecowitt/query.py index 9b76ad8..16cd1d4 100644 --- a/lib/ecowitt/query.py +++ b/lib/ecowitt/query.py @@ -13,7 +13,7 @@ import asyncio import json import logging -from datetime import datetime, time, timedelta, UTC +from datetime import date, datetime, time, timedelta, UTC from statistics import fmean from ..captions import AVERAGE_CHART_HINT, CHART_HINT, STACK_CHART_HINT, DIRECTION_CHART_HINT, wants_chart @@ -51,6 +51,17 @@ DERIVED = ("feels_like", "app_temp", "app_tempin", "dew_point", "vpd") +def consecutive(days: list[date]) -> list[list[date]]: + """Sorted days in runs of consecutive ones: [1, 2, 3, 5, 6] -> [[1, 2, 3], [5, 6]].""" + runs: list[list[date]] = [] + for day in days: + if runs and day - runs[-1][-1] == timedelta(days=1): + runs[-1].append(day) + else: + runs.append([day]) + return runs + + def stack_names(args: dict, turn) -> list[str]: """The readings to put side by side (from the person's words, else the model's chart_fields), two or more.""" raw = turn.chart_fields or args.get("chart_fields") or [] @@ -77,8 +88,15 @@ def __init__(self, fetcher: Fetcher, args: dict, turn): self.rain_bars = Bars("Rain", "mm", [], [], 86400, "day") # for a stacked chart with rain self.rain_from_daily = False # some of its days come from daily records (older than the 30-minute readings) self.compass: Compass | None = None # the wind rose (counts per compass point by speed), drawn beside the wind chart + self.wind: Fetcher | None = None # the wind rose's own (5-minute readings), made when wind direction is counted async def run(self) -> str: + try: + return await self._run() + finally: # for the reply's timing line + self.turn.times["data"] = self.turn.times.get("data", 0.0) + self.f.seconds + (self.wind.seconds if self.wind else 0.0) + + async def _run(self) -> str: today = self.now.date() try: start = datetime.fromisoformat(str(self.args.get("start_date") or today).replace("T", " ")) @@ -130,8 +148,8 @@ async def _fine_ranges(self): pts = series["pts"] if derives_range(field) and pts and not any(r["cycle"] != "30min" or "low" in r or "high" in r for r in pts.values()): wanted[key] = (group, field, {t: r["value"][0] for t, r in pts.items() if "value" in r}) - ranges = await asyncio.to_thread(lambda: {key: self.f.cache.slot_ranges(self.f.mac, group, field, values, lo, hi) - for key, (group, field, values) in wanted.items()}) + ranges = await self.f.read(lambda: {key: self.f.cache.slot_ranges(self.f.mac, group, field, values, lo, hi) + for key, (group, field, values) in wanted.items()}) for key, (lows, highs) in ranges.items(): pts = self.store[key]["pts"] for t in lows: @@ -143,10 +161,7 @@ async def _held_locally(self) -> bool: f = self.f if (self.span.days + 1) * 48 * len(f.groups) * FIELDS_PER_GROUP > MAX_ROWS: return False - for t, e in spans("30min", self.start, self.end - timedelta(days=1)): - if not await f.covered("30min", f.epoch(t), f.epoch(e)): - return False - return True + return all(await f.covered_each("30min", [(f.epoch(t), f.epoch(e)) for t, e in spans("30min", self.start, self.end - timedelta(days=1))])) async def _chunks(self, cycle: str, t: datetime, until: datetime): for a, b in spans(cycle, t, until): @@ -186,17 +201,20 @@ async def _summarise_direction(self): if not keys: return f = self.f - wind = Fetcher(f.api, f.cache, f.hot, f.mac, ["wind"], self.tz) + wind = self.wind = Fetcher(f.api, f.cache, f.hot, f.mac, ["wind"], self.tz, f.urgent) first = max(self.start, self.end - timedelta(days=DIRECTION_DAYS)) + bounds = {} # each day of the period: its part of it, local times + for n in range((self.end.date() - first.date()).days + 1): + day = first.date() + timedelta(days=n) + a, b = max(first, datetime.combine(day, time())), min(self.end, datetime.combine(day, time(23, 59, 59))) + if a < b: + bounds[day] = a, b + held = await wind.covered_each("5min", [(wind.epoch(a), wind.epoch(b)) for a, b in bounds.values()]) + fine_days = {day for day, whole in zip(bounds, held) if whole or day == self.end.date()} # today's comes from memory or one request + fine: dict = {} + for run in consecutive(sorted(fine_days)): # a run of days is one read + collect(fine, await wind.get("5min", bounds[run[0]][0], bounds[run[-1]][1]), "5min") for key in keys: - fine: dict = {} - fine_days = set() - for n in range((self.end.date() - first.date()).days + 1): - day = first.date() + timedelta(days=n) - a, b = max(first, datetime.combine(day, time())), min(self.end, datetime.combine(day, time(23, 59, 59))) - if a < b and (day == self.end.date() or await wind.covered("5min", wind.epoch(a), wind.epoch(b))): - collect(fine, await wind.get("5min", a, b), "5min") # today's comes from memory or one request - fine_days.add(day) pts = self.store.pop(key)["pts"] speed_key = key[:-len("wind_direction")] + "wind_speed" speeds = {t: r["value"][0] for src in (self.store, fine) for t, r in src.get(speed_key, {}).get("pts", {}).items() @@ -221,8 +239,11 @@ def _monthly_extremes(self) -> dict: out = {} for key, series in self.store.items(): months: dict = {} + until = None # the end of the local day the reading falls in (epoch), its month worked out once a day for ts, rec in sorted(series["pts"].items()): - month = datetime.fromtimestamp(ts, UTC).astimezone(self.tz).strftime("%b %Y") + if until is None or ts >= until: + day = local_date(ts, self.tz) + month, until = day.strftime("%b %Y"), day_bounds(day, self.tz)[1] fold(months.setdefault(month, {}), ts, rec) out[key] = months return out @@ -298,9 +319,9 @@ def _answer(self, refined: int) -> str: if key.split(".", 1)[-1] not in DERIVED or key.split(".", 1)[-1] in wanted} # the rest: cached, not sent series_out.update(self.direction) shown = {k: v for k, v in series_out.items() if self._in_focus(k)} # what the model is sent (charts use all of series_out) - log.info("%s to %s: %d ranges, %d cached, %d in mem, %d req%s, %d series, %d refined", - f"{self.start:%Y-%m-%d}", f"{self.end:%m-%d %H:%M}", f.ranges, f.from_cache, f.from_memory, f.calls, - f", {len(f.errors)} failed" if f.errors else "", len(series_out), refined) + log.log(logging.DEBUG if self.turn.ahead else logging.INFO, "%s..%s: %d reads (%d cache, %d mem), %d req%s, %d refined", + f"{self.start:%m-%d}", f"{self.end:%m-%d}", f.ranges, f.from_cache, f.from_memory, f.calls, + f", {len(f.errors)} failed" if f.errors else "", refined) out = {"period": f"{self.start:%a %d %b %Y} - {self.end:%a %d %b %Y}", "series": shown if shown else series_out} if self.monthly and not self.detailed: out["monthly_note"] = ("Monthly figures for long periods come from daily data that runs 10am-10am, so they " @@ -481,10 +502,10 @@ async def _rain_bars(self) -> Bars: the daily records, the day's total, which are approximate (10am to 10am). A figure per month or year asked for in the person's words (`by`) gets one bar for each, whatever the length of the period.""" lo, hi = self.f.epoch(self.start), self.f.epoch(self.end) - (daily,) = await asyncio.to_thread(self.f.cache.slots, self.f.mac, "30min", "rainfall", ["daily"], lo, hi) + (daily,) = await self.f.read(self.f.cache.slots, self.f.mac, "30min", "rainfall", ["daily"], lo, hi) rain = rain_slots(daily) held = {local_date(t, self.tz) for t in rain} - (old,) = await asyncio.to_thread(self.f.cache.slots, self.f.mac, "1day", "rainfall", ["daily"], lo - 86400, hi + 86400) + (old,) = await self.f.read(self.f.cache.slots, self.f.mac, "1day", "rainfall", ["daily"], lo - 86400, hi + 86400) for t, mm in old.items(): # a daily bucket is dated by the local day its 10am start falls on day = local_date(t, self.tz) if day not in held and self.start.date() <= day <= self.end.date(): diff --git a/lib/ecowitt/source.py b/lib/ecowitt/source.py index b48ccd7..ff4c0e6 100644 --- a/lib/ecowitt/source.py +++ b/lib/ecowitt/source.py @@ -100,10 +100,11 @@ def get(group: str, field: str) -> str | None: obj = (data.get(group) or {}).get(field) if not isinstance(obj, dict) or "value" not in obj: return None - return f"{obj['value']} {obj.get('unit', '')}".replace("℃", "°C").replace("mm/hr", "mm/h").strip() + unit = obj.get("unit", "").replace("℃", "°C").replace("mm/hr", "mm/h") + return f"{obj['value']}{'' if unit in ('°C', '%') else ' '}{unit}".strip() wind = get("wind", "wind_speed") parts = [f"rain {rate}" if (rate := get("rainfall", "rain_rate")) else None, - f"{daily} today" if (daily := get("rainfall", "daily")) else None, + f"{daily}/day" if (daily := get("rainfall", "daily")) else None, get("outdoor", "temperature"), get("outdoor", "humidity"), f"wind {wind}" if wind else None, get("pressure", "relative")] return ", ".join(p for p in parts if p) or "no readings" @@ -164,8 +165,9 @@ def describe(self) -> str: created = f", created {self.created:%Y-%m-%d %H:%M}" if self.created else "" return f"Ecowitt weather station '{self.station_name}'{created} (people say \"ecowitt\" or \"weather\")" - def fetcher(self, groups: list[str]) -> Fetcher: - return Fetcher(self.api, self.cache, self.hot, self.mac, groups, self.tz) + def fetcher(self, groups: list[str], urgent: bool = False) -> Fetcher: + """urgent: for a question (its requests go before the background ones).""" + return Fetcher(self.api, self.cache, self.hot, self.mac, groups, self.tz, urgent) def now(self) -> datetime: return now_local(self.tz) @@ -178,7 +180,7 @@ async def _history(self, args: dict, turn: Turn | None = None) -> str: named = str(turn.chart_field or args.get("chart_field") or "").strip().lower() # ... and what a chart of one reading needs if (r := find(named)) and r.group not in groups: groups.append(r.group) - return await HistoryQuery(self.fetcher(groups), args, turn).run() + return await HistoryQuery(self.fetcher(groups, urgent=not turn.ahead), args, turn).run() async def _days(self, args: dict, turn: Turn | None = None) -> str: return await days_tool(self.cache, self.mac, self.tz, args, turn) @@ -228,7 +230,7 @@ async def live_rain(self) -> tuple[int, dict] | None: raise ValueError("no rain reading in the response") except Exception as e: # the alert still works from the history if self._live_ok is not False: # once per outage, not every minute - log.warning("Live readings failed (%s); the rain alert uses the 5-minute history", e) + log.warning("Live readings failed: %s; rain alert uses history", e) self._live_ok = False return None if self._live_ok is None: diff --git a/lib/ecowitt/store.py b/lib/ecowitt/store.py index 80262d6..27dc46d 100644 --- a/lib/ecowitt/store.py +++ b/lib/ecowitt/store.py @@ -133,6 +133,14 @@ def coverage(self, mac: str, cycle: str, grp: str) -> list[Interval]: """Every time range held for this group at this resolution (merged, oldest first).""" return merge(self._query("SELECT start, end FROM coverage WHERE mac=? AND cycle=? AND grp=?", mac, cycle, grp)) + def coverage_of(self, mac: str, cycle: str, groups: list[str]) -> dict[str, list[Interval]]: + """coverage() of each of these groups, in one query.""" + held: dict[str, list[Interval]] = {g: [] for g in groups} + for grp, start, end in self._query(f"SELECT grp, start, end FROM coverage WHERE mac=? AND cycle=? AND grp IN ({','.join('?' * len(held))})", + mac, cycle, *held): + held[grp].append((start, end)) + return {g: merge(spans) for g, spans in held.items()} + def missing(self, mac: str, cycle: str, grp: str, start: int, end: int) -> list[Interval]: return subtract((start, end), self.coverage(mac, cycle, grp)) diff --git a/lib/logs.py b/lib/logs.py new file mode 100644 index 0000000..4b09aee --- /dev/null +++ b/lib/logs.py @@ -0,0 +1,38 @@ +"""How the bot's log lines look. Every line of an INFO or DEBUG entry is cut to the log width (LOG_WIDTH), an ellipsis (…) marking the +cut; a WARN or an ERROR is never cut. A line that quotes a request or a chat message (the quote last on the line) also has the quote +put on one line first.""" + +import logging + +QUOTES = {"quotes": True} # extra= for a line whose last argument is a request or a chat message + + +def fit(text: str, room: int) -> str: + """The text in at most `room` characters, an ellipsis marking a cut (just the ellipsis when there is no room).""" + return text if len(text) <= room else text[:max(room - 1, 0)] + "…" + + +class LogFormat(logging.Formatter): + """"INFO message": the level (WARNING is shown as WARN) and the message, with no module name. Cuts each line to the width, + except for a WARN or an ERROR.""" + + def __init__(self, fmt: str, width: int = 0): + """width: the most characters of a line of the log; 0 cuts none.""" + super().__init__(fmt) + self.width = width + + def format(self, record): + shown, text = record.levelname, record.msg + record.levelname = "WARN" if shown == "WARNING" else shown + if isinstance(text, str): # the HTTP client's own lines: "POST https://... "HTTP/1.1 200 OK"", not "HTTP Request: POST ..." + record.msg = text.removeprefix("HTTP Request: ") + try: + line = super().format(record) + finally: + record.levelname, record.msg = shown, text + quote = str(record.args[-1]) if getattr(record, "quotes", False) and isinstance(record.args, tuple) and record.args else None + if quote and line.endswith(quote): # (not when a traceback follows it) + line = line[:len(line) - len(quote)] + " ".join(quote.split()) # the quote on one line + if not self.width or record.levelno >= logging.WARNING: + return line + return "\n".join(fit(part, self.width) for part in line.split("\n")) diff --git a/lib/periods.py b/lib/periods.py index edc13ff..ed34a03 100644 --- a/lib/periods.py +++ b/lib/periods.py @@ -2,6 +2,7 @@ for that period (Bot.on_period_button); the question behind each chart is remembered for that.""" from collections import OrderedDict +from dataclasses import dataclass from telegram import InlineKeyboardButton, InlineKeyboardMarkup @@ -59,12 +60,31 @@ def remember(self, chat_id: int, message_id: int, question: str): self.save() +@dataclass +class Drawn: + """A chart in the ImageCache.""" + at: float # when it was drawn (the cache's clock) + png: bytes + title: str + file_id: str | None = None # Telegram's copy, once the chart has been sent: it is sent again by this, not uploaded + caption: str | None = None # the answer's text (kept in the chat's history), when it was made in code for exactly the key's words + ttl: float = 10 * 60 # how long it is kept (ImageCache.CHART_TTL; PRELOAD_TTL for one uploaded ahead) + + @property + def photo(self) -> bytes | str: + return self.file_id or self.png + + class ImageCache: """The charts drawn lately, by question: toggling between the period buttons brings a chart back at once instead of drawing it - again. A chart is kept CHART_TTL seconds (the readings move slowly) and the newest CHART_IMAGES stay.""" + again, and once it has been sent, as Telegram's copy instead of an upload. A chart is kept CHART_TTL seconds (the readings move + slowly) and the newest CHART_IMAGES stay. The button charts are also drawn ahead (Bot.predraw): one drawn in the last FRESH + seconds is the answer to its question (ready).""" CHART_TTL = 10 * 60 CHART_IMAGES = 60 + PRELOAD_TTL = 30 * 60 # a chart uploaded ahead (TELEGRAM_PRELOAD_CHAT) is used for this long, then drawn afresh + FRESH = 330 # a little over the time between two rounds of drawing ahead, so a press always finds one def __init__(self, clock=None): import time @@ -77,16 +97,49 @@ def key(question: str, days: int, now) -> str: from . import intent return intent.with_period(question, days, now).lower() - def get(self, key: str) -> tuple[bytes, str] | None: + @classmethod + def canonical(cls, question: str, now) -> str | None: + """The key of a question that is its own key, word for word ("Temperature chart 7d", a button's): its chart is the same + whoever asks it. None for any other ("temperature chart today" is kept under the 1d key, but it is another chart).""" + from . import intent + key = cls.key(question, intent.period_days(question, now), now) + return key if key == " ".join(question.split()).lower() else None + + def get(self, key: str) -> tuple[bytes | str, str] | None: + """(the picture: Telegram's file id once it has been sent, else the PNG; its title), or None.""" entry = self.items.get(key) - if entry is None or self.clock() - entry[0] > self.CHART_TTL: + if entry is None or self.clock() - entry.at > entry.ttl: self.items.pop(key, None) return None self.items.move_to_end(key) - return entry[1], entry[2] + return entry.photo, entry.title - def put(self, key: str, png: bytes, title: str): - self.items[key] = (self.clock(), png, title) + def has_copy(self, key: str) -> bool: + """Does Telegram hold this chart (drawn lately and already sent, so it has a file id)?""" + entry = self.items.get(key) + return entry is not None and entry.file_id is not None and self.clock() - entry.at <= entry.ttl + + def ready(self, question: str, now) -> Drawn | None: + """The chart made in code for exactly these words in the last FRESH seconds (drawn ahead, or for someone's question), + with the text that went with it; None if there is none.""" + key = self.canonical(question, now) + entry = self.items.get(key) if key else None + if entry is None or entry.caption is None or self.clock() - entry.at > self.FRESH: + return None + self.items.move_to_end(key) + return entry + + def age(self, entry: Drawn) -> float: + return self.clock() - entry.at + + def put(self, key: str, png: bytes, title: str, caption: str | None = None, ttl: float | None = None): + """caption: the answer's text, only for a chart made in code for its canonical question (then it can be ready).""" + self.items[key] = Drawn(self.clock(), png, title, caption=caption, ttl=ttl or self.CHART_TTL) self.items.move_to_end(key) while len(self.items) > self.CHART_IMAGES: self.items.popitem(last=False) + + def sent(self, key: str | None, message): + """`message` (as Telegram returned it) carries the chart kept under `key`: later sends of it use Telegram's copy.""" + if (entry := self.items.get(key)) is not None and (sizes := getattr(message, "photo", None)): + entry.file_id = sizes[-1].file_id # the largest size: the picture as it was sent diff --git a/lib/pollen/source.py b/lib/pollen/source.py index 0022131..6337c0a 100644 --- a/lib/pollen/source.py +++ b/lib/pollen/source.py @@ -54,7 +54,7 @@ async def start(self): return try: await self.warm(True) - log.info("Pollen: %s", "; ".join(self.lines()) or "no levels on the page") + log.info("Pollen: %s", self.summary()) except Exception as e: # the site may just be briefly unreachable log.warning("Pollen not readable yet: %s", e) @@ -127,6 +127,16 @@ def current(self) -> dict: "asthma": {"level": asthma, "emoji": LEVEL_EMOJI[asthma], "updated": d.get("asthma_updated")} if asthma else None, "fetched_at": self.fetched_at} + def summary(self) -> str: + """The levels in a few words for the log: "grass 🔴 High (Sat 17 Oct), asthma 🟢 Low".""" + cur, parts = self.current(), [] + if g := cur["grass"]: + when = f" ({g['date']:%a %d %b})" if g["date"] and g["date"] != self.now().date() else "" + parts.append(f"grass {g['emoji']} {g['level']}{when}") + if a := cur["asthma"]: + parts.append(f"asthma {a['emoji']} {a['level']}") + return ", ".join(parts) or "no levels on the page" + def lines(self) -> list[str]: """The ready-made report lines; the asthma line is left out when there is no risk forecast (outside the season).""" cur, out = self.current(), [] diff --git a/lib/tools.py b/lib/tools.py index 123f19e..b69f719 100644 --- a/lib/tools.py +++ b/lib/tools.py @@ -6,6 +6,8 @@ from collections.abc import Awaitable, Callable from dataclasses import dataclass, field +from .logs import QUOTES + log = logging.getLogger(__name__) MAX_OUTPUT_CHARS = 60000 @@ -25,6 +27,8 @@ class Turn: per_day: bool = False # day by day figures were asked for: a long period keeps every day, not only its extremes result_chars: int = 0 # the size of every tool result so far this question (what the model will be sent) forecast_shown: list = field(default_factory=list) # the forecast days the answer includes, remembered once it is sent + times: dict = field(default_factory=dict) # seconds spent by step ("data", "draw", "png"), for the reply's timing line in the log + ahead: bool = False # a chart drawn ahead, nobody waiting: its requests wait their turn and its routine log lines are quiet @dataclass @@ -57,7 +61,8 @@ async def call(self, name: str, raw_args: str, turn: Turn | None = None) -> str: args = json.loads(raw_args) if raw_args else {} except json.JSONDecodeError as e: return f"Error: tool arguments were not valid JSON ({e})" - log.info("Tool call %s %s", name, raw_args[:300]) + level = logging.DEBUG if turn and turn.ahead else logging.INFO + log.log(level, "Tool call %s %s", name, raw_args, extra=QUOTES) try: out = await asyncio.wait_for(tool.handler(args, turn or Turn()), TOOL_TIMEOUT) except TimeoutError: @@ -69,7 +74,7 @@ async def call(self, name: str, raw_args: str, turn: Turn | None = None) -> str: if len(out) > MAX_OUTPUT_CHARS: log.warning("Tool %s output truncated: %d chars", name, len(out)) out = out[:MAX_OUTPUT_CHARS] + "\n...[truncated - request a shorter range]" - log.info("Tool result %s: %s", name, out if len(out) < 90 else f"{len(out)} chars") + log.log(level, "Tool result %s: %s", name, out if len(out) < 90 else f"{len(out)} chars", extra=QUOTES) if turn: turn.result_chars += len(out) return out diff --git a/tests/charts/air.png b/tests/charts/air.png index a6072bb..a164156 100644 Binary files a/tests/charts/air.png and b/tests/charts/air.png differ diff --git a/tests/charts/all_week.png b/tests/charts/all_week.png index 589979f..396a970 100644 Binary files a/tests/charts/all_week.png and b/tests/charts/all_week.png differ diff --git a/tests/charts/counts.png b/tests/charts/counts.png index 35c5dad..a7837b9 100644 Binary files a/tests/charts/counts.png and b/tests/charts/counts.png differ diff --git a/tests/charts/humidity_rain.png b/tests/charts/humidity_rain.png index d83024f..3795a04 100644 Binary files a/tests/charts/humidity_rain.png and b/tests/charts/humidity_rain.png differ diff --git a/tests/charts/quarter.png b/tests/charts/quarter.png index 4078747..4b76313 100644 Binary files a/tests/charts/quarter.png and b/tests/charts/quarter.png differ diff --git a/tests/charts/temperature.png b/tests/charts/temperature.png index d7fe352..09b6189 100644 Binary files a/tests/charts/temperature.png and b/tests/charts/temperature.png differ diff --git a/tests/test_airgradient.py b/tests/test_airgradient.py index dde7a9b..a9caccb 100644 --- a/tests/test_airgradient.py +++ b/tests/test_airgradient.py @@ -137,10 +137,10 @@ async def test_missing_days_are_fetched_inline_only_up_to_a_limit(tmp_path, monk async def test_long_charts_are_averaged_but_keep_the_true_peak(tmp_path): air, _ = await make(tmp_path) ts = list(range(1_780_000_000, 1_780_000_000 + 40 * 86400, 60)) # 40 days at 1-minute readings - rows = [{"ts": t, "pm2_5": 5.0 + (300.0 if t == ts[5000] else 0.0)} for t in ts] + rows = [{"ts": t, "pm2_5": 5.0 + (300.0 if ts[5000] <= t <= ts[5002] else 0.0)} for t in ts] # three readings: the air, not a glitch line = air._chart(["pm2_5"], rows, "period").panels[0].lines[0] assert len(line.x) <= 500 - assert line.records["high"] == (ts[5000], 305.0) and "low" not in line.records # PM's lowest is often 0: not marked + assert line.records["high"][1] == 305.0 and "low" not in line.records # PM's lowest is often 0: not marked rows = [{"ts": t, "co2": 450.0 + (300.0 if t == ts[5000] else 0.0)} for t in ts] assert air._chart(["co2"], rows, "period").panels[0].lines[0].records["low"][1] == 450.0 # CO2's lowest is await air.close() @@ -150,7 +150,7 @@ async def test_bucketed_air_lines_carry_a_band_and_the_peak_label_sits_on_its_to from lib.charts import _extreme air, _ = await make(tmp_path) ts = list(range(1_780_000_000, 1_780_000_000 + 40 * 86400, 60)) - rows = [{"ts": t, "pm2_5": 5.0 + (300.0 if t == ts[5000] else 0.0)} for t in ts] + rows = [{"ts": t, "pm2_5": 5.0 + (300.0 if ts[5000] <= t <= ts[5002] else 0.0)} for t in ts] # three readings: the air, not a glitch line = air._chart(["pm2_5"], rows, "period").panels[0].lines[0] # 40 days: 4-hour buckets assert line.low is not None and max(line.y) < 305.0 and max(line.high) == 305.0 assert _extreme(line, "high", range(len(line.x)))[1] == 305.0 # the label is the true peak, at the band's top @@ -326,8 +326,8 @@ def _air_store(tmp_path): for k in range(40): day = start + timedelta(days=k) lo = int(datetime.combine(day, datetime.min.time()).replace(tzinfo=TZ).timestamp()) - spike = 40.0 if k % 10 == 0 else 8.0 # every tenth day particles reach 40 µg/m³ - store.save_day(day, [{"ts": lo + 300 * i, "pm2_5": spike if i == 100 else 5.0, "co2": 420.0 + i} for i in range(288)]) + spike = 40.0 if k % 10 == 0 else 8.0 # every tenth day particles reach 40 µg/m³ (for three readings: the air, not a glitch) + store.save_day(day, [{"ts": lo + 300 * i, "pm2_5": spike if 100 <= i <= 102 else 5.0, "co2": 420.0 + i} for i in range(288)]) return store, start @@ -348,3 +348,81 @@ async def test_air_days_counts_days_by_a_metric_and_draws_the_months_as_bars(tmp assert [r["pm2_5_max"] for r in worst["days"]] == [40.0, 40.0] bad = json.loads(await air_days_tool(store, TZ, dict(start_date=str(start), end_date=str(end), where=[{"field": "uv_max", "op": ">", "value": 1}]))) assert "unusable condition" in bad["error"] + + +def test_a_single_reading_spike_of_the_particles_is_left_out_of_a_chart_and_a_lasting_rise_is_not(): + from lib.airgradient.metrics import without_spikes + base = 1_790_000_000 + row = lambda i, pm1, pm25, pm10, co2=450: {"ts": base + 300 * i, "pm1": pm1, "pm2_5": pm25, "pm10": pm10, "co2": co2} + rows = [row(0, 1, 2, 3), row(1, 1, 2, 4), row(2, 15.3, 14.2, 28.7, 900), row(3, 3, 8, 10), row(4, 2, 4, 5), row(5, 2, 3, 4)] + kept = without_spikes(rows) + assert [r["ts"] for r in kept] == [r["ts"] for r in rows] # the moment stays, for CO2 + assert "pm10" not in kept[2] and "pm1" not in kept[2] and "pm2_5" not in kept[2] and kept[2]["co2"] == 900 + assert all(r == o for i, (r, o) in enumerate(zip(kept, rows)) if i != 2) # nothing else is touched + lasting = [row(0, 1, 2, 3), row(1, 1, 2, 4), row(2, 20, 20, 30), row(3, 22, 21, 32), row(4, 2, 4, 5)] + assert without_spikes(lasting) == lasting # two readings in a row: the air + gentle = [row(0, 3, 4, 5), row(1, 3, 4, 9), row(2, 3, 4, 5)] + assert without_spikes(gentle) == gentle # high, but not by 5 µg/m³ and 2.5 times + ends = [row(0, 1, 2, 40), row(1, 1, 2, 3), row(2, 1, 2, 3)] + assert without_spikes(ends) == ends # the first reading has one side only + apart = [row(0, 1, 2, 3), {**row(1, 1, 2, 60), "ts": base + 3600}, {**row(2, 1, 2, 3), "ts": base + 7200}] + assert without_spikes(apart) == apart # hours from its neighbours: not a spike in a series + + +async def test_a_single_reading_spike_is_in_none_of_the_figures_either(tmp_path): + from datetime import timedelta + from lib.airgradient.days import daily_values + from lib.timeutil import local_date + from tests.fakes import TZ + air, _ = await make(tmp_path) + start = air._now().replace(hour=0, minute=0, second=0) - timedelta(hours=6) + t0 = int(start.replace(tzinfo=TZ).timestamp()) + levels = [(1, 2, 3), (1, 2, 4), (15.3, 14.2, 28.7), (3, 8, 10), (2, 4, 5), (2, 3, 4)] + rows = [{"ts": t0 + 300 * i, "pm1": a, "pm2_5": b, "pm10": c} for i, (a, b, c) in enumerate(levels)] + + async def held(t0_, t1_): + return rows, 0, [] + air.rows = held + out = await air.history(start.strftime("%Y-%m-%d %H:%M:%S"), None) + assert out["pm10"]["high"] == 10 and out["pm2_5"]["high"] == 8 and out["pm1"]["high"] == 3 # not 28.7, 14.2 and 15.3 + assert out["pm10"]["average"] == round((3 + 4 + 10 + 5 + 4) / 5, 1) # and not in the mean + assert daily_values(rows, ["pm10_max", "pm2_5_max"], TZ)["pm10_max"][local_date(t0, TZ)] == 10 + await air.close() + + +def test_the_readings_a_rise_needs_are_set_by_the_antispike_setting(): + from lib.airgradient import metrics + from lib.airgradient.metrics import without_spikes + base = 1_790_000_000 + row = lambda i, v: {"ts": base + 300 * i, "pm10": v} + two = [row(0, 3), row(1, 4), row(2, 30), row(3, 31), row(4, 4), row(5, 3)] # two readings in a row + one = [row(0, 3), row(1, 4), row(2, 30), row(3, 4), row(4, 3)] + assert without_spikes(two) == two # the default, 2: two in a row is the air + assert [r.get("pm10") for r in without_spikes(two, require=3)] == [3, 4, None, None, 4, 3] + assert without_spikes(one, require=1) == one # 1: the rule is off + assert [r.get("pm10") for r in without_spikes(one)] == [3, 4, None, 4, 3] + metrics.REQUIRE_CONSECUTIVE = 3 + try: + assert [r.get("pm10") for r in without_spikes(two)] == [3, 4, None, None, 4, 3] # the module's setting is the default + finally: + metrics.REQUIRE_CONSECUTIVE = 2 + + +async def test_current_gives_the_lowest_of_the_last_readings_for_the_alerts_and_not_to_the_model(tmp_path): + from lib.airgradient import metrics + air, _ = await make(tmp_path) + out = await air.current() + assert set(out["_sustained"]) >= {"pm2_5"} and out["_sustained"]["pm2_5"] <= out["pm2_5"]["value"] + assert "_sustained" not in json.loads(await air.handle({}, None)) + await air.close() + + +def test_the_antispike_setting_is_a_number_of_readings_default_2_and_off_is_1(monkeypatch): + from lib.config import Config + for k, v in (("TELEGRAM_BOT_TOKEN", "t"), ("XAI_API_KEY", "k"), ("ECOWITT_API_KEY", "a"), ("ECOWITT_APP_KEY", "b")): + monkeypatch.setenv(k, v) + monkeypatch.delenv("ANTISPIKE_REQUIRE_CONSECUTIVE", raising=False) + assert Config.from_env().antispike_require_consecutive == 2 + for value, want in (("3", 3), ("1", 1), ("0", 1), ("off", 1), ("junk", 2), ("", 2)): + monkeypatch.setenv("ANTISPIKE_REQUIRE_CONSECUTIVE", value) + assert Config.from_env().antispike_require_consecutive == want, value diff --git a/tests/test_alert_menu.py b/tests/test_alert_menu.py index 19a9e5f..582e483 100644 --- a/tests/test_alert_menu.py +++ b/tests/test_alert_menu.py @@ -489,6 +489,18 @@ async def test_an_alert_with_a_picture_is_the_picture_with_the_text_as_its_capti assert rows(kw["reply_markup"])[0][0] == ("➕ Sub ▸", "al:open:sub:temps") # the alert's own buttons ride along +async def test_an_alerts_picture_is_uploaded_once_and_the_other_chats_are_sent_telegrams_copy(tmp_path): + class Keeps(SentWithPhotos): # Telegram answers with the photo it now holds, in its sizes (the largest last) + async def send_photo(self, chat_id, photo, **kw): + await super().send_photo(chat_id, photo, **kw) + return NS(photo=(NS(file_id="small"), NS(file_id="large"))) + state, bot = AlertState(tmp_path / "s.json"), Keeps() + for chat in (1, 2, 3): + state.add_chat(chat, str(chat)) + await Notifier(bot, state, KINDS)("It's warmer outside", kind="temps", photo=b"png") + assert [(chat, photo) for chat, photo, _ in bot.photos] == [(1, b"png"), (2, "large"), (3, "large")] + + async def test_a_picture_telegram_refuses_falls_back_to_the_text_but_a_missing_chat_is_still_forgotten(tmp_path): from telegram.error import BadRequest state, bot = AlertState(tmp_path / "s.json"), SentWithPhotos(refuse=BadRequest("Wrong file identifier")) diff --git a/tests/test_alerts.py b/tests/test_alerts.py index 9bf0d2e..ee058fd 100644 --- a/tests/test_alerts.py +++ b/tests/test_alerts.py @@ -754,3 +754,24 @@ async def notify(text, link=None, **kw): await again.check(air_reading(90)) await again.check(air_reading(90)) assert len(sent) == 1 and "mask" in sent[0] + + +async def test_a_single_spike_never_makes_the_air_unhealthy_but_two_readings_in_a_row_do(tmp_path): + state, sent = AlertState(tmp_path / "s.json"), [] + + async def notify(text, link=None, **kw): + sent.append(text) + mon = ParticulatesMonitor(FakeAir(), state, notify) + stamp = datetime.now(timezone.utc) + + def reading(minutes, latest, before): + r = air_reading(latest) + r["_time_utc"] = stamp + timedelta(minutes=minutes) + r["_sustained"] = {"pm2_5": min(latest, before)} # what air.current() gives: the lowest of the last readings + return r + await mon.check(reading(0, 90, 5)) # a spike: the reading before it was clean + await mon.check(reading(30, 90, 5)) # and again at the next check, still alone + assert not sent and state.monitor["particulates"]["above"] == 0 + await mon.check(reading(60, 90, 80)) # two in a row at this check ... + await mon.check(reading(90, 85, 90)) # ... and at the next + assert len(sent) == 1 and "Unhealthy air" in sent[0] diff --git a/tests/test_bot.py b/tests/test_bot.py index f655542..b4efff9 100644 --- a/tests/test_bot.py +++ b/tests/test_bot.py @@ -290,7 +290,7 @@ async def run(self, *a, **k): async def reply_photo(photo, caption=None, **kw): sent.append(caption) - monkeypatch.setattr(botmod, "render_chart", lambda spec, tz: b"png") + monkeypatch.setattr(botmod, "render_chart", lambda spec, tz, times=None: b"png") sources = [NS(name="Ecowitt", wants=lambda t: True, poke=lambda: None, describe=lambda: "Ecowitt")] msg = NS(chat_id=1, message_thread_id=None, is_topic_message=False, reply_photo=reply_photo, chat=NS(type="private", title=None), from_user=NS(full_name="Rob"), reply_to_message=None) @@ -304,6 +304,79 @@ async def send_chat_action(*a, **k): assert sent == ["Humidity"] # the chart's title is its whole caption +async def test_a_chart_made_in_code_starts_no_refresh_and_other_questions_do(monkeypatch): + """The chart's own reads go first in line; a refresh started beside them would only queue requests ahead of the next ones.""" + import json + from lib import bot as botmod + from lib.specs import Chart, Line, Panel + pokes = [] + + class Tools: + async def call(self, name, raw, turn=None): + turn.charts.append(Chart("t", "", [Panel("Temperature", "°C", [Line("Outdoor", [1, 2], [1.0, 2.0])])])) + return json.dumps({"period": "Fri 25 Sep 2026 - Thu 01 Oct 2026", "series": { + "outdoor.temperature": {"unit": "℃", "low": "1", "high": "2"}}}) + + class Agent: + tools = Tools() + + async def run(self, *a, **k): + return "It is 12 degrees." + + async def reply(*a, **k): + return NS(message_id=7) + monkeypatch.setattr(botmod, "render_chart", lambda spec, tz, times=None: b"png") + sources = [NS(name="Ecowitt", wants=lambda t: True, poke=lambda: pokes.append(1), describe=lambda: "Ecowitt")] + msg = NS(chat_id=1, message_thread_id=None, is_topic_message=False, reply_photo=reply, reply_text=reply, + chat=NS(type="private", title=None), from_user=NS(full_name="Rob"), reply_to_message=None) + update = NS(effective_message=msg, effective_chat=msg.chat, effective_user=NS(username="rob", full_name="Rob")) + + async def send_chat_action(*a, **k): + pass + bot, context = Bot(NS(tz=TZ), Agent(), sources, None), NS(bot=NS(send_chat_action=send_chat_action, id=99)) + await bot.respond(update, context, "Temperature chart 7d") + assert pokes == [] + await bot.respond(update, context, "what do you think about the weather today?") + assert pokes == [1] + + +async def test_a_chart_reply_logs_how_long_each_step_took(tmp_path, monkeypatch, caplog): + """One line per chart reply: reading the data, the calculation, drawing, the PNG (and its size) and sending it.""" + import re + from lib.ecowitt import Ecowitt, api + from lib.tools import Tools + from tests.fakes import config, ecowitt_transport + monkeypatch.setattr(api, "MIN_GAP_SECONDS", 0) + eco = Ecowitt(config(tmp_path), transport=ecowitt_transport()[0]) + await eco.start() + + class Agent: + tools = Tools(eco.tools) + + async def run(self, *a, **k): + raise AssertionError("no model for a plain chart") + sent = [] + + async def reply_photo(photo, caption=None, **kw): + sent.append(photo) + return NS(message_id=7) + sources = [NS(name="Ecowitt", wants=lambda t: True, poke=lambda: None, describe=lambda: "Ecowitt")] + msg = NS(chat_id=1, message_thread_id=None, is_topic_message=False, reply_photo=reply_photo, + chat=NS(type="private", title=None), from_user=NS(full_name="Rob"), reply_to_message=None) + update = NS(effective_message=msg, effective_chat=msg.chat, effective_user=NS(username="rob", full_name="Rob")) + + async def send_chat_action(*a, **k): + pass + with caplog.at_level("INFO"): + await Bot(NS(tz=TZ), Agent(), sources, None).respond(update, NS(bot=NS(send_chat_action=send_chat_action, id=99)), + "Temperature chart 1d") + await eco.close() + lines = [r.getMessage() for r in caplog.records if r.getMessage().startswith("Chart ms:")] + assert len(sent) == 1 and len(lines) == 1, lines + kb = round(len(sent[0]) / 1024) + assert re.fullmatch(rf"Chart ms: data \d+, calc \d+, draw \d+, PNG \d+ \({kb} KB\), send \d+", lines[0]) + + async def _ask_with(tool_result, text, monkeypatch=None): """Ask `text` of a bot whose one tool returns `tool_result`; returns (replies, whether the model was asked).""" asked = [] @@ -498,7 +571,7 @@ async def run(self, *a, **k): async def reply_photo(photo, caption=None, **kw): sent.append(caption) - monkeypatch.setattr(botmod, "render_chart", lambda spec, tz: b"png") + monkeypatch.setattr(botmod, "render_chart", lambda spec, tz, times=None: b"png") sources = [NS(name="Ecowitt", wants=lambda t: True, poke=lambda: None, describe=lambda: "Ecowitt")] for text, caption_wanted in (("ecowitt all week", True), ("temperature chart 7d", True)): sent.clear() diff --git a/tests/test_chart_labels.py b/tests/test_chart_labels.py index a541605..7af43f9 100644 --- a/tests/test_chart_labels.py +++ b/tests/test_chart_labels.py @@ -51,6 +51,31 @@ def test_a_latest_value_of_zero_is_not_labelled_for_any_of_them_and_other_readin assert "52.3" not in labels_of(one_line("humidity", [50.0 + (i % 7) for i in range(40)] + [52.3])) # a peak or low is tagged, the latest is not +def test_the_wind_rose_is_scaled_as_matplotlib_scales_its_bars(): + """The rose is one collection of wedges, not polar bars, so its radius is set by hand: it must be the one the bars got (5% room + past the longest wedge, none when that wedge ends in an empty step), so the rose keeps its size whatever the readings.""" + import random + + import matplotlib.pyplot as plt + import numpy as np + rng = random.Random(3) + roses = [[[4, 2, 1]] * 16, [[3 + k % 5, 0, 0] for k in range(16)], [[0, 0, 0]] * 3 + [[5, 0, 2]] + [[0, 0, 0]] * 12, + [[4, 2, 1]] * 5 + [[9, 4, 0]] + [[4, 2, 1]] * 10] + for _ in range(20): + rose = [[rng.choice([0, 0, rng.randint(1, 40)]) for _ in range(3)] for _ in range(16)] + rose[rng.randrange(16)][0] += 1 # a rose is only drawn when there are readings + roses.append(rose) + for rose in roses: + percent = np.asarray(rose, dtype=float) / np.sum(rose) * 100 + ax = plt.figure().add_axes([0, 0, 1, 1], projection="polar") + bottom = np.zeros(16) + for i in range(3): # the rose as it was drawn before + ax.bar(np.deg2rad(np.arange(16) * 22.5), percent[:, i], width=np.deg2rad(20), bottom=bottom) + bottom += percent[:, i] + assert charts._rose_radius(np.cumsum(percent, axis=1)) == pytest.approx(ax.get_ylim()[1], rel=1e-12), rose + plt.close(ax.figure) + + def test_an_end_value_is_never_written_in_exponent_form(): assert charts._figure(1014.8) == "1014.8" and charts._figure(1015.0) == "1015" and charts._figure(67.53) == "67.5" assert charts._figure(0.858) == "0.858" and charts._figure(2.0) == "2" and charts._figure(0.0) == "0" diff --git a/tests/test_ecowitt.py b/tests/test_ecowitt.py index fdee33e..767a91d 100644 --- a/tests/test_ecowitt.py +++ b/tests/test_ecowitt.py @@ -152,6 +152,62 @@ async def test_warm_then_question_needs_no_new_requests_for_the_tail(station): assert fake.calls == [] # recent tail came from memory +async def test_a_question_the_cache_and_memory_can_answer_does_not_wait_for_a_refresh_that_is_fetching(tmp_path): + import httpx + fake, answering = FakeEcowitt(), asyncio.Event() + answering.set() + + async def handler(request): + await answering.wait() + return fake(request) + eco = Ecowitt(config(tmp_path), transport=httpx.MockTransport(handler)) + await eco.start() + await eco.warm(True) # memory holds the newest readings + answering.clear() # Ecowitt goes quiet + refresh = asyncio.create_task(eco.warm(True)) + await asyncio.sleep(0.05) # the refresh's requests are out, holding the 5-minute and 30-minute fetches + now = eco.now() + question = eco.fetcher(["outdoor"], urgent=True) + data = await asyncio.wait_for(question.get("5min", now - timedelta(hours=3), now), 2) + assert data["outdoor"]["temperature"]["list"] and question.calls == 0 and question.from_memory == 1 + assert not refresh.done() + answering.set() + await refresh + await eco.close() + + +async def test_a_questions_requests_go_before_the_background_ones_and_the_refreshes_wait_their_turn(station, monkeypatch): + eco, _ = station + marked = [] # each history request's urgent flag + send = eco.api._get + + async def watch(path, urgent=False, **params): + if path == "history": + marked.append(urgent) + return await send(path, urgent, **params) + monkeypatch.setattr(eco.api, "_get", watch) + await eco.warm(True) + assert marked and not any(marked) + marked.clear() + today = eco.now().date() + await eco.tools[1].handler({"groups": "outdoor,wind", "start_date": f"{today - timedelta(days=20)} 00:00:00", + "end_date": f"{today - timedelta(days=10)} 23:59:59"}, Turn()) + assert marked and all(marked) # the question's own, and its wind rose's + marked.clear() + await eco.tools[1].handler({"groups": "outdoor,wind", "start_date": f"{today - timedelta(days=40)} 00:00:00", + "end_date": f"{today - timedelta(days=30)} 23:59:59"}, Turn(ahead=True)) + assert marked and not any(marked) # a chart drawn ahead: nobody is waiting for it + + +async def test_the_refresh_builds_no_result_it_would_not_read(station, monkeypatch): + from lib.ecowitt import fetch + eco, fake = station + merged = [] + monkeypatch.setattr(fetch, "merge_data", lambda *a, **k: merged.append(a)) + await eco.warm(True) + assert any(c["path"] == "history" for c in fake.calls) and merged == [] + + # ---------- the whole history is cached ---------- async def test_archive_caches_every_cycle_so_long_questions_need_no_requests(tmp_path, monkeypatch): from lib.ecowitt import Archive, archive as archive_mod @@ -263,7 +319,7 @@ async def test_archive_reports_how_much_history_is_held(tmp_path, monkeypatch): eco = Ecowitt(config(tmp_path), transport=transport) await eco.start() arch = Archive(eco) - assert arch.held() == "0/0/0/0 days (5min/30min/4h/1d)" + assert arch.held() == "0/0/0/0 days (5m/30m/4h/1d)" await arch.run_once() held = {c: eco.cache.days_held(eco.mac, c, arch.groups) for c in ("5min", "30min", "4hour", "1day")} assert all(55 <= held[c] <= 62 for c in ("5min", "30min", "4hour", "1day")) @@ -440,6 +496,29 @@ async def test_history_reports_direction_by_compass_point_not_a_range(tmp_path, await eco.close() +async def test_the_wind_rose_reads_its_5_minute_readings_in_runs_of_days_not_a_day_at_a_time(tmp_path, archived_cache, monkeypatch): + from lib.ecowitt import query + from lib.ecowitt.query import consecutive + eco, _ = await archived_station(tmp_path, archived_cache) + made = [] + + class Recorded(query.Fetcher): # the wind rose's own fetcher, whose reads are counted + def __init__(self, *a, **k): + super().__init__(*a, **k) + made.append(self) + monkeypatch.setattr(query, "Fetcher", Recorded) + today = eco.now().date() + turn = Turn(chart_field="wind") + out = json.loads(await eco.tools[1].handler({"groups": "wind", "chart": True, "chart_field": "wind", + "start_date": f"{today - timedelta(days=30)} 00:00:00"}, turn)) + assert out["series"]["wind.wind_direction"] and turn.charts[0].compass + assert len(made) == 1 and made[0].ranges == 1 # the archived days and today in one read, not one for each of 31 days + await eco.close() + day = lambda n: today + timedelta(days=n) + assert consecutive([day(1), day(2), day(3), day(5), day(6), day(9)]) == [[day(1), day(2), day(3)], [day(5), day(6)], [day(9)]] + assert consecutive([]) == [] + + async def test_the_wind_chart_carries_the_compass_beside_the_speed_line(tmp_path, archived_cache): from lib.charts import render eco, _ = await archived_station(tmp_path, archived_cache) @@ -989,7 +1068,7 @@ async def test_a_job_stops_asking_once_ecowitt_has_refused_twice(tmp_path): class Refusing: calls = 0 - async def history(self, *a): + async def history(self, *a, **k): Refusing.calls += 1 raise EcowittError("humidity is invalid", transient=False) @@ -1017,7 +1096,7 @@ async def test_each_weather_chart_carries_its_reading_so_it_is_drawn_in_that_rea async def test_thirty_minute_readings_without_a_range_get_one_from_the_cached_five_minute_readings(tmp_path): import time as _time - from types import SimpleNamespace as NS + from lib.ecowitt.fetch import Fetcher from lib.ecowitt.query import HistoryQuery from lib.ecowitt.store import HistoryCache from tests.fakes import TZ @@ -1027,7 +1106,9 @@ async def test_thirty_minute_readings_without_a_range_get_one_from_the_cached_fi five = {"solar_and_uvi": {"solar": {"unit": "W/m2", "list": {str(base + 300 * i): str(10.0 + i) for i in range(12)}}}} # two slots of six cache.store("M", "5min", ["solar_and_uvi"], five, base, base + 3600) start, end = datetime(2026, 1, 1), datetime(2026, 1, 2) - q = HistoryQuery(NS(tz=TZ, epoch=lambda d: base if d == start else base + 7200, cache=cache, mac="M"), {}, Turn()) + fetcher = Fetcher(None, cache, None, "M", [], TZ) + fetcher.epoch = lambda d: base if d == start else base + 7200 + q = HistoryQuery(fetcher, {}, Turn()) q.start, q.end = start, end q.store = {"solar_and_uvi.solar": {"unit": "W/m²", "pts": {base: {"cycle": "30min", "value": (12.0, "12")}, base + 1800: {"cycle": "30min", "value": (18.0, "18")}}}, @@ -1125,6 +1206,20 @@ async def test_a_cancelled_waiter_does_not_block_the_lock(): lock.release() +async def test_a_waiter_cancelled_just_as_the_request_ahead_finishes_is_a_clean_cancel(): + from lib.ecowitt.api import _TurnLock + lock = _TurnLock() + await lock.acquire() # a request in flight + waiting = asyncio.create_task(lock.acquire()) + await asyncio.sleep(0) # one queued behind it + waiting.cancel() # its question gives up ... + lock.release() # ... and the request ahead finishes before it has seen that + with pytest.raises(asyncio.CancelledError): + await waiting # cancelled, as asked (not a ValueError from the queue) + await asyncio.wait_for(lock.acquire(), 1) # and the lock is free + lock.release() + + async def test_weather_now_is_served_from_the_minute_old_live_reading_and_asks_afresh_when_it_cannot(station): import time eco, fake = station @@ -1168,6 +1263,7 @@ def test_naming_one_of_temperature_and_humidity_leaves_the_other_out_of_what_the async def test_rain_older_than_the_30_minute_readings_is_drawn_from_the_daily_records_and_by_month_when_asked(tmp_path): + from lib.ecowitt.fetch import Fetcher from lib.ecowitt.query import HistoryQuery from lib.ecowitt.store import HistoryCache from lib.tools import Turn @@ -1179,8 +1275,7 @@ async def test_rain_older_than_the_30_minute_readings_is_drawn_from_the_daily_re days = [old + timedelta(days=k) for k in range(60)] cache.store("M", "1day", ["rainfall"], {"rainfall": {"daily": {"unit": "mm", "list": {str(day_ts(d)): "2.0" for d in days}}}}, day_ts(days[0]), day_ts(days[-1]) + 86400) - f = type("F", (), {})() - f.cache, f.mac, f.tz = cache, "M", TZ + f = Fetcher(None, cache, None, "M", [], TZ) f.epoch = lambda dt: int(dt.replace(tzinfo=TZ).timestamp()) q = HistoryQuery.__new__(HistoryQuery) q.f, q.tz, q.turn = f, TZ, Turn() @@ -1220,8 +1315,8 @@ def test_the_live_summary_gives_rain_first_then_a_few_readings_and_skips_what_is cell = lambda v, u: {"value": v, "unit": u} data = {"rainfall": {"rain_rate": cell("0.0", "mm/hr"), "daily": cell("0.8", "mm")}, "outdoor": {"temperature": cell("20.3", "℃"), "humidity": cell("62", "%")}, "wind": {"wind_speed": cell("12", "km/h")}, "pressure": {"relative": cell("1013.2", "hPa")}} - assert live_summary(data) == "rain 0.0 mm/h, 0.8 mm today, 20.3 °C, 62 %, wind 12 km/h, 1013.2 hPa" - assert live_summary({"rainfall": {"daily": cell("0.8", "mm")}}) == "0.8 mm today" and live_summary({}) == "no readings" + assert live_summary(data) == "rain 0.0 mm/h, 0.8 mm/day, 20.3°C, 62%, wind 12 km/h, 1013.2 hPa" + assert live_summary({"rainfall": {"daily": cell("0.8", "mm")}}) == "0.8 mm/day" and live_summary({}) == "no readings" async def test_a_single_reading_chart_is_labelled_with_the_reading_not_its_group(tmp_path, archived_cache): diff --git a/tests/test_periods.py b/tests/test_periods.py index aed2bf9..d87fadd 100644 --- a/tests/test_periods.py +++ b/tests/test_periods.py @@ -63,7 +63,7 @@ class Agent: async def run(self, *a, **k): raise AssertionError("no model for a plain chart") - monkeypatch.setattr(botmod, "render_chart", lambda spec, tz: b"png") + monkeypatch.setattr(botmod, "render_chart", lambda spec, tz, times=None: b"png") sources = [NS(name="Ecowitt", wants=lambda t: True, poke=lambda: None, describe=lambda: "Ecowitt")] return Bot(NS(tz=TZ), Agent(), sources, None) @@ -169,6 +169,29 @@ async def answer(text=None): assert [e[0] for e in edits] == ["Temperature"] * 4 # the repeated 30d press edited nothing and bot.charted[(1, 101)] == "Temperature chart 7d" +async def test_a_chart_once_sent_goes_again_as_telegrams_copy_not_a_new_upload(monkeypatch): + bot, uploads, edits = make_bot(monkeypatch), [], [] + + async def reply_photo(photo, caption=None, **kw): + uploads.append(photo) + return NS(message_id=101, photo=(NS(file_id="small"), NS(file_id="7d"))) + + async def edit_media(media, reply_markup=None): + edits.append(media.media if isinstance(media.media, str) else "upload") + return NS(photo=(NS(file_id="small"), NS(file_id=f"edit {len(edits)}"))) + msg = NS(chat_id=1, message_id=101, message_thread_id=None, is_topic_message=False, reply_photo=reply_photo, edit_media=edit_media, + chat=NS(type="private", title=None), from_user=NS(full_name="Rob"), reply_to_message=None) + update = NS(effective_message=msg, effective_chat=msg.chat, effective_user=NS(username="rob", full_name="Rob")) + await bot.respond(update, CONTEXT, "Temperature chart 7d") + + async def answer(text=None): + pass + for days in (30, 7, 30): + await bot.on_period_button(NS(callback_query=NS(data=f"pd:{days}", message=msg, answer=answer), effective_message=msg, + effective_chat=msg.chat, effective_user=NS(username="rob", full_name="Rob")), CONTEXT) + assert uploads == [b"png"] and edits == ["upload", "7d", "edit 1"] # 30d drawn and uploaded once; then both by Telegram's copy + + async def test_redrawing_a_chart_from_a_button_shows_no_typing_or_thinking(monkeypatch): bot = make_bot(monkeypatch) actions = [] @@ -186,3 +209,137 @@ async def draft(*a, **k): assert actions == [] and drafts == [] await bot.respond(update, context, "Temperature chart 7d") # a question of its own still does assert actions + + +def pressed(bot, sent): + """A private chat's message whose photos land in `sent` as (photo, caption).""" + async def reply_photo(photo, caption=None, **kw): + sent.append((photo, caption)) + return NS(message_id=100 + len(sent), photo=(NS(file_id="small"), NS(file_id=f"copy {len(sent)}"))) + msg = NS(chat_id=1, message_id=50, message_thread_id=None, is_topic_message=False, reply_photo=reply_photo, + chat=NS(type="private", title=None), from_user=NS(full_name="Rob"), reply_to_message=None) + return NS(effective_message=msg, effective_chat=msg.chat, effective_user=NS(username="rob", full_name="Rob")) + + +async def test_the_button_charts_are_drawn_ahead_and_a_press_is_sent_one_at_once(monkeypatch, caplog): + bot, sent = make_bot(monkeypatch), [] + with caplog.at_level("INFO"): + await bot.predraw() + calls = len(bot.agent.tools.calls) + assert calls >= 1 and all(name == "weather_history" for name, _ in bot.agent.tools.calls) # the station's buttons (no air sensor here) + assert [r.getMessage() for r in caplog.records if r.getMessage().startswith("Tool ")] == [] # quiet: nobody asked + assert any(r.getMessage().startswith("Button charts drawn ahead every 300s: ") for r in caplog.records) + update = pressed(bot, sent) + await bot.respond(update, CONTEXT, "Temperature chart 1d") + assert len(bot.agent.tools.calls) == calls and sent == [(b"png", "Temperature")] # no new question: the one drawn ahead + history = bot.chats[(1, None)].history + assert history[-1]["role"] == "assistant" and history[-1]["content"] # its text is kept, as a drawn one's is + await bot.respond(update, CONTEXT, "Temperature chart 1d") + assert len(bot.agent.tools.calls) == calls and sent[-1] == ("copy 1", "Temperature") # and once sent, by Telegram's copy + caplog.clear() + with caplog.at_level("INFO"): + await bot.predraw() + assert not any("drawn ahead" in r.getMessage() for r in caplog.records) # only the first round is logged + + +async def test_a_chart_drawn_ahead_is_drawn_again_once_it_is_stale_and_never_answers_other_words(monkeypatch): + clock = [0.0] + bot, sent = make_bot(monkeypatch), [] + bot.images = periods.ImageCache(lambda: clock[0]) + await bot.predraw() + calls = len(bot.agent.tools.calls) + await bot.respond(pressed(bot, sent), CONTEXT, "temperature chart today") # kept under the 1d key, but since midnight: drawn + assert len(bot.agent.tools.calls) == calls + 1 + await bot.predraw() + calls = len(bot.agent.tools.calls) + clock[0] = periods.ImageCache.FRESH + 1 + await bot.respond(pressed(bot, sent), CONTEXT, "Temperature chart 1d") # too old to send as it is + assert len(bot.agent.tools.calls) == calls + 1 + + +async def test_charts_are_not_drawn_ahead_while_a_question_is_being_answered(monkeypatch): + import asyncio + bot = make_bot(monkeypatch) + bot.answering = 1 + bot.idle.clear() + drawing = asyncio.create_task(bot.predraw()) + await asyncio.sleep(0.05) + assert bot.agent.tools.calls == [] + bot.answering = 0 + bot.idle.set() + await drawing + assert bot.agent.tools.calls + await bot.respond(pressed(bot, []), CONTEXT, "Temperature chart 7d") + assert bot.answering == 0 and bot.idle.is_set() # counted while it ran, and let go after + + + +def preloading_bot(monkeypatch, refuse=None, clock=None): + """A bot with TELEGRAM_PRELOAD_CHAT set, and the (chat, photo, silent) of every upload to it.""" + bot, uploads = make_bot(monkeypatch), [] + bot.cfg = NS(tz=TZ, telegram_preload_chat=-1001) + if clock: + bot.images = periods.ImageCache(lambda: clock[0]) + + async def send_photo(chat, photo, caption=None, disable_notification=False): + if refuse: + raise refuse + uploads.append((chat, photo, disable_notification)) + return NS(photo=(NS(file_id="small"), NS(file_id=f"file {len(uploads)}"))) + return bot, uploads, NS(bot=NS(send_chat_action=send_chat_action, send_photo=send_photo, id=99)) + + +async def settle(bot): + import asyncio + await asyncio.gather(*bot.background) + + +async def test_the_other_periods_of_a_chart_are_uploaded_ahead_and_a_press_edits_by_file_id(monkeypatch): + bot, uploads, context = preloading_bot(monkeypatch) + sent, edits = [], [] + + async def edit_media(media, reply_markup=None): + edits.append(media.media) + return NS(photo=()) + msg = chat_message(sent, edits) + msg.edit_media, msg.message_id = edit_media, 101 + await bot.respond(NS(effective_message=msg, effective_chat=msg.chat, effective_user=NS(username="rob", full_name="Rob")), + context, "Temperature chart 7d") + await settle(bot) + assert [(c, p, silent) for c, p, silent in uploads] == [(-1001, b"png", True)] * 4 # 1D, 1M, 3M, 1Y: not the 7D just sent + calls = len(bot.agent.tools.calls) + + async def answer(text=None): + pass + await bot.on_period_button(NS(callback_query=NS(data="pd:90", message=msg, answer=answer), effective_message=msg, + effective_chat=msg.chat, effective_user=NS(username="rob", full_name="Rob")), context) + assert len(bot.agent.tools.calls) == calls and edits[0].startswith("file ") # Telegram's copy, no new question or upload + + +async def test_nothing_is_uploaded_ahead_without_the_setting_and_a_refused_upload_stops_it_with_one_warning(monkeypatch, caplog): + bot, uploads, context = preloading_bot(monkeypatch) + bot.cfg = NS(tz=TZ) + update = pressed(bot, []) + await bot.respond(update, context, "Temperature chart 7d") + await settle(bot) + assert uploads == [] and not bot.background + from telegram.error import BadRequest + bot, uploads, context = preloading_bot(monkeypatch, refuse=BadRequest("Chat not found")) + with caplog.at_level("WARNING"): + await bot.respond(pressed(bot, []), context, "Temperature chart 7d") + await settle(bot) + await bot.respond(pressed(bot, []), context, "Temperature chart 30d") + await settle(bot) + assert [r for r in caplog.records if "TELEGRAM_PRELOAD_CHAT" in r.getMessage()].__len__() == 1 and uploads == [] + + +async def test_a_chart_uploaded_ahead_is_used_for_30_minutes_and_then_drawn_afresh(monkeypatch): + clock = [0.0] + bot, uploads, context = preloading_bot(monkeypatch, clock=clock) + await bot.respond(pressed(bot, []), context, "Temperature chart 7d") + await settle(bot) + key = bot.images.key("Temperature chart 30d", 30, NOW) + clock[0] = 29 * 60 + assert bot.images.has_copy(key) and isinstance(bot.images.get(key)[0], str) + clock[0] = 31 * 60 + assert not bot.images.has_copy(key) and bot.images.get(key) is None diff --git a/tests/test_specs.py b/tests/test_specs.py index 3895888..7f3a4b9 100644 --- a/tests/test_specs.py +++ b/tests/test_specs.py @@ -100,6 +100,32 @@ def test_a_taller_chart_for_more_panels(): assert sizes[0] == sizes[1] == (1280, 720) and sizes[2][1] > 720 and sizes[3][1] > sizes[2][1] +def test_a_chart_is_a_png_of_at_most_256_colours_and_its_own_colours_are_exact(): + from PIL import Image + import io + image = Image.open(io.BytesIO(render(Chart("One", "x", [Panel("Temperature", "°C", [line("Outdoor", reading="temperature")])]), TZ))) + colours = image.convert("RGB").getcolors(256) + assert image.mode == "P" and colours is not None # None: more than 256 + assert image.convert("RGB").getpixel((0, 0)) == (255, 255, 255) # the background is pure white + assert (0xFF, 0x98, 0x30) in {rgb for _, rgb in colours} # and the line its own orange + + +def test_the_palette_keeps_every_colour_when_there_are_few_and_the_common_ones_when_there_are_many(): + import numpy as np + from lib.charts import _palette + rng = np.random.default_rng(1) + few = rng.integers(0, 256, (200, 3)).astype(float) + taken, nearest = _palette(few, rng.integers(1, 1000, 200)) + assert sorted(taken) == list(range(200)) and [taken[i] for i in nearest] == list(range(200)) # each colour is its own entry + many = rng.integers(0, 256, (3000, 3)).astype(float) + counts = rng.integers(1, 10, 3000) + counts[:20] = 100_000 # a few colours cover areas + taken, nearest = _palette(many, counts) + assert len(taken) == 256 and set(range(20)) <= set(taken) # they are kept exactly + distance = ((many[:, None, :] - many[taken][None, :, :]) ** 2).sum(axis=2) + assert np.array_equal(distance[np.arange(len(many)), nearest], distance.min(axis=1)) # and every colour is drawn in its nearest + + def test_a_daily_air_line_has_its_range_band_alone_and_in_a_multi_panel_chart(): class Air: tz = TZ diff --git a/tests/test_startup.py b/tests/test_startup.py index 5ef1f26..c47ca0e 100644 --- a/tests/test_startup.py +++ b/tests/test_startup.py @@ -51,7 +51,8 @@ def __getattr__(self, name): async def test_main_starts_warms_and_shuts_down_cleanly(tmp_path, monkeypatch, caplog): caplog.set_level("INFO") - cfg = config(tmp_path) + cfg = config(tmp_path, log_width=120) + monkeypatch.setattr(ecobot._handler.formatter, "width", 0) # put back after the test eco_t, _ = ecowitt_transport(history_days=45) air_t, _ = air_transport(oldest=datetime.now(timezone.utc) - timedelta(days=5)) app = FakeApp() @@ -65,7 +66,7 @@ async def test_main_starts_warms_and_shuts_down_cleanly(tmp_path, monkeypatch, c async def stop_when_archived(): # the background archives finish, then we shut down for _ in range(300): - if " req, held " in caplog.text and "day(s) cached" in caplog.text: + if " req, held " in caplog.text and "days cached" in caplog.text: break await asyncio.sleep(0.1) os.kill(os.getpid(), signal.SIGTERM) @@ -75,10 +76,11 @@ async def stop_when_archived(): # the background archives finish, then we shut log = caplog.text assert "Weather station: Ecowitt 'Fairleigh'" in log and "@testbot + " in log assert "Startup warm-up:" in log and " req to fetch" in log and "AirGradient archive:" in log - final = next(line for line in log.splitlines() if "Ecowitt archive:" in line and " req, held " in line) - assert "failed" not in final and "days (5min/30min/4h/1d)" in final + final = next(line for line in log.splitlines() if "Ecowitt: " in line and " req, held " in line) + assert "failed" not in final and "days (5m/30m/4h/1d)" in final assert len(app.handlers) == 10 # start+help, reset, keyboard, alerts, usage, the one button handler, other updates, membership, messages, errors assert "Alerts: rain, gusts, UV, temperature crossing, particulates: 0 chats" in log and "Alerts NOT running" not in log + assert ecobot._handler.formatter.width == 120 # LOG_WIDTH reaches the bot's own log lines async def test_the_alerts_line_leaves_out_the_weather_alerts_when_they_could_not_start(tmp_path, monkeypatch, caplog): @@ -101,7 +103,7 @@ async def broken(self): async def stop_when_archived(): for _ in range(300): - if " req, held " in caplog.text and "day(s) cached" in caplog.text: + if " req, held " in caplog.text and "days cached" in caplog.text: break await asyncio.sleep(0.1) os.kill(os.getpid(), signal.SIGTERM) @@ -239,7 +241,7 @@ def tuya(request): async def stop_when_archived(): for _ in range(300): - if " req, held " in caplog.text and "day(s) cached" in caplog.text: + if " req, held " in caplog.text and "days cached" in caplog.text: break await asyncio.sleep(0.1) os.kill(os.getpid(), signal.SIGTERM) @@ -257,7 +259,67 @@ def test_a_long_alerts_line_continues_indented_at_90_characters(): assert all(line.startswith(" " * 13) and not line.startswith(" " * 14) for line in lines[1:]) -def test_log_lines_name_their_module_except_the_main_script(): - fmt = ecobot.LogFormat("%(levelname)s %(where)s%(message)s") +def test_log_lines_do_not_name_their_module(): + fmt = ecobot.LogFormat("%(levelname)s %(message)s") record = lambda name: logging.LogRecord(name, logging.INFO, __file__, 1, "hello", None, None) - assert fmt.format(record("ecobot")) == "INFO hello" and fmt.format(record("lib.bot")) == "INFO lib.bot: hello" + assert fmt.format(record("ecobot")) == "INFO hello" and fmt.format(record("lib.bot")) == "INFO hello" + + +def test_every_log_line_is_cut_to_the_log_width_and_a_quote_is_put_on_one_line(): + from lib.logs import QUOTES, LogFormat + fmt = LogFormat("%(levelname)s %(message)s", width=40) + + def line(msg, *args, quotes=True): + record = logging.LogRecord("lib.bot", logging.INFO, __file__, 1, msg, args, None) + if quotes: + record.__dict__.update(QUOTES) + return fmt.format(record) + asked = "what was the hottest it got last week, and when?" + assert line("[@rob] %s", "how hot\nis it") == "INFO [@rob] how hot is it" # short enough: whole, on one line + assert line("[@rob] %s", asked) == "INFO [@rob] what was the hottest it got…" # 40 characters, the ellipsis included + timings = "Chart ms: data 12, calc 6, draw 112, PNG 41, send 9, more words" + assert line(timings, quotes=False) == "INFO Chart ms: data 12, calc 6, draw 11…" # a line that quotes nothing is cut too + assert line("a\n b " + "x" * 50, quotes=False).split("\n")[1] == " b " + "x" * 33 + "…" # each line of a long message + for level, name in ((logging.WARNING, "WARN"), (logging.ERROR, "ERROR")): # a warning or an error is shown whole + record = logging.LogRecord("lib.bot", level, __file__, 1, timings, None, None) + assert fmt.format(record) == f"{name} " + timings and record.levelname == logging.getLevelName(level) + fmt.width = 0 + assert line("[@rob] %s", asked) == "INFO [@rob] " + asked # 0 cuts none + assert line(timings, quotes=False) == "INFO " + timings + + +def test_the_log_width_is_71_unless_set_and_0_cuts_none(monkeypatch): + from lib.config import Config + for k, v in (("TELEGRAM_BOT_TOKEN", "t"), ("XAI_API_KEY", "k"), ("ECOWITT_API_KEY", "a"), ("ECOWITT_APP_KEY", "b")): + monkeypatch.setenv(k, v) + monkeypatch.delenv("LOG_WIDTH", raising=False) + assert Config.from_env().log_width == 71 + for value, want in (("120", 120), ("0", 0), ("junk", 71), ("-5", 71), ("7.5", 71), ("²", 71)): + monkeypatch.setenv("LOG_WIDTH", value) + assert Config.from_env().log_width == want, value + + +def test_the_preload_chat_is_a_chat_id_or_off(monkeypatch): + from lib.config import Config + for k, v in (("TELEGRAM_BOT_TOKEN", "t"), ("XAI_API_KEY", "k"), ("ECOWITT_API_KEY", "a"), ("ECOWITT_APP_KEY", "b")): + monkeypatch.setenv(k, v) + monkeypatch.delenv("TELEGRAM_PRELOAD_CHAT", raising=False) + assert Config.from_env().telegram_preload_chat is None + for value, want in (("-1001234567890", -1001234567890), ("42", 42), ("", None), ("junk", None)): + monkeypatch.setenv("TELEGRAM_PRELOAD_CHAT", value) + assert Config.from_env().telegram_preload_chat == want, value + + +def test_the_http_clients_lines_do_not_start_with_http_request(): + from lib.logs import LogFormat + fmt = LogFormat("%(levelname)s %(message)s", width=80) + record = logging.LogRecord("httpx2", logging.INFO, __file__, 1, 'HTTP Request: %s %s "%s %d %s"', + ("POST", "https://api.x.ai/v1/chat/completions", "HTTP/1.1", 200, "OK"), None) + assert fmt.format(record) == 'INFO POST https://api.x.ai/v1/chat/completions "HTTP/1.1 200 OK"' + assert record.msg.startswith("HTTP Request: ") # the record itself is left as it was + + +def test_the_alerts_list_wraps_at_the_log_width(): + from lib.alerts.menu import LABELS + lines = ecobot.wrapped("Alerts: ", list(LABELS.values()), ": 3 chats", 40).splitlines() + assert len(lines) > 1 and all(len(line) <= 40 for line in lines) diff --git a/tests/test_store.py b/tests/test_store.py index 7872a5b..ce5053f 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -27,6 +27,18 @@ def test_only_settled_readings_are_kept(tmp_path): assert cache.missing("M", "5min", "outdoor", fresh, now) # the unsettled tail is still to fetch +def test_the_coverage_of_several_groups_is_each_groups_own_read_at_once(tmp_path): + cache, now = make(tmp_path), int(time.time()) + old = now - 9 * 86400 + both = {"outdoor": data(old)["outdoor"], "wind": {"wind_speed": {"unit": "km/h", "list": {str(old): "3"}}}} + cache.store("M", "30min", ["outdoor", "wind"], both, old, old + 86400) + cache.store("M", "30min", ["outdoor"], data(old + 3 * 86400), old + 3 * 86400, old + 4 * 86400) + cache.store("M", "5min", ["wind"], both, old, old + 3600) # another cycle: not counted + held = cache.coverage_of("M", "30min", ["outdoor", "wind", "rainfall"]) + assert held == {g: cache.coverage("M", "30min", g) for g in ("outdoor", "wind", "rainfall")} + assert len(held["outdoor"]) == 2 and len(held["wind"]) == 1 and held["rainfall"] == [] + + def test_empty_old_ranges_are_remembered(tmp_path): cache, now = make(tmp_path), int(time.time()) start, end = now - 30 * 86400, now - 29 * 86400