From afbc349f18b49e682130e63014cf501e559c2f92 Mon Sep 17 00:00:00 2001 From: itbird Date: Sat, 14 Nov 2026 05:03:42 +0000 Subject: [PATCH] Add trip and charging advice with briefing charts. Learn consumption from saved trips, flag charging habits against a private home tariff, and summarize week, month, and year history with a local chart. Charge energy is the increase during a plug-in, and older rows that cannot match the battery change stay stored but drop out of totals. --- .agents/skills/agentdrive/SKILL.md | 14 +- .agents/skills/agentdrive/prompts/install.md | 4 +- .gitignore | 1 + README.md | 11 +- core/advice.py | 520 +++++++++++++++++++ core/cli.py | 77 +++ core/family.py | 1 + core/history/charts.py | 232 +++++++++ core/history/energy.py | 154 ++++++ core/history/store.py | 139 ++++- core/history/summary.py | 245 +++++++++ core/history/usage.py | 188 ++++++- core/ops.py | 36 ++ core/tariff.py | 199 +++++++ core/update.py | 1 + docs/00-before-you-start.md | 6 + docs/01-install-and-safety.md | 2 +- docs/06-queries-and-commands.md | 14 +- docs/07-assistants.md | 2 +- docs/08-skills.md | 3 +- tests/test_advice.py | 438 ++++++++++++++++ 21 files changed, 2254 insertions(+), 33 deletions(-) create mode 100644 core/advice.py create mode 100644 core/history/charts.py create mode 100644 core/history/energy.py create mode 100644 core/history/summary.py create mode 100644 core/tariff.py create mode 100644 tests/test_advice.py diff --git a/.agents/skills/agentdrive/SKILL.md b/.agents/skills/agentdrive/SKILL.md index 8405362..824349d 100644 --- a/.agents/skills/agentdrive/SKILL.md +++ b/.agents/skills/agentdrive/SKILL.md @@ -1,6 +1,6 @@ --- name: agentdrive -description: Use AgentDrive to read a personal vehicle and send parked commands through the agentdrive program. Use when the user asks about battery, climate, lock state, tyre pressure, a health check, the daily vehicle report, a coarse area, an in-car charging timetable, charging current, a departure time that preconditions the cabin, sentry mode, seat or steering heat, defrost, cabin preconditioning, the rear trunk, the front trunk, remote start, or a parked command such as flash lights, lock, or charging. Also use it when the user has more than one car or a family member asks for an action the owner must approve. Do not sign vehicle commands in this skill. +description: Use AgentDrive to read a personal vehicle and send parked commands through the agentdrive program. Use when the user asks about battery, climate, lock state, tyre pressure, a health check, the daily vehicle report, a coarse area, an in-car charging timetable, charging current, a departure time that preconditions the cabin, sentry mode, seat or steering heat, defrost, cabin preconditioning, the rear trunk, the front trunk, remote start, or a parked command such as flash lights, lock, or charging. Also use it when the user has more than one car, a family member asks for an action the owner must approve, whether the battery is enough for a trip, a charge limit, charging habits, a home tariff, or a weekly, monthly, or yearly chart. Do not sign vehicle commands in this skill. --- # AgentDrive @@ -41,7 +41,7 @@ Security trade-off of a shared key: every agent that can read the private key fi 3. `agentdrive where` — a coarse area: suburb or locality, state or region, and country. It does not wake a sleeping vehicle and it does not send a command. A rounded point is sent to Nominatim. `agentdrive where --precise` prints precise latitude and longitude and a map link when the car is already online. It still does not wake the car. Precise coordinates are not written to the local history. Do not copy them into memory. 4. A vehicle command only after doctor and check have finished, and only when the user asks. Park and speed 0 are required. Unlock, turning sentry off, the rear trunk, the front trunk, remote start, charging changes, and setting or removing a departure schedule need a fresh confirmation. A schedule cannot supply that confirmation and cannot run sentry off, the trunk, the front trunk, remote start, or a departure schedule change. -`agentdrive sentry-on` and `agentdrive sentry-off` change sentry mode. Both wake a sleeping car: if the vehicle list says the car is not online, they send one wake and poll for up to 45 seconds, then send the signed command. `sentry-off` needs a fresh confirmation that names turning sentry off, for example `I confirm turn sentry off`. A schedule cannot supply that confirmation and cannot run `sentry-off`. `agentdrive check`, `agentdrive report`, `agentdrive where`, and history do not wake a sleeping car. +`agentdrive sentry-on` and `agentdrive sentry-off` change sentry mode. Both wake a sleeping car: if the vehicle list says the car is not online, they send one wake and poll for up to 45 seconds, then send the signed command. `sentry-off` needs a fresh confirmation that names turning sentry off, for example `I confirm turn sentry off`. A schedule cannot supply that confirmation and cannot run `sentry-off`. `agentdrive check`, `agentdrive report`, `agentdrive where`, history, `agentdrive advice`, and `agentdrive tariff` do not wake a sleeping car. Advice does not change a charge limit, a timetable, or preconditioning. `agentdrive seat-heat`, `agentdrive steering-heat`, `agentdrive defrost-on`, `agentdrive defrost-off`, and `agentdrive precondition` are parked climate commands. They use the same confirmation pattern as climate on, climate off, and set temperature. `seat-heat` takes a seat the signed tool accepts (`front-left`, `front-right`, `2nd-row-left`, `2nd-row-center`, `2nd-row-right`, `3rd-row-left`, `3rd-row-right`) and a level (`off`, `low`, `medium`, or `high`). `steering-heat` is `on` or `off`. Defrost is max defrost, on or off, for the front and rear together. `precondition` starts cabin preconditioning now. It does not add a departure schedule. A departure time is `agentdrive departure`. @@ -110,6 +110,16 @@ On install, schedule a routine that runs `agentdrive check --json` every 30 minu On install, create one daily job. Run `agentdrive report-schedule`. The default is 07:30 local time and enabled, until the user changes it. The job runs `agentdrive history due` and you send that usage briefing. On Monday it is the previous week, on the 1st it is the previous month, and on 1 January it is the previous year. Every other morning it is the previous day. The command reads saved history only. It does not wake the car. This one morning job is enough. A short tip is included only on some mornings, and the same tip is not repeated for two calendar months. A service reminder is included when one is due and was not sent in the last two months. On a Monday weekly briefing, include one new official item from the Tesla site for the profile country when `history due` prints one. The item is a short summary plus the official link. When that part is absent, send the briefing without it. Do not add a second morning job, and do not fill the gap with a usage tip. `agentdrive news` prints the same official item on demand and prints nothing when there is nothing new. `agentdrive report` is the live parked snapshot. Run it when the user wants the car's current status, and send the skip line when the car is asleep. +A weekly, monthly, or yearly briefing prints a summary and a `Chart:` path. Attach that PNG to the message. The `Dashboard:` path is one local HTML file with week, month, and year views. It works offline. Do not upload the history database or the history key. The summary names kilometres, trips, average kilometres per day, charges, kWh added, the previous period, and coverage. History is opportunistic, not a continuous record. Say so when the briefing says coverage is incomplete. + +## Trip and charging advice + +`agentdrive advice trip ` answers whether the battery is enough for that many kilometres. It uses a non-waking reading when the car is already online, and otherwise the last saved battery. It learns battery percent per kilometre from saved trips. If fewer than three usable trips are saved, say that plainly and do not invent a consumption figure. It also suggests an everyday charge limit, usually 80%, and a higher limit only before a long trip. When a departure schedule exists and it is cold or hot, it suggests preconditioning. Advice only. Do not change the charge limit, start preconditioning, or edit a departure schedule unless the user asks, and then use the existing command and confirmation. + +`agentdrive advice charge` flags often charging to 100% and leaving it, often dropping very low before charging, and charging in a peak window. Store the owner's home hours with `agentdrive tariff set off-peak HH:MM-HH:MM` and `agentdrive tariff set peak HH:MM-HH:MM`. `--price` is optional, per kWh. `agentdrive tariff show` prints them. The file is `$AGENTDRIVE_HOME/tariff.yaml`. Suggest moving scheduled charging into the cheap window when the advice says so. Do not run `charge-schedule set` unless the user asks, and then it still needs a fresh confirmation. A schedule cannot supply that confirmation. + +Pass `--car ` when several cars are saved. Advice and charts stay with that car. With one car, do not pass `--car`. + `agentdrive history trips`, `history charges`, and `history snapshots` read the same saved history. `agentdrive tips` says whether a tip is due. `agentdrive service` shows kilometres since a tyre change or a service the user told you about. Log that with `agentdrive service log rotation`, `replacement`, or `maintenance`. If no odometer is saved yet, wait for the next online check or report, then log it. That does not wake the car. If the user asks for a different time, run `agentdrive report-schedule set HH:MM` and move the job to that time. If the user asks to turn the briefing off, run `agentdrive report-schedule disable` and remove the job. `agentdrive report-schedule enable` turns it back on. AgentDrive only stores the time in the private config home. It does not run in the background and it does not update the job for you. A host cron can use the example line the schedule command prints. Do not install a crontab unless the user asked for cron. diff --git a/.agents/skills/agentdrive/prompts/install.md b/.agents/skills/agentdrive/prompts/install.md index b97cbcd..3b889bf 100644 --- a/.agents/skills/agentdrive/prompts/install.md +++ b/.agents/skills/agentdrive/prompts/install.md @@ -38,7 +38,9 @@ AgentDrive does not run that job and does not install cron. Then read `agentdrive report-schedule`. The default is 07:30 local time and enabled. Create one daily job at that time that runs `agentdrive history due` and send me that usage briefing. On Monday send the previous week, on the 1st the previous month, and on 1 January the previous year. -Every other morning send the previous day. `agentdrive report` is the live snapshot when I want the car's current status, including the skip line when the vehicle is asleep. +Every other morning send the previous day. When a weekly, monthly, or yearly briefing prints a Chart path, attach that PNG. The Dashboard path is a local HTML file. Do not upload the history database. +If I ask whether the battery is enough for a trip, run `agentdrive advice trip `. If I ask about charging habits or cheap hours, run `agentdrive advice charge`. Save home tariff hours with `agentdrive tariff set` when I give you the times. Advice does not change the car. A charging change still needs my confirmation. +`agentdrive report` is the live snapshot when I want the car's current status, including the skip line when the vehicle is asleep. If I later ask for a different time, run `agentdrive report-schedule set HH:MM` and update the job. If I ask to turn the briefing off, run `agentdrive report-schedule disable` and remove the job. AgentDrive stores the time and does not run in the background. diff --git a/.gitignore b/.gitignore index 85d6e4d..5e3ef3b 100644 --- a/.gitignore +++ b/.gitignore @@ -47,6 +47,7 @@ config/runtime.yaml /cars.yaml /family.yaml /requests.json +/tariff.yaml # Editor and OS noise .DS_Store diff --git a/README.md b/README.md index f96d020..13b2b73 100644 --- a/README.md +++ b/README.md @@ -34,7 +34,7 @@ Say it in a sentence. The AI agent does the rest. **Get a health alert when something new needs you.** Tyre pressure, a low battery, a charge problem, or the car left unlocked or open. You hear about it when it is new. The same alert then stays quiet until it clears and comes back. -**Review how the car was used.** Ask for a usage review, or receive a morning briefing: the previous day, the previous week on Monday, the previous month on the 1st, and the previous year on 1 January. Some mornings include a short tip, a service reminder when one is due, and on Monday one official Tesla note when there is something new. +**Review how the car was used.** Ask for a usage review, or receive a morning briefing: the previous day, the previous week on Monday, the previous month on the 1st, and the previous year on 1 January. A week, a month, or a year includes a short summary and a chart saved on this computer. Ask whether the battery is enough for a trip, or whether charging habits and your home electricity hours suggest a change. Some mornings include a short tip, a service reminder when one is due, and on Monday one official Tesla note when there is something new. ## A conversation @@ -57,6 +57,15 @@ Alerts: none Tip: You do not need to memorize commands. Say what you want in plain words. The AI agent turns that into AgentDrive. ``` +On Monday the briefing is the previous week and can look like this. Illustrative sample. The figures are sample data, not a vehicle reading or a transcript. + +```text +Summary: 210 km driven, 9 trips, 30 km per day. Charges: 4, energy added 48.0 kWh, average 12.0 kWh per charge. +Compared with the previous week (2026-10-26 to 2026-11-01): 210 km versus 180 km, 9 trips versus 8, 48.0 kWh versus 40.0 kWh. +Coverage is incomplete: saved readings on 5 of 7 days. History is opportunistic, not a continuous record. +Chart: /home/example/briefings/weekly-2026-11-02.png +``` + ## How to start Have an AI agent that can run shell commands. Paste the prompt below. Then only handle the confirmations it asks for. About an hour, mostly approvals: you accept the disclaimer, approve a GitHub login, create the Tesla developer app, sign in to Tesla in the browser, and approve the key in the Tesla app. diff --git a/core/advice.py b/core/advice.py new file mode 100644 index 0000000..72178fb --- /dev/null +++ b/core/advice.py @@ -0,0 +1,520 @@ +"""Trip and charging advice from local history. + +A live battery or timetable is used only when the car is already online. +Nothing here wakes the car, and nothing changes a setting. A later change +still goes through the existing confirmation. +""" + +from __future__ import annotations + +import math +from datetime import datetime, timedelta, timezone + +from core.errors import ConfigError +from core.history.clock import resolve_timezone +from core.history.energy import annotate_charge +from core.history.store import history_session, latest_parked, list_named, parked_marks +from core.history.text import dumps, fmt_km, fmt_pct +from core.tariff import Tariff, TariffWindow, in_window, load_tariff + +_MIN_TRIPS = 3 +_MIN_DAYS = 3 +_MIN_RATE = 0.04 +_MAX_RATE = 1.2 +_RESERVE_PERCENT = 10 +_DAILY_BUFFER = 15 +_FULL_PERCENT = 99 +_LOW_PERCENT = 20 +_MIN_HABIT = 3 +_OFTEN = 0.5 +_ISO = "%Y-%m-%dT%H:%M:%SZ" + + +def parse_kilometers(value: str) -> float: + try: + number = float(str(value).strip()) + except (TypeError, ValueError) as exc: + raise ConfigError("Trip distance must be a number of kilometres.") from exc + if number != number or not 0 < number <= 2000: + raise ConfigError("Trip distance must be greater than 0 and at most 2000 km.") + return number + + +def render_trip_advice(home, kilometers: float, *, as_json: bool, store, peek, on=None) -> str: + zone = resolve_timezone(home) + today = on or datetime.now(zone).date() + bundle = _history(store) + tariff = load_tariff(home) + payload = trip_advice( + kilometers=kilometers, + trips=bundle["trips"], + parked=bundle["parked"], + peek=peek, + tariff=tariff, + today=today, + tz=zone, + ) + if as_json: + return dumps(payload) + return payload["text"] + + +def render_charge_advice(home, *, as_json: bool, store, peek, on=None) -> str: + zone = resolve_timezone(home) + today = on or datetime.now(zone).date() + bundle = _history(store) + tariff = load_tariff(home) + payload = charge_advice( + trips=bundle["trips"], + charges=bundle["charges"], + marks=bundle["marks"], + parked=bundle["parked"], + peek=peek, + tariff=tariff, + today=today, + tz=zone, + ) + if as_json: + return dumps(payload) + return payload["text"] + + +def trip_advice(*, kilometers, trips, parked, peek, tariff, today, tz) -> dict: + battery, battery_source = _battery(peek, parked) + temperature, temp_source = _temperature(peek, parked) + factor, weather = _weather(temperature, temp_source) + rate, used = _consumption(trips) + departure = _departure(peek) + lines = [ + "Trip advice from saved history on this computer. This does not change the car and does not wake it." + ] + enough = None + needed = None + if battery is None: + lines.append("No battery reading is available, so this cannot say whether the trip will fit.") + else: + lines.append(f"Battery is {fmt_pct(battery)} {battery_source}.") + if rate is None: + lines.append( + f"Not enough saved trips to estimate consumption: {used} usable " + f"{'trip' if used == 1 else 'trips'}. At least {_MIN_TRIPS} are needed. No guess was made." + ) + else: + lines.append( + f"Recent driving averages {rate:.2f}% of the battery per km, from {used} saved trips." + ) + if weather: + lines.append(weather) + needed = kilometers * rate * factor + required = needed + _RESERVE_PERCENT + lines.append( + f"A trip of {fmt_km(kilometers)} needs about {required:.0f}% " + f"({needed:.0f}% to drive and {_RESERVE_PERCENT}% in reserve)." + ) + if battery is not None: + enough = battery + 1e-9 >= required + if enough: + lines.append(f"{fmt_pct(battery)} is enough for this trip.") + else: + short = required - battery + lines.append(f"{fmt_pct(battery)} is not enough for this trip. It is short by about {short:.0f}%.") + limit = _charge_limit(trips, rate, factor, kilometers, today, tz) + lines.extend(limit["lines"]) + lines.extend(departure["lines"]) + return { + "text": "\n".join(lines), + "kilometers": kilometers, + "battery_percent": battery, + "battery_source": battery_source, + "consumption_percent_per_km": rate, + "trips_used": used, + "enough": enough, + "needed_percent": None if needed is None else round(needed + _RESERVE_PERCENT, 1), + "weather_factor": factor, + "charge_limit_percent": limit["daily_limit"], + "trip_limit_percent": limit["trip_limit"], + "precondition": departure["suggest"], + "advice_only": True, + "wakes_car": False, + } + + +def charge_advice(*, trips, charges, marks, parked, peek, tariff, today, tz) -> dict: + closed = [annotate_charge(item) for item in charges if not item.get("open")] + flags = _habits(closed, trips, marks, tariff, tz) + limit = _charge_limit(trips, *_rate_factor(trips, peek, parked), kilometers=None, today=today, tz=tz) + schedule = _schedule_suggestion(peek, tariff) + lines = [ + "Charging advice from saved history on this computer. This does not change the car and does not wake it." + ] + if flags["lines"]: + lines.extend(flags["lines"]) + else: + lines.append("No charging habit stands out in the saved charges.") + lines.extend(limit["lines"]) + lines.extend(_tariff_lines(tariff)) + lines.extend(schedule["lines"]) + return { + "text": "\n".join(lines), + "habits": flags["habits"], + "charge_limit_percent": limit["daily_limit"], + "tariff": [item.as_dict() for item in tariff.windows], + "schedule_suggestion": schedule["suggest"], + "advice_only": True, + "wakes_car": False, + } + + +def _history(store) -> dict: + with history_session(store, write=False) as conn: + if conn is None: + return {"trips": [], "charges": [], "parked": None, "marks": []} + trips, _total = list_named(conn, "trips", 80) + charges, _charge_total = list_named(conn, "charges", 80) + return { + "trips": trips, + "charges": charges, + "parked": latest_parked(conn), + "marks": parked_marks(conn), + } + + +def _battery(peek, parked) -> tuple[int | None, str]: + if peek is not None and getattr(peek, "state", "") == "online": + observation = getattr(peek, "observation", None) + level = getattr(observation, "battery_level", None) + if level is not None: + return int(level), "from a non-waking reading" + if parked and parked.get("battery_percent") is not None: + when = parked.get("last_seen") or "an earlier check" + return int(parked["battery_percent"]), ( + f"from the last saved snapshot at {when}. A live reading was not available, and the car was not woken" + ) + return None, "" + + +def _temperature(peek, parked) -> tuple[float | None, str]: + if peek is not None and getattr(peek, "state", "") == "online": + observation = getattr(peek, "observation", None) + temp = getattr(observation, "outside_temp", None) + if temp is not None: + return float(temp), "non-waking reading" + if parked and parked.get("outside_temp_c") is not None: + return float(parked["outside_temp_c"]), "last saved snapshot" + return None, "" + + +def _weather(temperature: float | None, source: str) -> tuple[float, str | None]: + if temperature is None: + return 1.0, "No outside temperature was saved, so no weather adjustment was applied." + shown = f"{temperature:.0f} C" + if temperature <= 0: + return 1.2, f"Outside temperature is {shown} ({source}), so the estimate adds 20 percent for cold weather." + if temperature <= 10: + return 1.1, f"Outside temperature is {shown} ({source}), so the estimate adds 10 percent for cool weather." + if temperature >= 35: + return 1.1, f"Outside temperature is {shown} ({source}), so the estimate adds 10 percent for hot weather." + return 1.0, f"Outside temperature is {shown} ({source}). No weather adjustment." + + +def _consumption(trips: list[dict]) -> tuple[float | None, int]: + rates: list[float] = [] + ordered = sorted(trips, key=lambda item: item.get("started_at") or "", reverse=True) + for trip in ordered: + if trip.get("open"): + continue + distance = trip.get("distance_km") + start = trip.get("start_battery_percent") + end = trip.get("end_battery_percent") + if distance is None or start is None or end is None: + continue + if float(distance) < 1 or int(end) >= int(start): + continue + rate = (int(start) - int(end)) / float(distance) + if _MIN_RATE <= rate <= _MAX_RATE: + rates.append(rate) + if len(rates) >= 30: + break + if len(rates) < _MIN_TRIPS: + return None, len(rates) + return sum(rates) / len(rates), len(rates) + + +def _rate_factor(trips, peek, parked) -> tuple[float | None, float]: + rate, _used = _consumption(trips) + temperature, source = _temperature(peek, parked) + factor, _text = _weather(temperature, source) + return rate, factor + + +def _charge_limit(trips, rate, factor, kilometers, today, tz) -> dict: + lines = [] + daily_km, days = _typical_day(trips, today, tz) + daily_limit = None + trip_limit = None + if rate is None or days < _MIN_DAYS or daily_km is None: + lines.append( + "Not enough saved days to learn a daily charge limit from your own driving. " + "80% is the usual everyday limit, and a long trip can justify a higher one. " + "This does not change the charge limit." + ) + if rate is not None and kilometers is not None: + needed = kilometers * rate * factor + _RESERVE_PERCENT + trip_limit = 80 if needed <= 80 else _round_limit(needed) + if trip_limit > 80: + lines.append( + f"For this trip, about {trip_limit}% would cover the estimate. " + "This does not change the charge limit." + ) + return {"lines": lines, "daily_limit": None, "trip_limit": trip_limit} + daily_need = daily_km * rate * factor + _DAILY_BUFFER + daily_limit = 80 if daily_need <= 80 else _round_limit(daily_need) + lines.append( + f"A normal day in the saved history is about {fmt_km(daily_km)}. " + f"A charge limit of {daily_limit}% covers that. This does not change the charge limit." + ) + if kilometers is not None: + needed = kilometers * rate * factor + _RESERVE_PERCENT + trip_limit = daily_limit if needed <= daily_limit else _round_limit(needed) + if trip_limit > daily_limit: + lines.append( + f"Before this trip, about {trip_limit}% would cover it. " + "This does not change the charge limit." + ) + else: + lines.append( + f"The everyday limit of {daily_limit}% already covers this trip. " + "This does not change the charge limit." + ) + return {"lines": lines, "daily_limit": daily_limit, "trip_limit": trip_limit} + + +def _typical_day(trips, today, tz) -> tuple[float | None, int]: + window_start = today - timedelta(days=28) + totals: dict = {} + for trip in trips: + if trip.get("open") or trip.get("distance_km") is None: + continue + moment = _parse(trip.get("started_at")) + if moment is None: + continue + day = moment.astimezone(tz).date() + if day < window_start or day >= today: + continue + totals[day] = totals.get(day, 0.0) + float(trip["distance_km"]) + if not totals: + return None, 0 + ordered = sorted(totals.values()) + middle = ordered[len(ordered) // 2] + return middle, len(ordered) + + +def _round_limit(percent: float) -> int: + stepped = int(math.ceil(percent / 5.0) * 5) + return max(50, min(100, stepped)) + + +def _departure(peek) -> dict: + if peek is None or getattr(peek, "state", "") != "online": + return { + "suggest": False, + "lines": [ + "Departure schedules were not read because the car is not online. The car was not woken." + ], + } + observation = getattr(peek, "observation", None) + if observation is None or not getattr(observation, "departure_schedule_section_present", False): + return { + "suggest": False, + "lines": ["Departure schedules were not in the non-waking reading."], + } + enabled = [item for item in getattr(observation, "departure_schedules", ()) or () if getattr(item, "enabled", False)] + if not enabled: + return {"suggest": False, "lines": ["No departure schedule is set, so there is no preconditioning suggestion."]} + temperature = getattr(observation, "outside_temp", None) + if temperature is None: + return { + "suggest": False, + "lines": ["A departure schedule is set. Outside temperature was not in the reading, so preconditioning was not suggested."], + } + if temperature <= 5 or temperature >= 30: + word = "cold" if temperature <= 5 else "hot" + return { + "suggest": True, + "lines": [ + ( + f"A departure schedule is set and it is {word} ({temperature:.0f} C). " + "Preconditioning the cabin before you leave is worth doing. " + "This is advice only. It does not start preconditioning and it does not change the departure schedule." + ) + ], + } + return { + "suggest": False, + "lines": [ + f"A departure schedule is set. Outside temperature is {temperature:.0f} C, so there is no extra preconditioning suggestion." + ], + } + + +def _habits(charges, trips, marks, tariff: Tariff, tz) -> dict: + known = [item for item in charges if item.get("start_percent") is not None or item.get("end_percent") is not None] + habits = [] + lines = [] + full = [item for item in known if item.get("end_percent") is not None and int(item["end_percent"]) >= _FULL_PERCENT] + left = [item for item in full if _left_full(item, trips, marks)] + if len(known) >= _MIN_HABIT and len(left) >= _MIN_HABIT and len(left) / len(known) >= _OFTEN: + habits.append("full_and_left") + lines.append( + f"Often charging to 100% and leaving it: {len(left)} of {len(known)} saved charges " + "ended at 99% or more and were still high hours later. " + "A lower everyday limit is easier on the battery. This does not change the charge limit." + ) + low = [item for item in known if item.get("start_percent") is not None and int(item["start_percent"]) <= _LOW_PERCENT] + if len(known) >= _MIN_HABIT and len(low) >= _MIN_HABIT and len(low) / len(known) >= _OFTEN: + habits.append("low_before_charge") + lines.append( + f"Often dropping very low before charging: {len(low)} of {len(known)} saved charges " + f"started at {_LOW_PERCENT}% or below. Charging sooner avoids that. This does not change the car." + ) + peaks = tariff.of_kind("peak") + if peaks: + in_peak = [item for item in known if _in_any(item, peaks, tz)] + if len(known) >= _MIN_HABIT and len(in_peak) >= _MIN_HABIT and len(in_peak) / len(known) >= _OFTEN: + habits.append("peak_window") + window = _window_phrase(peaks) + lines.append( + f"Often charging in the peak window ({window}): {len(in_peak)} of {len(known)} saved charges started then. " + "This does not change the car." + ) + return {"habits": habits, "lines": lines} + + +def _left_full(charge, trips, marks) -> bool: + end_at = _parse(charge.get("ended_at")) + if end_at is None: + return False + next_drive = None + next_battery = None + for trip in trips: + started = _parse(trip.get("started_at")) + if started is None or started <= end_at: + continue + if next_drive is None or started < next_drive: + next_drive = started + next_battery = trip.get("start_battery_percent") + if next_drive is not None and next_drive - end_at < timedelta(hours=1): + return False + if next_drive is not None and next_drive - end_at >= timedelta(hours=3): + if next_battery is None or int(next_battery) >= 90: + return True + for mark in marks: + seen = _parse(mark.get("last_seen")) + if seen is None or seen <= end_at: + continue + if next_drive is not None and seen > next_drive: + continue + battery = mark.get("battery_percent") + if battery is not None and int(battery) >= 95 and seen - end_at >= timedelta(hours=2): + return True + return False + + +def _tariff_lines(tariff: Tariff) -> list[str]: + if not tariff.windows: + return [ + "No home tariff is saved. Add off-peak and peak hours with `agentdrive tariff set` if you want this compared with your cheap window. This does not change the car." + ] + lines = ["Home tariff windows:"] + for item in tariff.windows: + price = "" if item.price_per_kwh is None else f", {item.price_per_kwh:.4g} per kWh" + lines.append(f"- {item.kind} {item.start}-{item.end}{price}") + cheap = tariff.of_kind("off-peak") + if cheap: + lines.append( + f"The cheap window is {_window_phrase(cheap)}. " + "Moving scheduled charging there is advice only. Any change still needs a confirmation." + ) + return lines + + +def _schedule_suggestion(peek, tariff: Tariff) -> dict: + cheap = tariff.of_kind("off-peak") + if not cheap: + return {"suggest": False, "lines": []} + phrase = _window_phrase(cheap) + if peek is None or getattr(peek, "state", "") != "online": + return { + "suggest": True, + "lines": [ + ( + f"The in-car charging timetable was not read because the car is not online. The car was not woken. " + f"When you want charging in the cheap window ({phrase}), changing the timetable still needs a confirmation." + ) + ], + } + observation = getattr(peek, "observation", None) + if observation is None or not getattr(observation, "charge_schedule_section_present", False): + return { + "suggest": False, + "lines": ["The in-car charging timetable was not in the non-waking reading."], + } + enabled = [item for item in getattr(observation, "charge_schedules", ()) or () if getattr(item, "enabled", False)] + if not enabled: + return { + "suggest": True, + "lines": [ + ( + f"No in-car charging timetable is set. The cheap window is {phrase}. " + "Adding one is a charging change and still needs a confirmation. This advice does not change the car." + ) + ], + } + outside = [] + for item in enabled: + if not getattr(item, "start_enabled", False) or getattr(item, "start_time", None) is None: + continue + if not any(in_window(int(item.start_time), window) for window in cheap): + outside.append(item) + if not outside: + return { + "suggest": False, + "lines": [f"The in-car charging timetable already starts in the cheap window ({phrase})."], + } + starts = ", ".join(_minutes(item.start_time) for item in outside) + return { + "suggest": True, + "lines": [ + ( + f"The in-car timetable starts at {starts}, outside the cheap window ({phrase}). " + "Moving it is a charging change and still needs a confirmation. This advice does not change the timetable." + ) + ], + } + + +def _in_any(charge, windows: tuple[TariffWindow, ...], tz) -> bool: + moment = _parse(charge.get("started_at")) + if moment is None: + return False + local = moment.astimezone(tz) + minute = local.hour * 60 + local.minute + return any(in_window(minute, window) for window in windows) + + +def _window_phrase(windows: tuple[TariffWindow, ...]) -> str: + return ", ".join(f"{item.start}-{item.end}" for item in windows) + + +def _minutes(value) -> str: + hour, minute = divmod(int(value), 60) + return f"{hour:02d}:{minute:02d}" + + +def _parse(value) -> datetime | None: + if not value: + return None + try: + return datetime.strptime(str(value), _ISO).replace(tzinfo=timezone.utc) + except ValueError: + return None diff --git a/core/cli.py b/core/cli.py index 62b31b9..b476345 100644 --- a/core/cli.py +++ b/core/cli.py @@ -42,6 +42,7 @@ from core.requests import approve_request, create_request, deny_request, list_requests from core.fleet.http import UrllibClient from core.fleet.report import format_report, format_report_json +from core.advice import parse_kilometers, render_charge_advice, render_trip_advice from core.history.official import render_news from core.history.service import perform_log, render_service_status from core.history.usage import render_briefing, render_list, render_tips @@ -62,6 +63,8 @@ from core.paths import resolve_home, resolve_key, resolve_token_file from core.redact import scrub from core.safety.install_gate import accept, assert_install_allowed, present +from core.tariff import clear_windows, format_tariff, load_tariff, set_window +from core.history.text import dumps as dump_json from core.update import update_install @@ -232,6 +235,28 @@ def build_parser() -> argparse.ArgumentParser: service_status = service_sub.add_parser("status", help="Show kilometres since logged service. Does not wake the car.") service_status.add_argument("--json", action="store_true") + advice = sub.add_parser( + "advice", + help="Trip and charging advice from saved history. Does not wake the car or change a setting.", + ) + advice_sub = advice.add_subparsers(dest="advice_action", required=True) + trip = advice_sub.add_parser("trip", help="Say whether the battery covers a trip of N km. Does not wake the car.") + trip.add_argument("kilometers", help="Trip distance in kilometres") + trip.add_argument("--json", action="store_true") + charging = advice_sub.add_parser("charge", help="Flag charging habits and a cheap window. Does not change the car.") + charging.add_argument("--json", action="store_true") + + tariff = sub.add_parser("tariff", help="Store home off-peak and peak hours. Does not change the car.") + tariff_sub = tariff.add_subparsers(dest="tariff_action", required=True) + tariff_show = tariff_sub.add_parser("show", help="Show the saved home tariff.") + tariff_show.add_argument("--json", action="store_true") + tariff_set = tariff_sub.add_parser("set", help="Save one off-peak or peak window.") + tariff_set.add_argument("kind", choices=("off-peak", "peak")) + tariff_set.add_argument("window", help="Local hours HH:MM-HH:MM") + tariff_set.add_argument("--price", type=float, default=None, help="Optional price per kWh") + tariff_clear = tariff_sub.add_parser("clear", help="Remove saved tariff windows.") + tariff_clear.add_argument("kind", nargs="?", choices=("off-peak", "peak")) + sub.add_parser("update") cars = sub.add_parser("cars", help="List cars on this account, or save a nickname. Listing does not wake a car.") @@ -306,6 +331,10 @@ def main(argv: list[str] | None = None) -> int: return _news(args, home) if args.command == "service": return _service(args, home) + if args.command == "advice": + return _advice(args, home) + if args.command == "tariff": + return _tariff(args, home) if args.command == "cars": return _cars(args, home) if args.command == "family": @@ -481,6 +510,54 @@ def _service(args, home: Path) -> int: return 0 +def _advice(args, home: Path) -> int: + _guard_read(args, home, "advice") + store = store_for_saved(home, args.car or None) + peek = _peek_parked(home, args) + if peek is not None and peek.store is not None: + store = peek.store + if args.advice_action == "trip": + kilometers = parse_kilometers(args.kilometers) + print(scrub(render_trip_advice(home, kilometers, as_json=args.json, store=store, peek=peek))) + return 0 + if args.advice_action == "charge": + print(scrub(render_charge_advice(home, as_json=args.json, store=store, peek=peek))) + return 0 + raise ConfigError("unknown command") + + +def _tariff(args, home: Path) -> int: + if getattr(args, "actor", ""): + raise ConfigError("Only the owner can edit the home tariff.") + if args.tariff_action == "show": + tariff = load_tariff(home) + if args.json: + print(dump_json({"windows": [item.as_dict() for item in tariff.windows]})) + else: + print(format_tariff(tariff)) + return 0 + if args.tariff_action == "set": + tariff = set_window(home, args.kind, args.window, args.price) + print(format_tariff(tariff)) + print("Saved in the private config home. This does not change the car.") + return 0 + if args.tariff_action == "clear": + tariff = clear_windows(home, args.kind) + print(format_tariff(tariff)) + return 0 + raise ConfigError("unknown command") + + +def _peek_parked(home: Path, args): + """Best-effort non-waking read. A missing token still leaves advice on saved history.""" + try: + return _open_app(home, args).peek(preferred=preferred_vin()) + except ConfigError: + raise + except AgentDriveError: + return None + + def _charge_schedule(args, home: Path) -> int: if args.schedule_action == "show": _guard_read(args, home, "charge-schedule-show") diff --git a/core/family.py b/core/family.py index 5b583e3..0a22b3d 100644 --- a/core/family.py +++ b/core/family.py @@ -23,6 +23,7 @@ # Low-risk reads and parked actions. The owner opts in per member. DIRECT_ACTIONS = { + "advice", "cars", "charge-schedule-show", "check", diff --git a/core/history/charts.py b/core/history/charts.py new file mode 100644 index 0000000..17650db --- /dev/null +++ b/core/history/charts.py @@ -0,0 +1,232 @@ +"""Local trend charts for a usage briefing. + +A PNG is written under the car's history directory so a chat assistant can +attach it. An HTML file next to it shows week, month, and year views with +no network access. matplotlib is used when it is installed. The builtin +PNG writer is the fallback and needs no extra package. +""" + +from __future__ import annotations + +import html +import struct +import zlib +from pathlib import Path + +_GLYPHS = { + " ": (" ", " ", " ", " ", " ", " ", " "), + "-": (" ", " ", " ", "#####", " ", " ", " "), + ".": (" ", " ", " ", " ", " ", " ## ", " ## "), + "0": (" ### ", "# #", "# ##", "# # #", "## #", "# #", " ### "), + "1": (" # ", " ## ", " # ", " # ", " # ", " # ", " ### "), + "2": (" ### ", "# #", " #", " ## ", " # ", "# ", "#####"), + "3": (" ### ", "# #", " #", " ### ", " #", "# #", " ### "), + "4": (" # ", " ## ", " # # ", "# # ", "#####", " # ", " # "), + "5": ("#####", "# ", "#### ", " #", " #", "# #", " ### "), + "6": (" ### ", "# ", "# ", "#### ", "# #", "# #", " ### "), + "7": ("#####", " #", " # ", " # ", " # ", " # ", " # "), + "8": (" ### ", "# #", "# #", " ### ", "# #", "# #", " ### "), + "9": (" ### ", "# #", "# #", " ####", " #", " #", " ### "), + "W": ("# #", "# #", "# #", "# # #", "# # #", "# # #", " # # "), + "e": (" ", " ", " ### ", "# ", "#### ", "# ", " ### "), + "h": ("# ", "# ", "#### ", "# #", "# #", "# #", "# #"), + "k": ("# ", "# #", "# # ", "### ", "# # ", "# #", "# #"), + "m": (" ", " ", "## # ", "# # #", "# # #", "# #", "# #"), + "n": (" ", " ", "#### ", "# #", "# #", "# #", "# #"), + "o": (" ", " ", " ### ", "# #", "# #", "# #", " ### "), + "M": ("# #", "## ##", "# # #", "# # #", "# #", "# #", "# #"), + "a": (" ", " ", " ### ", " #", " ####", "# #", " ####"), + "r": (" ", " ", "# ## ", "## ", "# ", "# ", "# "), + "y": (" ", " ", "# #", "# #", " ####", " #", " ### "), +} + + +def write_trend_png(path: Path, title: str, km_labels: list[str], km_values: list[float], charge_labels: list[str], charge_values: list[float]) -> str: + """Write a PNG. Return ``matplotlib`` or ``builtin``.""" + path.parent.mkdir(parents=True, exist_ok=True) + try: + _matplotlib_png(path, title, km_labels, km_values, charge_labels, charge_values) + return "matplotlib" + except Exception: + _builtin_png(path, km_labels, km_values, charge_labels, charge_values) + return "builtin" + + +def write_builtin_png(path: Path, km_labels: list[str], km_values: list[float], charge_labels: list[str], charge_values: list[float]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + _builtin_png(path, km_labels, km_values, charge_labels, charge_values) + + +def write_dashboard(path: Path, views: list[dict]) -> None: + """Write one offline HTML file. ``views`` hold numbers and short labels only.""" + path.parent.mkdir(parents=True, exist_ok=True) + sections = "\n".join(_section(view) for view in views) + document = f""" + + + +Vehicle history + + + +

Vehicle history

+

Saved on this computer. History is opportunistic, not a continuous record. This file does not contact the car.

+{sections} + + +""" + path.write_text(document, encoding="utf-8") + + +def _section(view: dict) -> str: + title = html.escape(str(view.get("title") or "Period")) + note = html.escape(str(view.get("note") or "")) + km = _svg("Kilometres", view.get("km_labels") or [], view.get("km_values") or [], "#2f6f4e") + energy = _svg("kWh per charge", view.get("charge_labels") or [], view.get("charge_values") or [], "#2c5f8a") + return f"

{title}

{note}

{km}{energy}
" + + +def _svg(caption: str, labels: list[str], values: list[float], color: str) -> str: + width = 640 + height = 160 + left = 36 + bottom = 28 + plot_w = width - left - 12 + plot_h = height - bottom - 22 + numbers = [float(item) for item in values] or [0.0] + peak = max(numbers) if numbers else 0.0 + peak = peak if peak > 0 else 1.0 + count = max(len(numbers), 1) + gap = plot_w / count + bars = [] + texts = [] + for index, value in enumerate(numbers): + bar_h = 0 if not values else (float(value) / peak) * plot_h + x = left + index * gap + gap * 0.15 + y = 18 + plot_h - bar_h + bars.append( + f'' + ) + if labels and index < len(labels) and (count <= 16 or index % max(1, count // 12) == 0): + texts.append( + f'{html.escape(str(labels[index]))}' + ) + caption_text = html.escape(caption) + return ( + f'' + f'{caption_text}' + f'' + + "".join(bars) + + "".join(texts) + + "" + ) + + +def _matplotlib_png(path, title, km_labels, km_values, charge_labels, charge_values) -> None: + import matplotlib + + matplotlib.use("Agg") + import matplotlib.pyplot as plt + + fig, axes = plt.subplots(2, 1, figsize=(8.2, 5.2), dpi=100) + _bars(axes[0], km_labels, km_values, "Kilometres", "#2f6f4e") + _bars(axes[1], charge_labels, charge_values, "kWh per charge", "#2c5f8a") + fig.suptitle(title) + fig.tight_layout() + fig.savefig(path) + plt.close(fig) + + +def _bars(axis, labels, values, ylabel, color) -> None: + points = [float(item) for item in values] + axis.bar(range(len(points)), points, color=color) + axis.set_ylabel(ylabel) + if labels: + step = max(1, len(labels) // 12) + ticks = list(range(0, len(labels), step)) + axis.set_xticks(ticks) + axis.set_xticklabels([labels[index] for index in ticks], rotation=0) + else: + axis.set_xticks([]) + + +def _builtin_png(path: Path, km_labels, km_values, charge_labels, charge_values) -> None: + width = 720 + height = 420 + pixels = bytearray([255, 255, 255] * width * height) + _panel(pixels, width, 16, 24, 688, 170, "km", km_labels, km_values, (47, 111, 78)) + _panel(pixels, width, 16, 214, 688, 170, "kWh", charge_labels, charge_values, (44, 95, 138)) + path.write_bytes(_encode_png(width, height, pixels)) + + +def _panel(pixels, width, x, y, panel_w, panel_h, caption, labels, values, color) -> None: + _text(pixels, width, len(pixels) // (width * 3), x, y, caption, (20, 20, 20), 2) + plot_top = y + 24 + plot_h = panel_h - 36 + numbers = [float(item) for item in values] + if not numbers: + _text(pixels, width, len(pixels) // (width * 3), x, plot_top + 20, "none", (80, 80, 80), 2) + return + peak = max(numbers) if max(numbers) > 0 else 1.0 + gap = panel_w / len(numbers) + for index, value in enumerate(numbers): + bar_h = int((float(value) / peak) * (plot_h - 8)) + bar_x = int(x + index * gap + gap * 0.2) + bar_w = max(1, int(gap * 0.6)) + bar_y = plot_top + plot_h - bar_h + _rect(pixels, width, len(pixels) // (width * 3), bar_x, bar_y, bar_w, bar_h, color) + _hline(pixels, width, len(pixels) // (width * 3), x, plot_top + plot_h, panel_w, (180, 180, 180)) + + +def _rect(pixels, width, height, x, y, rect_w, rect_h, color) -> None: + for py in range(y, y + rect_h): + if py < 0 or py >= height: + continue + for px in range(x, x + rect_w): + if 0 <= px < width: + index = (py * width + px) * 3 + pixels[index : index + 3] = bytes(color) + + +def _hline(pixels, width, height, x, y, line_w, color) -> None: + _rect(pixels, width, height, x, y, line_w, 1, color) + + +def _text(pixels, width, height, x, y, text, color, scale) -> None: + cursor = x + for char in text: + glyph = _GLYPHS.get(char, _GLYPHS[" "]) + for row, bits in enumerate(glyph): + for col, mark in enumerate(bits): + if mark != "#": + continue + for sy in range(scale): + for sx in range(scale): + px = cursor + col * scale + sx + py = y + row * scale + sy + if 0 <= px < width and 0 <= py < height: + index = (py * width + px) * 3 + pixels[index : index + 3] = bytes(color) + cursor += 6 * scale + + +def _encode_png(width: int, height: int, pixels: bytearray) -> bytes: + raw = bytearray() + stride = width * 3 + for y in range(height): + raw.append(0) + raw.extend(pixels[y * stride : (y + 1) * stride]) + + def chunk(tag: bytes, data: bytes) -> bytes: + crc = zlib.crc32(tag + data) & 0xFFFFFFFF + return struct.pack(">I", len(data)) + tag + data + struct.pack(">I", crc) + + ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) + return b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + chunk(b"IDAT", zlib.compress(bytes(raw), 9)) + chunk(b"IEND", b"") diff --git a/core/history/energy.py b/core/history/energy.py new file mode 100644 index 0000000..ad33e70 --- /dev/null +++ b/core/history/energy.py @@ -0,0 +1,154 @@ +"""Session energy from charge_energy_added readings. + +The car reports energy added in the current charging session. That counter +sometimes still holds the previous session, or keeps climbing across plugs +without returning to zero. Taking the maximum then stores that stale or +cumulative figure as this session's energy. + +The value saved for a new session is the increase that belongs to this plug +in. A drop is a reset: energy before the drop counts only if it rose during +this session, and energy after the reset counts from the new counter. +Summaries use the same rule on samples so an old row can be left as it was +and still kept out of a total when the saved number cannot match the battery +change. +""" + +from __future__ import annotations + +# A full-size pack is about 1.3 kWh per percent at the battery. Wall energy +# and a few kWh of battery heating sit above that. Past this line the +# reported figure does not match the percent change. +_OVERHEAD_KWH = 6.0 +_KWH_PER_PERCENT = 1.6 +_ABSOLUTE_MAX_KWH = 160.0 +_RESET_DROP_KWH = 1.0 + + +def plausible_energy(kwh: float | None, start_percent, end_percent) -> bool: + """True when kWh can belong to the observed battery change.""" + if kwh is None: + return False + try: + number = float(kwh) + except (TypeError, ValueError): + return False + if number != number or number < 0 or number > _ABSOLUTE_MAX_KWH: + return False + if start_percent is None or end_percent is None: + return True + try: + gain = int(end_percent) - int(start_percent) + except (TypeError, ValueError): + return True + if gain <= 0: + return number <= _OVERHEAD_KWH + ceiling = _OVERHEAD_KWH + gain * _KWH_PER_PERCENT + if number <= 0.05 and gain >= 5: + return False + return number <= ceiling + + +def resolve_charge_energy(values: list[float], start_percent, end_percent) -> float | None: + """Return kWh added during this plug-in, or None when nothing fits. + + ``values`` are raw charge_energy_added readings in time order. They are + not summed. A session counter that starts near zero contributes its last + reading. A counter that is already high at the first reading contributes + only its increase. A drop starts a new counter. + """ + cleaned = _clean(values) + if not cleaned: + return None + segments = _segments(cleaned) + if len(segments) > 1: + total = max(0.0, segments[0][-1] - segments[0][0]) + for segment in segments[1:]: + total += segment[-1] + if plausible_energy(total, start_percent, end_percent): + return round(total, 3) + return None + first = segments[0][0] + last = segments[0][-1] + delta = max(0.0, last - first) + session_ok = plausible_energy(last, start_percent, end_percent) + delta_ok = len(cleaned) > 1 and plausible_energy(delta, start_percent, end_percent) + cumulative = first >= 5 and last > 0 and (first / last) >= 0.45 + if cumulative and delta_ok: + return round(delta, 3) + if session_ok: + return round(last, 3) + if delta_ok: + return round(delta, 3) + return None + + +def annotate_charge(charge: dict) -> dict: + """Add the energy a summary should count. The stored number is not edited. + + ``energy_counted_kwh`` is what a total should add. ``energy_ignored`` is + true when the saved value was replaced or left out. + """ + if "energy_counted_kwh" in charge and "energy_ignored" in charge: + return charge + start = charge.get("start_percent") + end = charge.get("end_percent") + readings = charge.get("readings") or [] + raw = [item.get("energy_added_kwh") for item in readings if isinstance(item, dict)] + corrected = resolve_charge_energy(raw, start, end) if any(item is not None for item in raw) else None + stored = charge.get("energy_added_kwh") + counted, ignored = _choose(corrected, stored, start, end) + updated = dict(charge) + updated["energy_counted_kwh"] = counted + updated["energy_ignored"] = ignored + return updated + + +def _choose(corrected, stored, start, end) -> tuple[float | None, bool]: + if corrected is not None: + if stored is None: + return corrected, False + try: + saved = float(stored) + except (TypeError, ValueError): + return corrected, True + if abs(saved - corrected) <= 0.05: + return saved, False + return corrected, True + if stored is None: + return None, False + try: + saved = float(stored) + except (TypeError, ValueError): + return None, True + if plausible_energy(saved, start, end): + return saved, False + return None, True + + +def _clean(values: list[float]) -> list[float]: + cleaned: list[float] = [] + for value in values: + if value is None: + continue + try: + number = float(value) + except (TypeError, ValueError): + continue + if number != number or number < 0 or number > 1000: + continue + cleaned.append(number) + return cleaned + + +def _segments(values: list[float]) -> list[list[float]]: + segments = [[values[0]]] + for value in values[1:]: + if _reset(segments[-1][-1], value): + segments.append([value]) + else: + segments[-1].append(value) + return segments + + +def _reset(previous: float, current: float) -> bool: + return current + 0.3 < previous and previous - current >= _RESET_DROP_KWH diff --git a/core/history/store.py b/core/history/store.py index e896f1a..91d0a34 100644 --- a/core/history/store.py +++ b/core/history/store.py @@ -28,6 +28,7 @@ seal, write_private_bytes, ) +from core.history.energy import resolve_charge_energy MILES_TO_KM = 1.609344 MIN_TRIP_KM = 0.1 @@ -261,6 +262,47 @@ def latest_odometer(conn) -> float | None: return float(row[0]) +def latest_parked(conn) -> dict | None: + """Latest parked snapshot fields used by advice. No coordinates.""" + row = conn.execute( + """ + SELECT last_seen, battery_percent, outside_temp_c, charge_limit_percent + FROM snapshots + ORDER BY last_seen DESC, id DESC + LIMIT 1 + """ + ).fetchone() + if row is None: + return None + return { + "last_seen": row["last_seen"], + "battery_percent": row["battery_percent"], + "outside_temp_c": row["outside_temp_c"], + "charge_limit_percent": row["charge_limit_percent"], + } + + +def parked_marks(conn, limit: int = 400) -> list[dict]: + """Recent snapshot times and battery, newest last-seen first.""" + rows = conn.execute( + """ + SELECT observed_at, last_seen, battery_percent + FROM snapshots + ORDER BY last_seen DESC, id DESC + LIMIT ? + """, + (int(limit),), + ).fetchall() + return [ + { + "observed_at": row["observed_at"], + "last_seen": row["last_seen"], + "battery_percent": row["battery_percent"], + } + for row in rows + ] + + def recent_snapshots(conn, limit: int = 20) -> list[dict]: rows = conn.execute( """ @@ -386,15 +428,18 @@ def range_summary(conn, start_iso: str, end_iso: str) -> dict: (end_iso, start_iso), ) ] - snapshots = int( - conn.execute( + snapshot_rows = [ + {"observed_at": row["observed_at"], "last_seen": row["last_seen"]} + for row in conn.execute( """ - SELECT COUNT(*) FROM snapshots + SELECT observed_at, last_seen FROM snapshots WHERE observed_at < ? AND last_seen >= ? + ORDER BY observed_at ASC, id ASC """, (end_iso, start_iso), - ).fetchone()[0] - ) + ) + ] + snapshots = len(snapshot_rows) alerts = [ _alert_dict(row) for row in conn.execute( @@ -406,7 +451,13 @@ def range_summary(conn, start_iso: str, end_iso: str) -> dict: (start_iso, end_iso), ) ] - return {"trips": trips, "charges": charges, "snapshots": snapshots, "alerts": alerts} + return { + "trips": trips, + "charges": charges, + "snapshots": snapshots, + "snapshot_rows": snapshot_rows, + "alerts": alerts, + } def _ingest(conn, reading: Reading, observation: Observation, *, tyre_bar: float, low_battery_percent: int) -> None: @@ -586,17 +637,18 @@ def _apply_charge(conn, reading: Reading) -> None: open_id = _meta_get(conn, "open_charge_id") if reading.plugged is True: if open_id is None: + # A new plug-in must not inherit the previous counter. + _meta_set(conn, "open_charge_energy_raw", None) cursor = conn.execute( """ INSERT INTO charges( started_at, ended_at, start_percent, end_percent, energy_added_kwh, plugged_in, charging, open - ) VALUES(?, NULL, ?, ?, ?, 1, ?, 1) + ) VALUES(?, NULL, ?, ?, NULL, 1, ?, 1) """, ( reading.observed_at, reading.battery_percent, reading.battery_percent, - reading.energy_added_kwh, 1 if reading.charging else 0, ), ) @@ -604,31 +656,32 @@ def _apply_charge(conn, reading: Reading) -> None: _meta_set(conn, "open_charge_id", open_id) else: _touch_charge(conn, int(open_id), reading) + _append_charge_raw(conn, reading.energy_added_kwh) _maybe_sample(conn, int(open_id), reading) + _write_charge_energy(conn, int(open_id), reading.battery_percent) return if reading.plugged is False and open_id is not None: _close_charge(conn, int(open_id), reading) _meta_set(conn, "open_charge_id", None) + _meta_set(conn, "open_charge_energy_raw", None) def _touch_charge(conn, charge_id: int, reading: Reading) -> None: - row = conn.execute("SELECT energy_added_kwh, charging FROM charges WHERE id = ?", (charge_id,)).fetchone() + row = conn.execute("SELECT charging FROM charges WHERE id = ?", (charge_id,)).fetchone() if row is None: return - energy = row[0] - if reading.energy_added_kwh is not None: - energy = reading.energy_added_kwh if energy is None else max(float(energy), reading.energy_added_kwh) - charging = 1 if row[1] or reading.charging else 0 + charging = 1 if row[0] or reading.charging else 0 conn.execute( - "UPDATE charges SET end_percent = ?, energy_added_kwh = ?, charging = ?, plugged_in = 1 WHERE id = ?", - (reading.battery_percent, energy, charging, charge_id), + "UPDATE charges SET end_percent = ?, charging = ?, plugged_in = 1 WHERE id = ?", + (reading.battery_percent, charging, charge_id), ) def _close_charge(conn, charge_id: int, reading: Reading) -> None: - row = conn.execute("SELECT energy_added_kwh, charging FROM charges WHERE id = ?", (charge_id,)).fetchone() - energy = None if row is None else row[0] - charging = 0 if row is None else row[1] + row = conn.execute("SELECT charging FROM charges WHERE id = ?", (charge_id,)).fetchone() + charging = 0 if row is None else row[0] + # The unplug reading often reports 0. It is not energy added by this session. + energy = _resolved_open_energy(conn, charge_id, reading.battery_percent) conn.execute( """ UPDATE charges @@ -639,6 +692,56 @@ def _close_charge(conn, charge_id: int, reading: Reading) -> None: ) +def _write_charge_energy(conn, charge_id: int, end_percent) -> None: + energy = _resolved_open_energy(conn, charge_id, end_percent) + conn.execute("UPDATE charges SET energy_added_kwh = ? WHERE id = ?", (energy, charge_id)) + + +def _resolved_open_energy(conn, charge_id: int, end_percent) -> float | None: + series = _parse_raw(_meta_get(conn, "open_charge_energy_raw") or "") + if not series: + series = [ + float(row[0]) + for row in conn.execute( + """ + SELECT energy_added_kwh FROM charge_samples + WHERE charge_id = ? AND energy_added_kwh IS NOT NULL + ORDER BY observed_at ASC, id ASC + """, + (charge_id,), + ) + if row[0] is not None + ] + row = conn.execute("SELECT start_percent FROM charges WHERE id = ?", (charge_id,)).fetchone() + start = None if row is None else row[0] + return resolve_charge_energy(series, start, end_percent) + + +def _append_charge_raw(conn, raw: float | None) -> None: + if raw is None: + return + values = _parse_raw(_meta_get(conn, "open_charge_energy_raw") or "") + if values and abs(values[-1] - float(raw)) <= 1e-6: + return + values.append(float(raw)) + if len(values) > 2000: + values = values[-2000:] + _meta_set(conn, "open_charge_energy_raw", ",".join(f"{item:.3f}" for item in values)) + + +def _parse_raw(text: str) -> list[float]: + values: list[float] = [] + for part in str(text or "").split(","): + piece = part.strip() + if not piece: + continue + try: + values.append(float(piece)) + except ValueError: + continue + return values + + def _maybe_sample(conn, charge_id: int, reading: Reading) -> None: last = conn.execute( "SELECT observed_at FROM charge_samples WHERE charge_id = ? ORDER BY observed_at DESC, id DESC LIMIT 1", diff --git a/core/history/summary.py b/core/history/summary.py new file mode 100644 index 0000000..e902c74 --- /dev/null +++ b/core/history/summary.py @@ -0,0 +1,245 @@ +"""Period totals for weekly, monthly, and yearly briefings. + +Energy uses the counted figure from each charge. A saved value that cannot +match the battery change is left in the database and left out of the total. +""" + +from __future__ import annotations + +from datetime import date, datetime, timedelta, timezone + +from core.history.energy import annotate_charge +from core.history.text import fmt_km, fmt_kwh + +_ISO = "%Y-%m-%dT%H:%M:%SZ" +_LONG = {"weekly", "monthly", "yearly"} + + +def is_long_period(period: str) -> bool: + return period in _LONG + + +def previous_window(period: str, start: date, end: date) -> tuple[date, date]: + if period == "monthly": + previous_end = start + previous_start = (start - timedelta(days=1)).replace(day=1) + return previous_start, previous_end + if period == "yearly": + return date(start.year - 1, 1, 1), start + length = end - start + return start - length, start + + +def prepare_summary(summary: dict) -> dict: + """Return the same shape with each charge annotated. Rows are copies.""" + charges = [annotate_charge(item) for item in summary.get("charges") or []] + return { + "trips": list(summary.get("trips") or []), + "charges": charges, + "snapshots": summary.get("snapshots") or 0, + "snapshot_rows": list(summary.get("snapshot_rows") or []), + "alerts": list(summary.get("alerts") or []), + } + + +def period_stats(summary: dict, start: date, end: date, tz) -> dict: + trips = summary["trips"] + charges = summary["charges"] + distance, distance_complete = _distance(trips) + counted = [item.get("energy_counted_kwh") for item in charges] + energies = [float(item) for item in counted if item is not None] + ignored = sum(1 for item in charges if item.get("energy_ignored")) + replaced = sum(1 for item in charges if item.get("energy_ignored") and item.get("energy_counted_kwh") is not None) + omitted = ignored - replaced + days = max((end - start).days, 1) + energy = float(sum(energies)) if energies else None + average_charge = (energy / len(energies)) if energies else None + average_day = (distance / days) if distance is not None else None + covered = _covered_days(summary, start, end, tz) + return { + "distance_km": distance, + "distance_complete": distance_complete, + "trip_count": len(trips), + "average_km_per_day": average_day, + "charge_count": len(charges), + "energy_added_kwh": energy, + "average_kwh_per_charge": average_charge, + "charges_with_energy": len(energies), + "energy_replaced": replaced, + "energy_omitted": omitted, + "days": days, + "days_with_readings": covered, + "coverage_incomplete": covered < days, + } + + +def comparison_text(period: str, current: dict, previous: dict, prev_start: date, prev_end: date) -> str: + label = {"weekly": "week", "monthly": "month", "yearly": "year"}.get(period, "period") + inclusive = prev_end - timedelta(days=1) + window = f"{prev_start.isoformat()} to {inclusive.isoformat()}" + return ( + f"Compared with the previous {label} ({window}): " + f"{_km(current)} versus {_km(previous)}, " + f"{current['trip_count']} trips versus {previous['trip_count']}, " + f"{_kwh(current)} versus {_kwh(previous)}." + ) + + +def coverage_text(stats: dict) -> str: + seen = stats["days_with_readings"] + days = stats["days"] + if stats["coverage_incomplete"]: + return ( + f"Coverage is incomplete: saved readings on {seen} of {days} days. " + "History is opportunistic, not a continuous record." + ) + return ( + f"Saved readings fall on each day of this period. " + "History is opportunistic, not a continuous record." + ) + + +def energy_note(stats: dict) -> str | None: + replaced = stats["energy_replaced"] + omitted = stats["energy_omitted"] + if not replaced and not omitted: + return None + parts = [] + if omitted: + noun = "charge" if omitted == 1 else "charges" + parts.append(f"{omitted} {noun} left out") + if replaced: + noun = "charge" if replaced == 1 else "charges" + parts.append(f"{replaced} {noun} counted from the increase during the plug-in") + detail = " and ".join(parts) + return ( + f"Saved energy that does not match the battery change: {detail}. " + "Those saved rows were not changed." + ) + + +def summary_lines(period: str, stats: dict, previous: dict, prev_start: date, prev_end: date) -> list[str]: + distance = "distance was not recorded" if stats["distance_km"] is None else fmt_km(stats["distance_km"]) + average = "not available" if stats["average_km_per_day"] is None else f"{fmt_km(stats['average_km_per_day'])} per day" + if stats["energy_added_kwh"] is None: + energy = "energy added was not available" + per_charge = "average per charge was not available" + else: + energy = f"energy added {fmt_kwh(stats['energy_added_kwh'])}" + per_charge = f"average {fmt_kwh(stats['average_kwh_per_charge'])} per charge" + lines = [ + ( + f"Summary: {distance} driven, {stats['trip_count']} trips, {average}. " + f"Charges: {stats['charge_count']}, {energy}, {per_charge}." + ), + comparison_text(period, stats, previous, prev_start, prev_end), + coverage_text(stats), + ] + note = energy_note(stats) + if note: + lines.append(note) + return lines + + +def km_series(trips: list[dict], start: date, end: date, tz, *, monthly: bool) -> tuple[list[str], list[float]]: + if monthly: + totals: dict[tuple[int, int], float] = {} + cursor = date(start.year, start.month, 1) + last = end - timedelta(days=1) + while cursor <= last: + totals[(cursor.year, cursor.month)] = 0.0 + if cursor.month == 12: + cursor = date(cursor.year + 1, 1, 1) + else: + cursor = date(cursor.year, cursor.month + 1, 1) + for trip in trips: + moment = _local_day(trip.get("started_at"), tz) + if moment is None or not start <= moment < end or trip.get("distance_km") is None: + continue + key = (moment.year, moment.month) + if key in totals: + totals[key] += float(trip["distance_km"]) + labels = [f"{month:02d}" for (_year, month) in totals] + return labels, [round(totals[key], 3) for key in totals] + totals_by_day: dict[date, float] = {} + cursor = start + while cursor < end: + totals_by_day[cursor] = 0.0 + cursor += timedelta(days=1) + for trip in trips: + moment = _local_day(trip.get("started_at"), tz) + if moment in totals_by_day and trip.get("distance_km") is not None: + totals_by_day[moment] += float(trip["distance_km"]) + labels = [day.strftime("%d") for day in totals_by_day] + return labels, [round(totals_by_day[day], 3) for day in totals_by_day] + + +def charge_series(charges: list[dict], tz, *, limit: int = 80) -> tuple[list[str], list[float]]: + rows = [] + for charge in charges: + counted = charge.get("energy_counted_kwh") + if counted is None: + continue + moment = _local_day(charge.get("started_at"), tz) + label = moment.strftime("%d") if moment else "" + rows.append((charge.get("started_at") or "", label, float(counted))) + rows.sort() + if len(rows) > limit: + rows = rows[-limit:] + return [item[1] for item in rows], [round(item[2], 3) for item in rows] + + +def _distance(trips: list[dict]) -> tuple[float | None, bool]: + if not trips: + return 0.0, True + known = [float(item["distance_km"]) for item in trips if item.get("distance_km") is not None] + if not known: + return None, False + return float(sum(known)), len(known) == len(trips) + + +def _covered_days(summary: dict, start: date, end: date, tz) -> int: + found: set[date] = set() + for row in summary.get("snapshot_rows") or []: + first = _local_day(row.get("observed_at"), tz) + last = _local_day(row.get("last_seen"), tz) + if first is None or last is None: + continue + if last < first: + first, last = last, first + day = max(first, start) + stop = min(last, end - timedelta(days=1)) + while day <= stop: + found.add(day) + day += timedelta(days=1) + for trip in summary["trips"]: + moment = _local_day(trip.get("started_at"), tz) + if moment is not None and start <= moment < end: + found.add(moment) + for charge in summary["charges"]: + moment = _local_day(charge.get("started_at"), tz) + if moment is not None and start <= moment < end: + found.add(moment) + return len(found) + + +def _local_day(value, tz) -> date | None: + if not value: + return None + try: + moment = datetime.strptime(str(value), _ISO).replace(tzinfo=timezone.utc) + except ValueError: + return None + return moment.astimezone(tz).date() + + +def _km(stats: dict) -> str: + if stats["distance_km"] is None: + return "distance not recorded" + return fmt_km(stats["distance_km"]) + + +def _kwh(stats: dict) -> str: + if stats["energy_added_kwh"] is None: + return "energy not recorded" + return fmt_kwh(stats["energy_added_kwh"]) diff --git a/core/history/usage.py b/core/history/usage.py index 3baa956..c347d74 100644 --- a/core/history/usage.py +++ b/core/history/usage.py @@ -7,11 +7,24 @@ from __future__ import annotations from datetime import date, datetime, timedelta +from pathlib import Path from core.errors import ConfigError +from core.history.charts import write_dashboard, write_trend_png from core.history.clock import local_today, resolve_timezone, timezone_name +from core.history.energy import annotate_charge from core.history.service import take_service_reminders from core.history.store import database_exists, history_root, history_session, isoformat_utc, list_named, range_summary +from core.history.summary import ( + charge_series, + coverage_text, + is_long_period, + km_series, + period_stats, + prepare_summary, + previous_window, + summary_lines, +) from core.history.text import dumps, fmt_km, fmt_kwh, fmt_pct, yes_no from core.history.official import take_official from core.history.tips import choose_tip, is_tip_day @@ -68,6 +81,8 @@ def render_list(home, kind: str, *, limit: int, as_json: bool, store=None) -> st rows, total = [], 0 else: rows, total = list_named(conn, kind, limit) + if kind == "charges": + rows = [annotate_charge(row) for row in rows] if as_json: return dumps({kind: rows, "returned": len(rows), "total": total}) if total == 0: @@ -106,21 +121,48 @@ def render_briefing(home, period: str, *, as_json: bool, on: date | None = None, start, end = covered(selected, today) start_iso, end_iso = _bounds(start, end, zone) root = history_root(home, store) + views: dict[str, tuple[date, date, dict]] = {} + previous_summary = _blank_summary() if not database_exists(root) and not is_tip_day(today): - summary = {"trips": [], "charges": [], "snapshots": 0, "alerts": []} + summary = _blank_summary() tip = None reminders: list[dict] = [] + if is_long_period(selected): + views = _empty_views(today) else: with history_session(root, write=True) as conn: - summary = range_summary(conn, start_iso, end_iso) + if is_long_period(selected): + views = _load_views(conn, today, zone) + summary = views[selected][2] + prev_start, prev_end = previous_window(selected, start, end) + previous_summary = range_summary(conn, *_bounds(prev_start, prev_end, zone)) + else: + summary = range_summary(conn, start_iso, end_iso) tip, _reason = choose_tip(conn, today) reminders = take_service_reminders(conn, today) + if is_long_period(selected): + if not views: + views = _empty_views(today) + views = { + name: (item_start, item_end, prepare_summary(item)) + for name, (item_start, item_end, item) in views.items() + } + summary = views[selected][2] + previous_summary = prepare_summary(previous_summary) + else: + summary = prepare_summary(summary) official = None if selected == "weekly": try: official = take_official(home, today, tz=zone, store=root) except Exception: official = None + files = None + if is_long_period(selected): + try: + files = _write_files(root, selected, start, end, summary, views, zone) + except Exception: + files = None payload = _payload( selected=selected, explicit=period, @@ -131,20 +173,36 @@ def render_briefing(home, period: str, *, as_json: bool, on: date | None = None, tip=tip, reminders=reminders, official=official, + zone=zone, + files=files, + previous_summary=previous_summary, ) if as_json: return dumps(payload) return payload["text"] -def _payload(*, selected, explicit, start, end, zone_name, summary, tip, reminders, official) -> dict: +def _payload(*, selected, explicit, start, end, zone_name, summary, tip, reminders, official, zone, files, previous_summary) -> dict: inclusive_end = end - timedelta(days=1) trips = summary["trips"] charges = summary["charges"] alerts = summary["alerts"] distance, distance_complete = _distance(trips) - energy = _sum(charges, "energy_added_kwh") + energy = _counted_energy(charges) readings = sum(len(item["readings"]) for item in charges) + period_block = None + if is_long_period(selected): + prev_start, prev_end = previous_window(selected, start, end) + previous = period_stats(previous_summary, prev_start, prev_end, zone) + stats = period_stats(summary, start, end, zone) + period_block = { + "stats": stats, + "previous": previous, + "prev_start": prev_start, + "prev_end": prev_end, + "lines": summary_lines(selected, stats, previous, prev_start, prev_end), + "files": files, + } trip_rows, trips_truncated = _cap(trips) charge_rows, charges_truncated = _cap(charges) alert_rows, alerts_truncated = _cap(alerts) @@ -166,6 +224,7 @@ def _payload(*, selected, explicit, start, end, zone_name, summary, tip, reminde tip=tip, reminders=reminders, official=official, + period_block=period_block, ) return { "selected": selected, @@ -195,6 +254,7 @@ def _payload(*, selected, explicit, start, end, zone_name, summary, tip, reminde "tip": None if tip is None else {"id": tip["id"], "text": tip["text"]}, "service_reminders": reminders, "official": official, + "summary": None if period_block is None else _public_summary(period_block), } @@ -216,6 +276,7 @@ def _text( tip, reminders, official, + period_block, ) -> str: lines = [] if explicit == "due": @@ -236,6 +297,15 @@ def _text( lines.append(f"- and {len(alerts) - 8} more") else: lines.append("Alerts: none") + if period_block is not None: + lines.extend(period_block["lines"]) + files = period_block.get("files") or {} + if files.get("chart"): + lines.append(f"Chart: {files['chart']}") + if files.get("dashboard"): + lines.append(f"Dashboard: {files['dashboard']}") + elif is_long_period(selected): + lines.append("Chart: could not be saved on this computer.") if reminders: lines.append("Service reminders:") for item in reminders: @@ -281,7 +351,11 @@ def _charge_summary(charges, energy, readings) -> str: energy_text = "energy added was not in the saved readings" else: energy_text = f"energy added {fmt_kwh(energy)}" - return f"Charges: {len(charges)}, {energy_text}, {readings} charge readings." + ignored = sum(1 for item in charges if item.get("energy_ignored")) + note = "" + if ignored: + note = " Saved energy that does not match the battery change was left out of this total or corrected. Those saved rows were not changed." + return f"Charges: {len(charges)}, {energy_text}, {readings} charge readings.{note}" def _list_line(kind: str, row: dict) -> str: @@ -298,7 +372,21 @@ def _list_line(kind: str, row: dict) -> str: ) if kind == "charges": end = row["ended_at"] or "still plugged in" - energy = fmt_kwh(row["energy_added_kwh"]) if row["energy_added_kwh"] is not None else "energy not recorded" + counted = row.get("energy_counted_kwh") + if row.get("energy_ignored") and counted is None: + saved = fmt_kwh(row["energy_added_kwh"]) if row.get("energy_added_kwh") is not None else "saved value" + energy = f"{saved} not counted (does not match the battery change)" + elif row.get("energy_ignored") and counted is not None: + energy = ( + f"{fmt_kwh(counted)} counted " + f"(saved {fmt_kwh(row.get('energy_added_kwh'))} does not match the battery change)" + ) + elif counted is not None: + energy = fmt_kwh(counted) + elif row.get("energy_added_kwh") is not None: + energy = fmt_kwh(row["energy_added_kwh"]) + else: + energy = "energy not recorded" return ( f"{row['started_at']} to {end} | battery {fmt_pct(row['start_percent'])} to {fmt_pct(row['end_percent'])} | " f"{energy} | {len(row['readings'])} charge readings" @@ -335,6 +423,94 @@ def _distance(trips: list[dict]) -> tuple[float | None, bool]: return float(sum(known)), len(known) == len(trips) +def _counted_energy(charges: list[dict]) -> float | None: + values = [float(item["energy_counted_kwh"]) for item in charges if item.get("energy_counted_kwh") is not None] + if not values: + return None + return float(sum(values)) + + +def _public_summary(period_block: dict) -> dict: + stats = period_block["stats"] + previous = period_block["previous"] + files = period_block.get("files") or {} + return { + "distance_km": stats["distance_km"], + "trip_count": stats["trip_count"], + "average_km_per_day": stats["average_km_per_day"], + "charge_count": stats["charge_count"], + "energy_added_kwh": stats["energy_added_kwh"], + "average_kwh_per_charge": stats["average_kwh_per_charge"], + "energy_replaced": stats["energy_replaced"], + "energy_omitted": stats["energy_omitted"], + "coverage": { + "days_with_readings": stats["days_with_readings"], + "days": stats["days"], + "incomplete": stats["coverage_incomplete"], + }, + "previous": { + "distance_km": previous["distance_km"], + "trip_count": previous["trip_count"], + "charge_count": previous["charge_count"], + "energy_added_kwh": previous["energy_added_kwh"], + }, + "chart": files.get("chart"), + "dashboard": files.get("dashboard"), + "chart_renderer": files.get("renderer"), + } + + +def _blank_summary() -> dict: + return {"trips": [], "charges": [], "snapshots": 0, "snapshot_rows": [], "alerts": []} + + +def _empty_views(today: date) -> dict: + return {name: (*covered(name, today), _blank_summary()) for name in ("weekly", "monthly", "yearly")} + + +def _load_views(conn, today: date, zone) -> dict: + loaded = {} + for name in ("weekly", "monthly", "yearly"): + start, end = covered(name, today) + loaded[name] = (start, end, range_summary(conn, *_bounds(start, end, zone))) + return loaded + + +def _write_files(root, selected, start, end, summary, views, zone) -> dict: + folder = Path(root) / "briefings" + monthly = selected == "yearly" + km_labels, km_values = km_series(summary["trips"], start, end, zone, monthly=monthly) + charge_labels, charge_values = charge_series(summary["charges"], zone) + chart = folder / f"{selected}-{start.isoformat()}.png" + title = f"{selected} vehicle history {start.isoformat()}" + renderer = write_trend_png(chart, title, km_labels, km_values, charge_labels, charge_values) + html_views = [] + for name in ("weekly", "monthly", "yearly"): + view_start, view_end, view = views[name] + view_monthly = name == "yearly" + labels, values = km_series(view["trips"], view_start, view_end, zone, monthly=view_monthly) + c_labels, c_values = charge_series(view["charges"], zone) + stats = period_stats(view, view_start, view_end, zone) + inclusive = view_end - timedelta(days=1) + html_views.append( + { + "title": f"{name.title()} {view_start.isoformat()} to {inclusive.isoformat()}", + "note": coverage_text(stats), + "km_labels": labels, + "km_values": values, + "charge_labels": c_labels, + "charge_values": c_values, + } + ) + dashboard = folder / "dashboard.html" + write_dashboard(dashboard, html_views) + return { + "chart": str(chart.resolve()), + "dashboard": str(dashboard.resolve()), + "renderer": renderer, + } + + def _sum(rows: list[dict], key: str) -> float | None: values = [float(item[key]) for item in rows if item.get(key) is not None] if not values: diff --git a/core/ops.py b/core/ops.py index f43fc0c..d66251b 100644 --- a/core/ops.py +++ b/core/ops.py @@ -15,6 +15,7 @@ import json import sys import time +from dataclasses import dataclass from pathlib import Path from adapters.template import TemplateAdapter @@ -227,6 +228,16 @@ def resolve_thresholds( return float(tyre), int(battery) +@dataclass(frozen=True) +class ParkedPeek: + """A read that did not wake the car. ``observation`` is set only when it is already online.""" + + state: str + observation: Observation | None + store: Path | None + label: str + + class App: """Wires the fleet reader, token manager, and signed-command runner.""" @@ -276,6 +287,31 @@ def check(self, *, preferred: str | None, tyre_bar: float, low_battery_percent: self._remember(payload, observation, tyre_bar=tyre_bar, low_battery_percent=low_battery_percent, store=selection.store) return apply_rising_edge(selection.store, report) + def peek(self, *, preferred: str | None) -> ParkedPeek: + """Read battery, climate, and schedules when the car is already online. + + A car that is not online is left asleep. This never calls wake. + """ + try: + selection = self._bind(preferred) + except ConfigError: + raise + except (VehicleAsleep, TimeoutError, HttpError, QueryError, TokenError): + return ParkedPeek("unavailable", None, None, "") + state = vehicle_state_name(selection.row) or "unknown" + if state != "online": + return ParkedPeek(state, None, selection.store, selection.label) + try: + payload = self.reader.vehicle_data( + vehicle_key(selection.row), + [*DATA_ENDPOINTS, CHARGE_SCHEDULE_ENDPOINT, DEPARTURE_ENDPOINT], + ) + except (VehicleAsleep, TimeoutError, HttpError, QueryError, TokenError): + return ParkedPeek("unavailable", None, selection.store, selection.label) + observation = from_vehicle_data(payload) + self._remember(payload, observation, store=selection.store) + return ParkedPeek("online", observation, selection.store, selection.label) + def report(self, *, preferred: str | None) -> VehicleReport: """Read-only briefing. A car that is not online, a timeout, or HTTP 408 is a skip. diff --git a/core/tariff.py b/core/tariff.py new file mode 100644 index 0000000..a12b8a7 --- /dev/null +++ b/core/tariff.py @@ -0,0 +1,199 @@ +"""Home electricity windows stored in the private config home. + +The file is ``$AGENTDRIVE_HOME/tariff.yaml``. It is advice input only. +Nothing here talks to the car or changes a charging timetable. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path + +import yaml + +from core.atomic import write_atomic +from core.errors import ConfigError + +TARIFF_NAME = "tariff.yaml" +_KINDS = {"off-peak", "peak"} + + +@dataclass(frozen=True) +class TariffWindow: + kind: str + start: str + end: str + start_minutes: int + end_minutes: int + price_per_kwh: float | None = None + + def as_dict(self) -> dict: + body = {"kind": self.kind, "start": self.start, "end": self.end} + if self.price_per_kwh is not None: + body["price_per_kwh"] = self.price_per_kwh + return body + + +@dataclass(frozen=True) +class Tariff: + windows: tuple[TariffWindow, ...] = () + + def of_kind(self, kind: str) -> tuple[TariffWindow, ...]: + return tuple(item for item in self.windows if item.kind == kind) + + +def tariff_path(home: Path) -> Path: + return Path(home) / TARIFF_NAME + + +def load_tariff(home: Path) -> Tariff: + path = tariff_path(home) + if not path.is_file(): + return Tariff() + try: + data = yaml.safe_load(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, yaml.YAMLError) as exc: + raise ConfigError("tariff.yaml could not be read.") from exc + if data is None: + return Tariff() + if not isinstance(data, dict): + raise ConfigError("tariff.yaml must be a mapping.") + unknown = set(data) - {"version", "windows"} + if unknown: + raise ConfigError("tariff.yaml contains a field this toolkit will not store.") + windows = tuple(_window(item) for item in data.get("windows") or []) + return Tariff(windows=windows) + + +def save_tariff(home: Path, tariff: Tariff) -> None: + payload = {"version": 1, "windows": [item.as_dict() for item in tariff.windows]} + text = yaml.safe_dump(payload, sort_keys=False, allow_unicode=False) + write_atomic(tariff_path(home), text) + + +def set_window(home: Path, kind: str, window: str, price: float | None) -> Tariff: + parsed = _parse_window(kind, window, price) + current = load_tariff(home) + kept = [ + item + for item in current.windows + if not (item.kind == parsed.kind and item.start == parsed.start and item.end == parsed.end) + ] + updated = Tariff(windows=tuple(kept + [parsed])) + save_tariff(home, updated) + return updated + + +def clear_windows(home: Path, kind: str | None) -> Tariff: + current = load_tariff(home) + if kind is None: + updated = Tariff() + else: + name = _kind(kind) + updated = Tariff(windows=tuple(item for item in current.windows if item.kind != name)) + save_tariff(home, updated) + return updated + + +def format_tariff(tariff: Tariff) -> str: + if not tariff.windows: + return ( + "No home tariff is saved.\n" + "Set one with `agentdrive tariff set off-peak 22:00-06:00` or " + "`agentdrive tariff set peak 16:00-21:00`. " + "An optional `--price` is per kWh. This does not change the car." + ) + lines = ["Home tariff windows saved on this computer. This does not change the car."] + for item in tariff.windows: + price = "" if item.price_per_kwh is None else f" {item.price_per_kwh:.4g} per kWh" + lines.append(f"{item.kind} {item.start}-{item.end}{price}") + return "\n".join(lines) + + +def in_window(minute: int, window: TariffWindow) -> bool: + start = window.start_minutes + end = window.end_minutes + if start < end: + return start <= minute < end + return minute >= start or minute < end + + +def clock_text(minutes: int) -> str: + hour, minute = divmod(int(minutes), 60) + return f"{hour:02d}:{minute:02d}" + + +def _window(item) -> TariffWindow: + if not isinstance(item, dict): + raise ConfigError("A tariff window must be a mapping.") + unknown = set(item) - {"kind", "start", "end", "price_per_kwh"} + if unknown: + raise ConfigError("A tariff window has a field this toolkit will not store.") + kind = _kind(str(item.get("kind") or "")) + start = _clock(str(item.get("start") or "")) + end = _clock(str(item.get("end") or "")) + if start == end: + raise ConfigError("A tariff window needs a start and a different end.") + price = _price(item.get("price_per_kwh")) + return TariffWindow( + kind=kind, + start=_format_clock(start), + end=_format_clock(end), + start_minutes=start, + end_minutes=end, + price_per_kwh=price, + ) + + +def _parse_window(kind: str, window: str, price: float | None) -> TariffWindow: + text = str(window or "").strip() + if "-" not in text: + raise ConfigError("A tariff window is HH:MM-HH:MM.") + start_text, end_text = text.split("-", 1) + start = _clock(start_text) + end = _clock(end_text) + if start == end: + raise ConfigError("A tariff window needs a start and a different end.") + return TariffWindow( + kind=_kind(kind), + start=_format_clock(start), + end=_format_clock(end), + start_minutes=start, + end_minutes=end, + price_per_kwh=_price(price), + ) + + +def _kind(value: str) -> str: + text = str(value or "").strip().lower() + if text not in _KINDS: + raise ConfigError("A tariff window is off-peak or peak.") + return text + + +def _clock(value: str) -> int: + text = str(value or "").strip() + parts = text.split(":") + if len(parts) != 2 or not parts[0].isdigit() or not parts[1].isdigit(): + raise ConfigError("A tariff time is HH:MM in 24-hour time.") + hour = int(parts[0]) + minute = int(parts[1]) + if hour > 23 or minute > 59 or len(parts[0]) > 2 or len(parts[1]) != 2: + raise ConfigError("A tariff time is HH:MM in 24-hour time.") + return hour * 60 + minute + + +def _format_clock(minutes: int) -> str: + return clock_text(minutes) + + +def _price(value) -> float | None: + if value is None or value == "": + return None + try: + number = float(value) + except (TypeError, ValueError) as exc: + raise ConfigError("A tariff price must be a number of currency units per kWh.") from exc + if number != number or number < 0 or number > 100: + raise ConfigError("A tariff price must be from 0 to 100 per kWh.") + return round(number, 6) diff --git a/core/update.py b/core/update.py index 4435f7b..03c373a 100644 --- a/core/update.py +++ b/core/update.py @@ -34,6 +34,7 @@ "history.key", "history.db", ".history.lock", + "tariff.yaml", ) diff --git a/docs/00-before-you-start.md b/docs/00-before-you-start.md index 2addf8b..40d2a3c 100644 --- a/docs/00-before-you-start.md +++ b/docs/00-before-you-start.md @@ -29,6 +29,12 @@ Every other morning send the previous day. That one job is the whole morning bri the usage report, a short tip on some mornings, a service reminder when one is due, and on a Monday weekly briefing one official Tesla note only when the local Tesla site has something new. If it does not, leave that part out. Do not add a second morning job. +When the briefing prints a Chart path, attach that PNG to the weekly, monthly, or yearly message. +The Dashboard path is a local HTML file on this computer. Do not upload the history database. +If I ask whether the battery is enough for a trip, run `agentdrive advice trip `. +If I ask about charging habits or cheap hours, run `agentdrive advice charge`. +Save home off-peak and peak hours with `agentdrive tariff set` when I tell you the times. +Advice does not change the car. A charging change still needs my confirmation. Also schedule a health check that runs agentdrive check --json every 30 minutes. Notify me only when it exits 1. Exit 0 means nothing new, including when the car is asleep and when the same alert is still active. That alert stays quiet until diff --git a/docs/01-install-and-safety.md b/docs/01-install-and-safety.md index f3c147b..c42b8a0 100644 --- a/docs/01-install-and-safety.md +++ b/docs/01-install-and-safety.md @@ -36,7 +36,7 @@ On install, the AI agent schedules `agentdrive check --json` every 30 minutes. T ## After install — daily report -AgentDrive does not run in the background. After `setup/install.sh` finishes, the AI agent reads `agentdrive report-schedule`. The default is 07:30 local time and enabled. The AI agent creates one daily job at that time, runs `agentdrive history due`, and sends that usage briefing. On Monday it is the previous week, on the 1st the previous month, and on 1 January it is the previous year. `agentdrive report` is still the live snapshot when you want the car's current status, including the skip line when the car is not online. A later request to change the time or turn the briefing off is `agentdrive report-schedule set HH:MM`, `disable`, or `enable`. The AI agent then updates its job. The command only writes private config under `$AGENTDRIVE_HOME`. It does not edit the AI agent's routine and it does not install cron. +AgentDrive does not run in the background. After `setup/install.sh` finishes, the AI agent reads `agentdrive report-schedule`. The default is 07:30 local time and enabled. The AI agent creates one daily job at that time, runs `agentdrive history due`, and sends that usage briefing. On Monday it is the previous week, on the 1st the previous month, and on 1 January it is the previous year. A weekly, monthly, or yearly briefing includes a summary and a chart path. Attach that PNG to the message you send. `agentdrive report` is still the live snapshot when you want the car's current status, including the skip line when the car is not online. A later request to change the time or turn the briefing off is `agentdrive report-schedule set HH:MM`, `disable`, or `enable`. The AI agent then updates its job. The command only writes private config under `$AGENTDRIVE_HOME`. It does not edit the AI agent's routine and it does not install cron. ## Update an existing install diff --git a/docs/06-queries-and-commands.md b/docs/06-queries-and-commands.md index 67e55fa..8655d8a 100644 --- a/docs/06-queries-and-commands.md +++ b/docs/06-queries-and-commands.md @@ -36,9 +36,17 @@ A location read needs the official `vehicle_location` scope. The owner sign-in s `agentdrive history` reads the encrypted history file in the config home. `history due` is the morning usage briefing. On 1 January it covers the previous calendar year. On any other 1st it covers the previous calendar month. On Monday it covers the previous seven days, Monday through Sunday. Every other morning it covers the previous local day. Only one of those is sent. `history daily`, `weekly`, `monthly`, and `yearly` print that period when you ask. `history trips`, `history charges`, and `history snapshots` list the saved rows. `--json` prints the same facts as an object. These commands do not call the car. +A weekly, monthly, or yearly briefing adds a summary: kilometres driven, number of trips, average kilometres per day, number of charges, total kWh added, average per charge, and a comparison with the previous period. It says when coverage is incomplete. History is saved only while the car was already online, so it is opportunistic, not a continuous record. The briefing writes a PNG chart under the car's directory in `$AGENTDRIVE_HOME` and prints that path. It also writes `briefings/dashboard.html`, a single file with week, month, and year views that opens offline. If matplotlib is installed it draws the PNG. Otherwise AgentDrive draws the PNG itself. No extra package is required. Attach the PNG when you send the briefing. Do not upload the history database. + +`agentdrive advice trip ` says whether the battery covers that distance. It uses a non-waking reading when the car is already online, and otherwise the last saved battery. It does not wake the car. Consumption is the battery percent per kilometre from the car's own saved trips. Fewer than three usable trips is reported as too little data, and no guess is filled in. A cold or hot saved temperature adjusts the estimate. The same command suggests an everyday charge limit, usually 80%, and a higher limit before a long trip. It does not change the limit. When a departure schedule is set and the reading is cold or hot, it suggests preconditioning and does not start it. + +`agentdrive advice charge` flags three habits when the saved charges show them: often charging to 100% and leaving it, often starting a charge very low, and often charging in a peak window you saved. `agentdrive tariff set off-peak 22:00-06:00` and `agentdrive tariff set peak 16:00-21:00` store those hours in `$AGENTDRIVE_HOME/tariff.yaml`. `--price` is optional, per kWh. The advice can suggest moving the in-car timetable into the cheap window. It does not change the timetable. Any change still needs the usual confirmation. + +With more than one saved car, pass `--car `. Advice, the chart, and the summary use that car's history. + A parked snapshot is saved when `check` or `report` already has a reading and the car is online. A check that finds nothing new to notify still saves that snapshot. Repeated checks that find the same parked condition update that snapshot's last time, sample count, battery, and tyre readings. A new snapshot starts when the car leaves Park, charging changes, the lock, sentry, doors, windows, or trunk change, a new alert appears, or the odometer moves. Charge readings are kept with the charge session, about once a minute when a reading was seen. The database is encrypted with AES-256-GCM. The key is `$AGENTDRIVE_HOME/history.key`, created on first save, mode 600 on macOS and Linux. The Python dependency is `cryptography`. SQLCipher is not required. The key is not in the git repo and must not be pasted into chat. -What is saved: trip summaries (time, distance from the odometer change, battery, lock, gear, and sentry), charge sessions (time, start and end percent, energy added when the car sent it, and whether it was plugged in), those charge readings, parked snapshots, alerts when they appear, and software or unit settings when they change. A service you mention, such as a tyre rotation, is stored against the last saved odometer. How full the charge curve looks depends on how often the usual check runs while the car is already online or charging. AgentDrive does not run in the background to fill the gaps. +What is saved: trip summaries (time, distance from the odometer change, battery, lock, gear, and sentry), charge sessions (time, start and end percent, energy added during that plug-in, and whether it was plugged in), those charge readings, parked snapshots, alerts when they appear, and software or unit settings when they change. Energy added is the increase during the plug-in. A counter that still holds the previous session, or that keeps climbing across plugs, is not stored as this session's total. Older rows are left as they were. A summary ignores a saved energy figure that cannot match the battery change. A service you mention, such as a tyre rotation, is stored against the last saved odometer. How full the charge curve looks depends on how often the usual check runs while the car is already online or charging. AgentDrive does not run in the background to fill the gaps. `agentdrive news` prints one new official item from `https://www.tesla.com` for the profile country, or for the computer's timezone when the profile has no country. The Monday weekly briefing includes that same item: a short summary and the official link. A safety notice is preferred. The same item is not shown again for two calendar months. When there is nothing new, the command prints nothing and the briefing leaves that part out. It does not wake the car. @@ -77,7 +85,9 @@ Commands: | `frunk-open` | signed `frunk-open` | yes, one time. The sentence names open frunk | | `remote-start` | signed `drive` | yes, one time. The sentence names enable remote start | | `history trips`, `history charges`, `history snapshots` | no signed command; saved history only | no, and it does not contact the car | -| `history due`, `history daily`, `history weekly`, `history monthly`, `history yearly` | no signed command; saved history only | no | +| `history due`, `history daily`, `history weekly`, `history monthly`, `history yearly` | no signed command; saved history only. Week, month, and year also write a local chart | no | +| `advice trip `, `advice charge` | no signed command; saved history, plus a non-waking read when the car is already online | no, and it does not wake the car or change a setting | +| `tariff show`, `tariff set`, `tariff clear` | no signed command; writes `$AGENTDRIVE_HOME/tariff.yaml` | no, and it does not change the car | | `tips` | no signed command | no, and it does not contact the car | | `news` | no signed command; reads the official Tesla site for the owner's country | no, and it does not wake the car | | `service`, `service log rotation`, `service log replacement`, `service log maintenance` | no signed command; uses the saved odometer | no, and it does not wake the car | diff --git a/docs/07-assistants.md b/docs/07-assistants.md index 05454b0..092213c 100644 --- a/docs/07-assistants.md +++ b/docs/07-assistants.md @@ -40,7 +40,7 @@ The daily report is a second job, separate from the health check. The health che The default is 07:30 in the local timezone, and the report is enabled. `agentdrive report-schedule` prints that. The time is stored in `$AGENTDRIVE_HOME/runtime.yaml` under `report_schedule` when you change it. With no `report_schedule` key, the default above is what the command shows. -After install, the AI agent creates one daily job at the configured time. The job runs `agentdrive history due`, and the AI agent sends that usage briefing. On Monday it covers the previous week, on the 1st the previous month, and on 1 January the previous year. Every other morning it covers the previous day. The command reads saved history only. It does not wake the car. `agentdrive report` remains the live parked snapshot when the user wants the car's current status, including the skip line when the vehicle is not online. +After install, the AI agent creates one daily job at the configured time. The job runs `agentdrive history due`, and the AI agent sends that usage briefing. On Monday it covers the previous week, on the 1st the previous month, and on 1 January the previous year. Every other morning it covers the previous day. The command reads saved history only. It does not wake the car. When the briefing prints a Chart path, attach that PNG. The Dashboard path is a local HTML file. `agentdrive report` remains the live parked snapshot when the user wants the car's current status, including the skip line when the vehicle is not online. When you ask for a different time, the AI agent runs `agentdrive report-schedule set HH:MM` and moves the job to that time. When you ask to turn the report off, the AI agent runs `agentdrive report-schedule disable` and removes the job. `agentdrive report-schedule enable` turns it back on. AgentDrive does not update the job itself. The same steps apply to Grok Bot, Muse, and any shell-capable AI agent. diff --git a/docs/08-skills.md b/docs/08-skills.md index dc1995a..82d54eb 100644 --- a/docs/08-skills.md +++ b/docs/08-skills.md @@ -10,7 +10,8 @@ Phase 1 answers a question you just asked, or sends a parked command you just as - **Flash lights** requires Park and speed 0, sends one signed command, and reads the car again. - **Climate on or off** uses the same park rule and checks the climate flag afterward. Seat heat, steering-wheel heat, defrost, and preconditioning the cabin now use that same park rule. Turning sentry off needs a fresh confirmation. Sentry on does not. - **`agentdrive check`** is the health check. It does not wake a sleeping car. On install, schedule `agentdrive check --json` every 30 minutes. With several cars, run that check once per car with `--car `. Notify only when it exits 1. The same alert exits 0 until it clears and later returns. An alert is remembered per car. See [AI agents](07-assistants.md). -- **`agentdrive report`** is the live parked snapshot. **`agentdrive history due`** is the morning usage briefing. On install, create one daily job at the time in `agentdrive report-schedule` (default 07:30 local time, enabled). Run `agentdrive history due` and send that briefing. On Monday it is the previous week, on the 1st the previous month, and on 1 January the previous year. Run `agentdrive report` when you want the car's current status, including the skip line. AgentDrive does not run that job itself. It is separate from the health check. +- **`agentdrive report`** is the live parked snapshot. **`agentdrive history due`** is the morning usage briefing. On install, create one daily job at the time in `agentdrive report-schedule` (default 07:30 local time, enabled). Run `agentdrive history due` and send that briefing. On Monday it is the previous week, on the 1st the previous month, and on 1 January the previous year. When that briefing prints a Chart path, attach the PNG. The Dashboard path is a local HTML file. Run `agentdrive report` when you want the car's current status, including the skip line. AgentDrive does not run that job itself. It is separate from the health check. +- **`agentdrive advice trip `** says whether the saved driving history and the current battery cover a trip. **`agentdrive advice charge`** flags charging habits and compares them with a home tariff. **`agentdrive tariff set`** stores off-peak and peak hours. These are advice only. They do not wake a sleeping car and they do not change a charge limit, a timetable, or preconditioning. A later change still needs a confirmation. - **`agentdrive where`** is a coarse area: suburb or locality, state or region, and country. It does not wake a sleeping car. A rounded point is sent to Nominatim. `where --precise` prints latitude, longitude, and a map link when the car is already online, and it still does not wake the car. Precise coordinates are not saved in the local history. It is not part of the recurring health check. - **`agentdrive charge-schedule show`**, **`set`**, and **`remove`** read and change the in-car charging timetable. `show` does not wake a sleeping car. `set` and `remove` need Park, speed 0, and a fresh confirmation. A scheduled job cannot confirm them. - **`agentdrive charge-amps`** sets the charging current. It needs Park, speed 0, and a fresh confirmation. The car may use less than the amps you ask for. A scheduled job cannot run it. diff --git a/tests/test_advice.py b/tests/test_advice.py new file mode 100644 index 0000000..3dd90d3 --- /dev/null +++ b/tests/test_advice.py @@ -0,0 +1,438 @@ +"""Trip advice, charging advice, briefing summaries, and charge-energy correction.""" + +import json +from datetime import date, timedelta +from pathlib import Path +from types import SimpleNamespace + +from core.advice import charge_advice, parse_kilometers, trip_advice +from core.cli import main +from core.errors import ConfigError +from core.family import add_member +from core.garage import load_garage, remember_car, store_for +from core.history.charts import write_builtin_png +from core.history.crypto import database_paths +from core.history.energy import plausible_energy, resolve_charge_energy +from core.history.store import history_session +from core.history.usage import render_briefing, render_list +from core.ops import App +from core.safety.catalog import schedule_blocks +from core.tariff import Tariff, TariffWindow, load_tariff +from tests.test_cars_and_family import VIN_A, VIN_B, row +from tests.test_history import UTC, at, record + +import pytest + + +def _trip(day, km=20.0, start=80, end=70): + stamp = f"2026-03-{day:02d}T08:00:00Z" + return { + "open": False, + "started_at": stamp, + "ended_at": stamp, + "distance_km": km, + "start_battery_percent": start, + "end_battery_percent": end, + } + + +def _charge(day, hour, start, end, energy, readings=None): + started = f"2026-03-{day:02d}T{hour:02d}:00:00Z" + return { + "open": False, + "started_at": started, + "ended_at": f"2026-03-{day:02d}T{hour + 2:02d}:00:00Z", + "start_percent": start, + "end_percent": end, + "energy_added_kwh": energy, + "readings": readings or [], + } + + +def _peek(temp=18, battery=60, departures=(), schedules=(), departure_section=True, charge_section=True): + observation = SimpleNamespace( + battery_level=battery, + outside_temp=temp, + departure_schedules=departures, + departure_schedule_section_present=departure_section, + departure_schedule_list_present=True, + charge_schedules=schedules, + charge_schedule_section_present=charge_section, + charge_schedule_list_present=True, + ) + return SimpleNamespace(state="online", observation=observation, store=None, label="") + + +def _window(kind, start, end, start_minutes, end_minutes, price=None): + return TariffWindow(kind, start, end, start_minutes, end_minutes, price) + + +def test_resolve_keeps_a_normal_session_and_drops_a_stale_or_cumulative_counter(): + assert resolve_charge_energy([0.2, 0.5, 5.0], 30, 40) == pytest.approx(5.0) + assert resolve_charge_energy([54.2, 0.2, 0.8, 1.2], 70, 72) == pytest.approx(1.2) + assert resolve_charge_energy([54.2], 70, 72) is None + assert resolve_charge_energy([17.1, 25.0, 35.8], 61, 77) == pytest.approx(18.7) + assert plausible_energy(17.1, 37, 51) is True + assert plausible_energy(35.8, 61, 77) is False + assert plausible_energy(54.2, 70, 72) is False + + +def test_too_few_trips_are_not_guessed(): + payload = trip_advice( + kilometers=40, + trips=[_trip(2), _trip(3)], + parked={"battery_percent": 70, "outside_temp_c": 15, "last_seen": "2026-03-04T08:00:00Z"}, + peek=None, + tariff=Tariff(), + today=date(2026, 3, 5), + tz=UTC, + ) + assert payload["consumption_percent_per_km"] is None + assert payload["enough"] is None + assert "No guess was made" in payload["text"] + assert "2 usable trips" in payload["text"] + assert "% of the battery per km" not in payload["text"] + assert "was not woken" in payload["text"] + + +def test_trip_sufficiency_charge_limit_and_cold_preconditioning(): + trips = [_trip(2), _trip(3), _trip(4)] + mild = trip_advice( + kilometers=20, + trips=trips, + parked=None, + peek=_peek(temp=18, battery=50), + tariff=Tariff(), + today=date(2026, 3, 6), + tz=UTC, + ) + assert mild["consumption_percent_per_km"] == pytest.approx(0.5) + assert mild["enough"] is True + assert mild["needed_percent"] == pytest.approx(20.0) + assert mild["charge_limit_percent"] == 80 + assert mild["weather_factor"] == pytest.approx(1.0) + assert "is enough" in mild["text"] + assert "does not change the charge limit" in mild["text"] + assert "non-waking reading" in mild["text"] + + cold = trip_advice( + kilometers=20, + trips=trips, + parked=None, + peek=_peek(temp=-2, battery=50, departures=(SimpleNamespace(enabled=True),)), + tariff=Tariff(), + today=date(2026, 3, 6), + tz=UTC, + ) + assert cold["weather_factor"] == pytest.approx(1.2) + assert cold["needed_percent"] == pytest.approx(22.0) + assert cold["precondition"] is True + assert "cold" in cold["text"] + assert "advice only" in cold["text"] + assert "does not start preconditioning" in cold["text"] + + short = trip_advice( + kilometers=20, + trips=trips, + parked=None, + peek=_peek(temp=18, battery=15), + tariff=Tariff(), + today=date(2026, 3, 6), + tz=UTC, + ) + assert short["enough"] is False + assert "not enough" in short["text"] + + long_trip = trip_advice( + kilometers=300, + trips=trips, + parked=None, + peek=_peek(temp=18, battery=90, departures=()), + tariff=Tariff(), + today=date(2026, 3, 6), + tz=UTC, + ) + assert long_trip["trip_limit_percent"] == 100 + assert "100%" in long_trip["text"] + assert "No departure schedule is set" in long_trip["text"] + + +def test_charge_habits_tariff_and_schedule_suggestion_do_not_print_coordinates(): + charges = [] + for day, hour, start in ((2, 18, 10), (3, 18, 12), (4, 18, 8), (5, 7, 40)): + charges.append(_charge(day, hour, start, 100, 30)) + marks = [ + {"last_seen": "2026-03-02T22:00:00Z", "battery_percent": 100, "observed_at": "2026-03-02T20:00:00Z"}, + {"last_seen": "2026-03-03T22:00:00Z", "battery_percent": 99, "observed_at": "2026-03-03T20:00:00Z"}, + {"last_seen": "2026-03-04T22:00:00Z", "battery_percent": 98, "observed_at": "2026-03-04T20:00:00Z"}, + ] + tariff = Tariff( + windows=( + _window("off-peak", "22:00", "06:00", 22 * 60, 6 * 60, 0.18), + _window("peak", "16:00", "21:00", 16 * 60, 21 * 60, 0.42), + ) + ) + schedule = SimpleNamespace( + enabled=True, + start_enabled=True, + start_time=18 * 60, + latitude=55.1234, + longitude=12.5678, + ) + payload = charge_advice( + trips=[_trip(2), _trip(3), _trip(4)], + charges=charges, + marks=marks, + parked=None, + peek=_peek(temp=18, battery=60, schedules=(schedule,)), + tariff=tariff, + today=date(2026, 3, 6), + tz=UTC, + ) + assert "full_and_left" in payload["habits"] + assert "low_before_charge" in payload["habits"] + assert "peak_window" in payload["habits"] + assert "Often charging to 100% and leaving it" in payload["text"] + assert "Often dropping very low before charging" in payload["text"] + assert "Often charging in the peak window" in payload["text"] + assert "22:00-06:00" in payload["text"] + assert "outside the cheap window" in payload["text"] + assert "needs a confirmation" in payload["text"] + assert "does not change" in payload["text"] + assert "55.1234" not in payload["text"] + assert "12.5678" not in payload["text"] + assert payload["advice_only"] is True + assert payload["wakes_car"] is False + + +def test_tariff_commands_stay_in_the_config_home(tmp_path, capsys): + assert main(["--home", str(tmp_path), "tariff", "set", "off-peak", "22:00-06:00", "--price", "0.18"]) == 0 + assert main(["--home", str(tmp_path), "tariff", "set", "peak", "16:00-21:00", "--price", "0.42"]) == 0 + saved = load_tariff(tmp_path) + assert [item.kind for item in saved.windows] == ["off-peak", "peak"] + assert (tmp_path / "tariff.yaml").is_file() + capsys.readouterr() + assert main(["--home", str(tmp_path), "tariff", "show", "--json"]) == 0 + body = json.loads(capsys.readouterr().out) + assert body["windows"][0]["price_per_kwh"] == pytest.approx(0.18) + assert main(["--home", str(tmp_path), "tariff", "clear", "peak"]) == 0 + assert [item.kind for item in load_tariff(tmp_path).windows] == ["off-peak"] + with pytest.raises(ConfigError): + parse_kilometers("0") + add_member(tmp_path, "rider") + assert main(["--home", str(tmp_path), "--as", "rider", "tariff", "show"]) == 2 + assert "owner" in capsys.readouterr().err + + +def test_advice_does_not_wake_and_uses_the_selected_car(tmp_path, monkeypatch, capsys): + from core.config import load_regions + from core.fleet.queries import FleetReader + + class Http: + def __init__(self, state): + self.state = state + self.calls = [] + + def request_json(self, method, url, headers, timeout=30, body=None): + self.calls.append(url) + if url.rstrip("/").endswith("/vehicles"): + return 200, {"response": [{"vin": VIN_A, "state": self.state, "display_name": "Example"}]} + if "wake_up" in url: + raise AssertionError("advice must not wake the car") + return 200, {"response": {}} + + http = Http("asleep") + regions = load_regions() + tokens = SimpleNamespace(access_token=lambda force=False: "token") + app = App( + tmp_path, + reader=FleetReader(http, regions, "na", tokens), + tokens=tokens, + control=SimpleNamespace(run=lambda **kwargs: (_ for _ in ()).throw(AssertionError("no command"))), + region="na", + history=True, + ) + peek = app.peek(preferred=None) + assert peek.state == "asleep" + assert peek.observation is None + assert not any("wake_up" in url for url in http.calls) + assert not any("vehicle_data" in url for url in http.calls) + + rows = [row(VIN_A), row(VIN_B)] + remember_car(tmp_path, rows, "0001", "daily") + remember_car(tmp_path, rows, "0002", "weekend") + garage = load_garage(tmp_path) + weekend = store_for(tmp_path, garage, garage.by_nickname("weekend")) + assert weekend != tmp_path + for day in (2, 3, 4): + record( + weekend, + at(2026, 3, day, 8), + vehicle_state={"odometer": 100 + day}, + charge_state={"battery_level": 80, "charging_state": "Disconnected"}, + ) + record( + weekend, + at(2026, 3, day, 9), + vehicle_state={"odometer": 110 + day}, + charge_state={"battery_level": 70, "charging_state": "Disconnected"}, + ) + monkeypatch.setattr("core.cli._peek_parked", lambda home, args: None) + monkeypatch.setattr("core.history.usage.local_today", lambda tz: date(2026, 3, 6)) + assert main(["--home", str(tmp_path), "advice", "--car", "weekend", "trip", "20"]) == 0 + text = capsys.readouterr().out + assert "is enough" in text or "% of the battery per km" in text + assert main(["--home", str(tmp_path), "advice", "--car", "daily", "trip", "20"]) == 0 + other = capsys.readouterr().out + assert "No guess was made" in other + assert schedule_blocks("advice trip 20") is False + assert schedule_blocks("advice charge") is False + assert schedule_blocks("tariff set off-peak 22:00-06:00") is False + assert schedule_blocks("charge-schedule set weekdays 22:00-06:00 1 2") is True + + +def test_recorded_charge_energy_uses_the_session_increase(tmp_path): + home = tmp_path / "home" + start = at(2026, 3, 2, 22) + series = [ + (start, 70, 54.2), + (start + timedelta(minutes=2), 70, 0.2), + (start + timedelta(minutes=4), 71, 0.8), + (start + timedelta(minutes=6), 72, 1.2), + ] + for moment, battery, energy in series: + record( + home, + moment, + charge_state={ + "battery_level": battery, + "charging_state": "Charging", + "charge_energy_added": energy, + "charger_power": 7, + }, + vehicle_state={"odometer": 100}, + ) + record( + home, + start + timedelta(minutes=8), + charge_state={"battery_level": 72, "charging_state": "Disconnected", "charge_energy_added": 0}, + vehicle_state={"odometer": 100}, + ) + charge = json.loads(render_list(home, "charges", limit=5, as_json=True))["charges"][0] + assert charge["energy_added_kwh"] == pytest.approx(1.2) + assert charge["start_percent"] == 70 + assert charge["end_percent"] == 72 + + home_b = tmp_path / "cumulative" + begin = at(2026, 3, 4, 1) + for offset, battery, energy in ((0, 61, 17.1), (2, 68, 25.0), (4, 77, 35.8)): + record( + home_b, + begin + timedelta(minutes=offset), + charge_state={ + "battery_level": battery, + "charging_state": "Charging", + "charge_energy_added": energy, + "charger_power": 7, + }, + vehicle_state={"odometer": 80}, + ) + record( + home_b, + begin + timedelta(minutes=6), + charge_state={"battery_level": 77, "charging_state": "Disconnected", "charge_energy_added": 0}, + vehicle_state={"odometer": 80}, + ) + cumulative = json.loads(render_list(home_b, "charges", limit=5, as_json=True))["charges"][0] + assert cumulative["energy_added_kwh"] == pytest.approx(18.7) + + +def test_legacy_bad_energy_is_left_in_place_and_kept_out_of_the_summary(tmp_path): + home = tmp_path / "home" + record(home, at(2026, 3, 2, 8), vehicle_state={"odometer": 10}) + with history_session(home, write=True) as conn: + conn.execute( + """ + INSERT INTO charges( + started_at, ended_at, start_percent, end_percent, energy_added_kwh, plugged_in, charging, open + ) VALUES + ('2026-03-03T01:00:00Z', '2026-03-03T02:00:00Z', 70, 72, 54.2, 1, 1, 0), + ('2026-03-03T03:00:00Z', '2026-03-03T04:00:00Z', 37, 51, 17.1, 1, 1, 0), + ('2026-03-03T05:00:00Z', '2026-03-03T06:00:00Z', 61, 77, 35.8, 1, 1, 0) + """ + ) + conn.execute( + """ + INSERT INTO charge_samples(charge_id, observed_at, battery_percent, energy_added_kwh, charger_power_kw) + VALUES + (3, '2026-03-03T05:00:00Z', 61, 17.1, 7), + (3, '2026-03-03T06:00:00Z', 77, 35.8, 7) + """ + ) + briefing = json.loads(render_briefing(home, "weekly", as_json=True, on=date(2026, 3, 9), tz=UTC)) + with history_session(home, write=False) as conn: + stored = [row[0] for row in conn.execute("SELECT energy_added_kwh FROM charges ORDER BY id")] + assert stored == pytest.approx([54.2, 17.1, 35.8]) + assert briefing["summary"]["energy_added_kwh"] == pytest.approx(35.8) + assert briefing["summary"]["energy_omitted"] == 1 + assert briefing["summary"]["energy_replaced"] == 1 + assert "does not match the battery change" in briefing["text"] + assert "Those saved rows were not changed" in briefing["text"] + listed = render_list(home, "charges", limit=5, as_json=False) + assert "not counted" in listed + assert "18.7 kWh counted" in listed + + +def test_weekly_summary_compares_the_previous_period_and_writes_a_chart(tmp_path): + home = tmp_path / "home" + record( + home, + at(2026, 3, 3, 8), + vehicle_state={"odometer": 100}, + charge_state={"battery_level": 80, "charging_state": "Disconnected"}, + ) + record( + home, + at(2026, 3, 3, 10), + vehicle_state={"odometer": 120}, + charge_state={"battery_level": 60, "charging_state": "Disconnected"}, + ) + record( + home, + at(2026, 2, 24, 8), + vehicle_state={"odometer": 50}, + charge_state={"battery_level": 70, "charging_state": "Disconnected"}, + ) + record( + home, + at(2026, 2, 24, 9), + vehicle_state={"odometer": 60}, + charge_state={"battery_level": 60, "charging_state": "Disconnected"}, + ) + text = render_briefing(home, "weekly", as_json=False, on=date(2026, 3, 9), tz=UTC) + body = json.loads(render_briefing(home, "weekly", as_json=True, on=date(2026, 3, 9), tz=UTC)) + assert "Summary:" in text + assert "Compared with the previous week" in text + assert "Coverage is incomplete" in text + assert "History is opportunistic, not a continuous record." in text + assert body["summary"]["trip_count"] == 1 + assert body["summary"]["previous"]["trip_count"] == 1 + assert body["summary"]["coverage"]["incomplete"] is True + chart = body["summary"]["chart"] + dashboard = body["summary"]["dashboard"] + assert chart.endswith(".png") + assert "Chart:" in text + assert "Dashboard:" in text + png = Path(chart).read_bytes() + assert png.startswith(b"\x89PNG\r\n\x1a\n") + page = Path(dashboard).read_text(encoding="utf-8") + assert "Week" in page and "Month" in page and "Year" in page + assert "http://" not in page and "https://" not in page + assert body["summary"]["chart_renderer"] == "builtin" + fallback = tmp_path / "fallback.png" + write_builtin_png(fallback, ["01"], [12.0], ["01"], [4.0]) + assert fallback.read_bytes().startswith(b"\x89PNG\r\n\x1a\n") + yearly = render_briefing(home, "yearly", as_json=False, on=date(2026, 1, 1), tz=UTC) + assert "Compared with the previous year" in yearly + monthly = render_briefing(home, "monthly", as_json=False, on=date(2026, 3, 1), tz=UTC) + assert "Compared with the previous month" in monthly