diff --git a/.cspell/custom-dictionary-workspace.txt b/.cspell/custom-dictionary-workspace.txt index ef4ded9fb..ad3a2510c 100644 --- a/.cspell/custom-dictionary-workspace.txt +++ b/.cspell/custom-dictionary-workspace.txt @@ -85,6 +85,7 @@ dayname daynumber daysymbol dedup +dedupe dend denorm derating @@ -306,6 +307,8 @@ predai predbat predbatt predheat +preseed +preseeded prevs psum pvbat @@ -395,6 +398,7 @@ TCWORLD tdata tdiff temperation +teslemetry Teslemetry testname thirdparty diff --git a/apps/predbat/components.py b/apps/predbat/components.py index 4951f79c5..12856c6f6 100644 --- a/apps/predbat/components.py +++ b/apps/predbat/components.py @@ -27,6 +27,7 @@ from axle import AxleAPI from solax import SolaxAPI from sigenergy import SigenergyAPI +from teslemetry import TeslemetryAPI from solis import SolisAPI from alertfeed import AlertFeed from web import WebInterface @@ -366,6 +367,17 @@ "phase": 1, "can_restart": True, }, + "teslemetry": { + "class": TeslemetryAPI, + "name": "Tesla Powerwall (Teslemetry)", + "event_filter": "predbat_teslemetry_", + "args": { + "key": {"required": True, "config": "teslemetry_key"}, + "site_id": {"required": True, "config": "teslemetry_site_id"}, + "base_url": {"required": False, "config": "teslemetry_base_url", "default": "https://api.teslemetry.com"}, + }, + "phase": 1, + }, "solax": { "class": SolaxAPI, "name": "SolaX Cloud API", diff --git a/apps/predbat/config.py b/apps/predbat/config.py index ec826f9c5..a8283129b 100644 --- a/apps/predbat/config.py +++ b/apps/predbat/config.py @@ -2200,6 +2200,9 @@ "fox_auth_method": {"type": "string", "empty": False}, "fox_token_expires_at": {"type": "string", "empty": False}, "fox_token_hash": {"type": "string", "empty": False}, + "teslemetry_key": {"type": "string", "empty": False}, + "teslemetry_site_id": {"type": "string", "empty": False}, + "teslemetry_base_url": {"type": "string", "empty": False}, "octopus_intelligent_slot": {"type": "sensor|sensor_list", "sensor_type": "boolean|action", "entries": "num_cars", "optional_entries": True}, "octopus_ready_time": {"type": "sensor|sensor_list", "sensor_type": "string", "entries": "num_cars", "optional_entries": True}, "octopus_charge_limit": {"type": "sensor|sensor_list", "sensor_type": "float", "entries": "num_cars", "optional_entries": True}, diff --git a/apps/predbat/teslemetry.py b/apps/predbat/teslemetry.py new file mode 100644 index 000000000..81d29923e --- /dev/null +++ b/apps/predbat/teslemetry.py @@ -0,0 +1,588 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Tesla Powerwall integration via the Teslemetry API. + +Teslemetry (https://api.teslemetry.com) proxies the Tesla Fleet API using the +same URI paths, authenticated with a static customer-supplied token (BYOK). +The base URL is configurable so a direct Fleet API connection can be swapped +in later without changing this component. + +Data path: polls live_status (power flows + SOC), site_info (capacity) and +calendar_history (daily energy) and publishes predbat_teslemetry_* sensors. +Control path: exposes virtual select/number/switch entities driven by the +TESLA service hooks; handlers issue REST commands (operation mode, backup +reserve, grid import/export rules and the export tariff-trick). + +Command dedupe: the TESLA service hooks carry `repeat: true`, so Predbat fires +each hook's loopback event every cycle (~5 minutes) even when nothing has +changed. Every control write is routed through `_apply_command`, which only +POSTs when the value actually differs from the last CONFIRMED-SENT value, +cutting Teslemetry command-credit spend without losing failure-retry (a +failed POST never updates the cache) or external-drift correction (see +`fetch_site_info`). +""" + +import asyncio +import json +from datetime import datetime, timezone + +import aiohttp + +from component_base import ComponentBase + +TESLEMETRY_DEFAULT_URL = "https://api.teslemetry.com" +TESLEMETRY_TIMEOUT = 30 +TESLEMETRY_RETRIES = 3 +LIVE_POLL_SECONDS = 120 +ENERGY_POLL_SECONDS = 300 + +OPERATION_MODES = ["self_consumption", "autonomous", "backup"] +EXPORT_RULES = ["never", "pv_only", "battery_ok"] +TARIFF_MODES = ["normal", "export_now"] + + +class TeslemetryAPI(ComponentBase): + """Tesla Powerwall component using the Teslemetry (or Fleet) REST API.""" + + EXPORT_SELL_RATE = 0.50 # GBP/kWh synthetic high sell price to force export now. + DEFAULT_IMPORT_RATE = 0.28 + DEFAULT_EXPORT_RATE = 0.15 + + def initialize(self, key="", site_id="", base_url=TESLEMETRY_DEFAULT_URL, **kwargs): + """Initialise the Teslemetry component from configuration. + + Args: + key: Teslemetry (or Fleet API) bearer token. + site_id: Tesla energy site id to poll and control. + base_url: REST API base URL (Teslemetry by default, swappable for a direct Fleet API connection). + kwargs: Reserved for future control-path configuration (accepted and ignored here). + """ + self.api_key = key + self.site_id = str(site_id) if site_id else "" + self.base_url = base_url + self.api_auth_failed = False + self.last_live_poll = 0 + self.last_energy_poll = 0 + self.site_info_done = False + self.reconcile_done = False + self._last_sent = {} + self.log("Info: TeslemetryAPI initialising site_id={}".format(self.site_id)) + self.log("Info: Teslemetry control drift-correction is transition-based (self-heals when Predbat's own desired value changes, plus a one-off refresh at boot) - full periodic device-state reconciliation is a pilot follow-up") + self.register_control_entities() + + def entity(self, suffix, domain="sensor"): + """Build a prefixed virtual entity id for this component.""" + return "{}.{}_teslemetry_{}".format(domain, self.prefix, suffix) + + async def _request(self, method, path, json_body=None): + """Make a Teslemetry REST request, returning parsed JSON or None on failure.""" + url = "{}{}".format(self.base_url, path) + headers = {"Authorization": "Bearer {}".format(self.api_key), "Content-Type": "application/json"} + timeout = aiohttp.ClientTimeout(total=TESLEMETRY_TIMEOUT) + for attempt in range(TESLEMETRY_RETRIES): + try: + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.request(method, url, headers=headers, json=json_body) as resp: + if resp.status in (401, 403): + self.api_auth_failed = True + self.log("Warn: Teslemetry auth failed ({}) on {} - token revoked or subscription lapsed".format(resp.status, path)) + return None + if resp.status == 429: + self.log("Warn: Teslemetry rate limited on {} attempt {}".format(path, attempt + 1)) + await asyncio.sleep(2 ** (attempt + 1)) + continue + if resp.status >= 400: + body = await resp.text() + self.log("Warn: Teslemetry HTTP {} on {}: {}".format(resp.status, path, body[:200])) + return None + self.api_auth_failed = False + return await resp.json() + except (aiohttp.ClientError, asyncio.TimeoutError) as err: + self.log("Warn: Teslemetry network error on {} attempt {}: {}".format(path, attempt + 1, err)) + await asyncio.sleep(2**attempt) + return None + + def publish_sensor(self, suffix, state, unit=None, state_class="measurement", friendly=None): + """Publish one virtual sensor via the dashboard.""" + attributes = {"friendly_name": friendly or suffix.replace("_", " ").title(), "state_class": state_class} + if unit: + attributes["unit_of_measurement"] = unit + self.dashboard_item(self.entity(suffix), state, attributes, app="teslemetry") + + async def fetch_live_status(self): + """Fetch live power flows and SOC, publishing power sensors.""" + data = await self._request("GET", "/api/1/energy_sites/{}/live_status".format(self.site_id)) + if not data: + return False + response = data.get("response", {}) + self.publish_sensor("soc", response.get("percentage_charged", 0), unit="%", friendly="Powerwall SOC") + self.publish_sensor("battery_power", response.get("battery_power", 0), unit="W", friendly="Powerwall Battery Power") + self.publish_sensor("grid_power", response.get("grid_power", 0), unit="W", friendly="Powerwall Grid Power") + self.publish_sensor("load_power", response.get("load_power", 0), unit="W", friendly="Powerwall Load Power") + self.publish_sensor("solar_power", response.get("solar_power", 0), unit="W", friendly="Powerwall Solar Power") + self.update_success_timestamp() + return True + + async def fetch_site_info(self): + """Fetch site info, publishing the battery capacity as soc_max and seeding control entity states. + + Returns True only once soc_max has actually been published. If nameplate_energy is absent or + zero, control-entity seeding still happens with whatever fields are present, but this returns + False so the run()-level site_info_done latch stays False and the fetch is retried on a later + cycle instead of permanently giving up on soc_max. + """ + data = await self._request("GET", "/api/1/energy_sites/{}/site_info".format(self.site_id)) + if not data: + return False + response = data.get("response", {}) + nameplate_wh = response.get("nameplate_energy", 0) + soc_max_published = False + if nameplate_wh: + self.publish_sensor("soc_max", round(nameplate_wh / 1000.0, 2), unit="kWh", state_class=None, friendly="Powerwall Capacity") + soc_max_published = True + # Seed the control entity STATES (display only, no commands) from the device so they reflect + # reality at boot instead of the hardcoded defaults set by register_control_entities(). + # + # Known limitation (accepted, not an oversight): fetch_site_info is only ever driven to + # completion ONCE per process - run()'s site_info_done latch stops calling it again once it + # returns True - so the operation_mode/backup_reserve drift-refresh below corrects drift at + # BOOT only. If the device drifts again mid-run (e.g. the customer changes mode/reserve via + # the Tesla app after startup), that drift is not detected again until the process restarts. + # In the meantime it self-heals only on a Predbat "transition" - i.e. whenever Predbat's OWN + # desired value actually changes, since a changed target never matches the stale cache + # regardless of the device's true state. A repeated assertion of the SAME target while the + # device has silently drifted away stays stuck until the next boot. This is a ship-now, + # close-in-pilot decision; continuous periodic device-state reconciliation is a follow-up. + default_mode = response.get("default_real_mode") + if default_mode in OPERATION_MODES: + self.publish_control(self.entity("operation_mode", domain="select"), default_mode) + # Drift correction: refresh the dedupe cache to the device's ACTUAL mode. If it matches what + # Predbat last sent, the cache is unchanged (no spurious resend); if the user changed it + # externally (e.g. the Tesla app), the cache now reflects that drift, so the next assertion + # of Predbat's desired mode finds a mismatch and re-POSTs instead of being silently skipped. + self._last_sent["operation_mode"] = default_mode + backup_reserve = response.get("backup_reserve_percent") + if isinstance(backup_reserve, (int, float)): + self.publish_control(self.entity("backup_reserve", domain="number"), backup_reserve) + self._last_sent["backup_reserve"] = int(backup_reserve) # Same drift-correction rationale as operation_mode above. + return soc_max_published + + async def fetch_energy_today(self): + """Fetch today's cumulative energy, publishing daily kWh sensors. + + The Fleet/Teslemetry calendar_history endpoint requires a "kind" and a "period" query + parameter; without them it does not reliably return the daily energy time_series. + """ + data = await self._request("GET", "/api/1/energy_sites/{}/calendar_history?kind=energy&period=day".format(self.site_id)) + if not data: + return False + series = data.get("response", {}).get("time_series", []) + solar = grid_import = grid_export = load = 0.0 + for entry in series: + solar += entry.get("solar_energy_exported", 0) + grid_import += entry.get("grid_energy_imported", 0) + grid_export += entry.get("grid_energy_exported_from_solar", 0) + entry.get("grid_energy_exported_from_battery", 0) + entry.get("grid_energy_exported_from_generator", 0) + load += entry.get("consumer_energy_imported_from_grid", 0) + entry.get("consumer_energy_imported_from_solar", 0) + entry.get("consumer_energy_imported_from_battery", 0) + entry.get("consumer_energy_imported_from_generator", 0) + self.publish_sensor("solar_today", round(solar / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Solar Today") + self.publish_sensor("import_today", round(grid_import / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Import Today") + self.publish_sensor("export_today", round(grid_export / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Export Today") + self.publish_sensor("load_today", round(load / 1000.0, 3), unit="kWh", state_class="total_increasing", friendly="Powerwall Load Today") + return True + + async def run(self, seconds=0, first=False): + """Main component loop: poll data at the configured cadences. + + Returns True on a successful (or gated no-op) cycle and False on failure — + ComponentBase.start() uses this to set api_started, clear the startup flag + and drive its retry backoff, so a failing cycle must return False. + + Boot reconciliation is gated on the `reconcile_done` latch (mirroring the `site_info_done` + pattern) rather than on `first`: if the very first cycle 401s, the auth-failed fast-path + above returns before reconcile ever runs, but `first` is still consumed by that cycle. Gating + on a latch instead means reconcile still runs on the first cycle that is NOT auth-failed, + however many cycles that takes to arrive - a boot that 401s then recovers still reconciles. + """ + if not self.api_key or not self.site_id: + self.log("Warn: Teslemetry key or site_id not configured, skipping run") + return False + if self.api_auth_failed: + # Do not hammer the API with a dead token; only live_status probes recovery. + # A successful probe clears the flag in _request; a failed one returns False + # so ComponentBase keeps backing off. + if seconds - self.last_live_poll >= LIVE_POLL_SECONDS: + self.last_live_poll = seconds + return await self.fetch_live_status() + return False + success = True + if not self.site_info_done: + self.site_info_done = await self.fetch_site_info() + if not self.reconcile_done: + self.reconcile_done = await self.reconcile_on_start() + if first or (seconds - self.last_live_poll >= LIVE_POLL_SECONDS): + self.last_live_poll = seconds + success = await self.fetch_live_status() + if first or (seconds - self.last_energy_poll >= ENERGY_POLL_SECONDS): + self.last_energy_poll = seconds + await self.fetch_energy_today() + return success + + def register_control_entities(self): + """Publish the virtual control entities with their initial states. + + Also records each entity's display attributes (options/min/max/friendly_name) in + self.control_attrs, so publish_control can re-apply them on every subsequent write instead + of a bare value-only write silently replacing them with an empty attributes dict. + """ + controls = ( + ("operation_mode", "select", "self_consumption", {"friendly_name": "Powerwall Operation Mode", "options": OPERATION_MODES}), + ("backup_reserve", "number", 0, {"friendly_name": "Powerwall Backup Reserve", "min": 0, "max": 100, "step": 1, "unit_of_measurement": "%"}), + ("allow_charging_from_grid", "switch", "on", {"friendly_name": "Powerwall Allow Grid Charging"}), + ("allow_export", "select", "never", {"friendly_name": "Powerwall Allow Export", "options": EXPORT_RULES}), + ("tariff_mode", "select", "normal", {"friendly_name": "Powerwall Tariff Mode", "options": TARIFF_MODES}), + ) + self.control_attrs = {} + for suffix, domain, initial_state, attributes in controls: + entity_id = self.entity(suffix, domain=domain) + self.control_attrs[entity_id] = attributes + self.dashboard_item(entity_id, initial_state, attributes, app="teslemetry") + + def publish_control(self, entity_id, value): + """Write a control entity's new state, re-applying the attributes register_control_entities + recorded for it at init (options/min/max/friendly_name). + + A bare set_state_wrapper(entity, value) write defaults attributes to {}, which silently + replaces the options/min/max/friendly_name published once at startup and never resent, so + every select/number/switch write - and the reconcile/site_info writes that mirror device + state into these same entities - must route through here instead of calling + set_state_wrapper directly. + """ + attrs = getattr(self, "control_attrs", {}).get(entity_id, {}) + self.dashboard_item(entity_id, value, attrs, app="teslemetry") + + async def _command(self, path, body): + """POST a command; return True only if both the transport and the application layer succeeded. + + A 2xx HTTP status with a parsed JSON body is not sufficient on its own: Teslemetry/Fleet API can + return HTTP 200 carrying an application-level failure, so the body is also checked for a truthy + "error" key or a numeric "response.code" of 400 or above. + """ + result = await self._request("POST", "/api/1/energy_sites/{}/{}".format(self.site_id, path), json_body=body) + if result is None: + return False + if isinstance(result, dict): + if result.get("error"): + self.log("Warn: Teslemetry command to {} failed (application error): {}".format(path, str(result)[:200])) + return False + response = result.get("response") + if isinstance(response, dict): + code = response.get("code") + if isinstance(code, (int, float)) and not isinstance(code, bool) and code >= 400: + self.log("Warn: Teslemetry command to {} failed (response code {}): {}".format(path, code, str(result)[:200])) + return False + return True + + async def _apply_command(self, key, signature, sender, force=False): + """Send a control command only when its value has actually changed, to cut Teslemetry command-credit spend. + + `key` identifies which control this is ("operation_mode", "backup_reserve", "grid_charging", + "export_rule" or "tariff") and `signature` is a comparable representation of the value being + asserted (e.g. the mode string, or a canonical JSON serialisation of a built tariff body). + When `signature` already matches the last confirmed-sent value for `key`, `sender` is not + called at all - this is the actual command-credit saving, since the TESLA service hooks fire + their loopback event every cycle regardless of whether anything changed. + + Invariant (failure-retry): the dedupe cache is updated ONLY after a confirmed-successful send + (a True return from `sender`), never on a skip or a failure. A dropped/failed command therefore + leaves the cache stale, so the very next cycle's loopback event re-sends the same value instead + of being silently skipped forever - failure-retry is preserved even though every-cycle sends of + an unchanged value are deduped away. + + `force=True` bypasses the cache check and always calls `sender`, still refreshing the cache on + success. This is used by `reconcile_on_start` so its corrective writes can never be silently + skipped by a (present or future) pre-seeded cache entry that happens to already match what the + recovery is about to send. + + Drift-refresh limitation (accepted, not an oversight): unlike "operation_mode" and + "backup_reserve" (refreshed once per process from the device's actual state - see the NOTE in + `fetch_site_info`), "grid_charging" and "export_rule" are NEVER device-drift-refreshed at all, + because the site_info response this component reads does not expose either field's actual + device state. If the device drifts on these two externally, the cache is never proactively + corrected; they self-heal only when Predbat's own desired value changes (a transition), since a + changed target never matches the stale cache irrespective of the device's true state. + """ + if not force and self._last_sent.get(key) == signature: + return True + ok = await sender() + if ok: + self._last_sent[key] = signature + return ok + + async def set_operation_mode(self, mode, force=False): + """Set the Powerwall operation mode (self_consumption/autonomous/backup), deduped on write-on-change.""" + return await self._apply_command("operation_mode", mode, lambda: self._command("operation", {"default_real_mode": mode}), force=force) + + async def set_backup_reserve(self, percent, force=False): + """Set the backup reserve percentage (cloud caps at 80; 80-100 treated as 100 by Predbat), deduped on write-on-change.""" + percent = int(percent) + return await self._apply_command("backup_reserve", percent, lambda: self._command("backup", {"backup_reserve_percent": percent}), force=force) + + async def set_grid_charging(self, allow, force=False): + """Allow or disallow charging the battery from the grid, deduped on write-on-change.""" + return await self._apply_command("grid_charging", bool(allow), lambda: self._command("grid_import_export", {"disallow_charge_from_grid_with_solar_installed": not allow}), force=force) + + async def set_export_rule(self, rule, force=False): + """Set the grid export rule (never/pv_only/battery_ok), deduped on write-on-change.""" + return await self._apply_command("export_rule", rule, lambda: self._command("grid_import_export", {"customer_preferred_export_rule": rule}), force=force) + + def current_rates(self): + """Return (import_rate, export_rate) in GBP/kWh from Predbat's rate data when available. + + Defaults to class constants if rates are unavailable. Rates from the base are in PENCE + (pence per kWh), so they are divided by 100.0 to convert to GBP. + """ + import_rate = self.DEFAULT_IMPORT_RATE + export_rate = self.DEFAULT_EXPORT_RATE + base = getattr(self, "base", None) + if base is not None: + minutes_now = getattr(base, "minutes_now", None) + rate_import = getattr(base, "rate_import", None) + rate_export = getattr(base, "rate_export", None) + if minutes_now is not None and rate_import: + import_rate = round(rate_import.get(minutes_now, self.DEFAULT_IMPORT_RATE * 100) / 100.0, 4) + if minutes_now is not None and rate_export: + export_rate = round(rate_export.get(minutes_now, self.DEFAULT_EXPORT_RATE * 100) / 100.0, 4) + return import_rate, export_rate + + def build_tariff(self, mode, now=None): + """Build a tariff_content_v2 dict for the requested mode. + + export_now: current 30-minute-aligned window becomes ON_PEAK with a + high sell price so autonomous mode exports immediately; the window + self-expires. normal: flat tariff from the customer's current rates. + + The Powerwall applies tou_periods windows in the SITE'S LOCAL wall-clock time (the same + clock Predbat schedules against), not UTC, so when `now` is not injected (production use) + it defaults to the base's local time - base.now_utc converted to base.local_tz - rather than + a bare UTC clock read, which would misplace the window by the site's UTC offset. Falls back + to system UTC only when no base is wired up (e.g. some unit tests), since there is then no + site-local timezone to derive from. + """ + if now is None: + base = getattr(self, "base", None) + base_now_utc = getattr(base, "now_utc", None) if base is not None else None + if base_now_utc is not None: + local_tz = getattr(base, "local_tz", None) or getattr(self, "local_tz", None) or timezone.utc + now = base_now_utc.astimezone(local_tz) + else: + now = datetime.now(timezone.utc) + import_rate, export_rate = self.current_rates() + # Keep the synthetic ON_PEAK sell strictly above the live export rate, otherwise the + # export trick silently inverts at the highest-value moments. + sell_high = max(self.EXPORT_SELL_RATE, round(export_rate * 2, 4)) + + def charges(off_peak_buy, on_peak_buy): + """Build an energy_charges block.""" + rates = {"SUPER_OFF_PEAK": off_peak_buy} + if mode == "export_now": + rates["ON_PEAK"] = on_peak_buy + return {"ALL": {"rates": {"ALL": 0}}, "AllYear": {"rates": rates}} + + tou_periods = {"SUPER_OFF_PEAK": {"periods": [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": 0, "toMinute": 0}]}} + if mode == "export_now": + start = now.replace(minute=0 if now.minute < 30 else 30, second=0, microsecond=0) + end_minute_total = start.hour * 60 + start.minute + 60 + wrapped = end_minute_total >= 24 * 60 + end_hour, end_min = (end_minute_total // 60) % 24, end_minute_total % 60 + window = {"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": start.hour, "fromMinute": start.minute, "toHour": end_hour, "toMinute": end_min} + off_periods = [] + if wrapped: + # ON_PEAK crosses midnight, so its circle complement is a single segment from the wrapped end back to the start. + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": end_hour, "fromMinute": end_min, "toHour": start.hour, "toMinute": start.minute}) + else: + if start.hour > 0 or start.minute > 0: + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": start.hour, "toMinute": start.minute}) + if end_hour > 0 or end_min > 0: + off_periods.append({"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": end_hour, "fromMinute": end_min, "toHour": 0, "toMinute": 0}) + tou_periods = {"SUPER_OFF_PEAK": {"periods": off_periods}, "ON_PEAK": {"periods": [window]}} + + seasons = {"AllYear": {"fromMonth": 1, "fromDay": 1, "toMonth": 12, "toDay": 31, "tou_periods": tou_periods}} + sell_rates = charges(export_rate, sell_high) + # Buy-side ON_PEAK mirrors the high sell rate to discourage grid-charging during the export window. + buy_rates = charges(import_rate, sell_high) + common = { + "min_applicable_demand": 0, + "max_applicable_demand": 0, + "monthly_minimum_bill": 0, + "monthly_charges": 0, + "daily_charges": [{"name": "Charge", "amount": 0}], + "demand_charges": {"ALL": {"rates": {"ALL": 0}}, "AllYear": {"rates": {}}}, + } + return { + "version": 1, + "utility": "Predbat", + "code": "PREDBAT-{}".format(mode.upper().replace("_", "-")), + "name": "Predbat ({})".format(mode), + "currency": "GBP", + "daily_demand_charges": {}, + "energy_charges": buy_rates, + "seasons": seasons, + "sell_tariff": {**common, "utility": "Predbat", "energy_charges": sell_rates, "seasons": seasons}, + **common, + } + + async def set_tariff(self, mode, force=False): + """Push the tariff for the requested mode via time_of_use_settings, deduped on write-on-change. + + The dedupe signature is a canonical (sort_keys) JSON serialisation of the BUILT tariff body + rather than just `mode`: the export_now window and sell price are time-varying (the ON_PEAK + window advances every 30 minutes and the sell price tracks the live export rate), so the same + `mode` string can legitimately need re-sending when the underlying built tariff has changed, + while an unchanged build (the common case every 5-minute cycle) is skipped. + """ + tariff = self.build_tariff(mode) + signature = json.dumps(tariff, sort_keys=True) + return await self._apply_command("tariff", signature, lambda: self._command("time_of_use_settings", {"tou_settings": {"tariff_content_v2": tariff}}), force=force) + + @staticmethod + def _find_tariff_code(node): + """Recursively search a parsed JSON structure for the first string "code" key. + + The Teslemetry/Fleet API tariff_rate response shape is not guaranteed to nest the + code at one fixed path (top-level, under tariff_content_v2, or elsewhere), so this + walks dicts and lists looking for a "code" key rather than assuming a fixed path. + """ + if isinstance(node, dict): + code = node.get("code") + if isinstance(code, str): + return code + for value in node.values(): + found = TeslemetryAPI._find_tariff_code(value) + if found is not None: + return found + elif isinstance(node, list): + for item in node: + found = TeslemetryAPI._find_tariff_code(item) + if found is not None: + return found + return None + + async def get_current_tariff_code(self): + """Fetch the site's current tariff_rate from the device and return its tariff "code", or None on failure. + + Logs distinct warnings for the two failure modes: the read itself failing (network/HTTP error, empty + body) versus a successful read whose response simply carries no "code" key anywhere. + """ + data = await self._request("GET", "/api/1/energy_sites/{}/tariff_rate".format(self.site_id)) + if not data: + self.log("Warn: Teslemetry tariff_rate read failed - no response from the device") + return None + code = self._find_tariff_code(data) + if code is None: + self.log("Warn: Teslemetry tariff_rate response contained no tariff code: {}".format(str(data)[:200])) + return code + + def _is_read_only(self): + """Return True if Predbat is configured read-only, so device-write commands must not be sent. + + Reads the configuration via get_arg rather than the base's set_read_only attribute: the attribute + defaults to True in the PredBat constructor and is only refreshed from config during the fetch cycle, + which runs AFTER phase-1 components start - so at reconcile_on_start time the attribute would always + read True even for read-write instances. load_user_config(load_config=True) runs before phase-1 + component start, so get_arg returns the real configured value here. + """ + if getattr(self, "base", None) is None: + return False + return bool(self.get_arg("set_read_only", False)) + + async def reconcile_on_start(self): + """Restore a safe state if a previous run died mid-export. + + Reads the Powerwall's ACTUAL current tariff from the device rather than the local + entity mirror: register_control_entities() unconditionally reseeds the mirror to + "normal" on every boot, so trusting it would make this recovery path unreachable. + If the device's tariff still carries the PREDBAT-EXPORT-NOW marker written by + build_tariff(), a previous run crashed mid-export, so normal tariff and disabled + export are restored (unless Predbat is running read-only, in which case the need + for recovery is logged but no write is sent). If no tariff code can be determined + (read failure or code-less response, each logged distinctly by + get_current_tariff_code), this skips rather than guessing at the device state. + + Returns True once this has actually executed against a live API (the tariff read + returned a usable code, whether or not a restore was needed), so run() can latch + reconcile_done and stop retrying. Returns False when the tariff read itself failed to + return anything usable, so run() retries on a later cycle instead of latching a no-op. + + The restore writes below pass force=True to `set_tariff`/`set_export_rule` so this recovery + path can NEVER be silently skipped by the write-on-change dedupe cache - at boot the cache is + empty so this is naturally a no-op today, but forcing makes that explicit and future-proof + against a cache-preseed feature that might otherwise make recovery look like a no-op. + """ + tariff_code = await self.get_current_tariff_code() + if tariff_code is None: + return False + export_now_code = self.build_tariff("export_now").get("code") + if tariff_code != export_now_code: + return True + if self._is_read_only(): + self.log("Warn: Teslemetry device tariff was left as {} - recovery needed but skipped (read-only mode)".format(tariff_code)) + return True + self.log("Warn: Teslemetry device tariff was left as {} - restoring normal tariff and disabling export".format(tariff_code)) + if await self.set_tariff("normal", force=True): + self.publish_control(self.entity("tariff_mode", domain="select"), "normal") + if await self.set_export_rule("never", force=True): + self.publish_control(self.entity("allow_export", domain="select"), "never") + return True + + async def select_event(self, entity_id, value): + """Handle select changes routed from the service-hook loopback.""" + success = False + if entity_id.endswith("_operation_mode") and value in OPERATION_MODES: + success = await self.set_operation_mode(value) + elif entity_id.endswith("_allow_export") and value in EXPORT_RULES: + success = await self.set_export_rule(value) + elif entity_id.endswith("_tariff_mode") and value in TARIFF_MODES: + success = await self.set_tariff(value) + else: + self.log("Warn: Teslemetry unhandled select_event {} = {}".format(entity_id, value)) + return + if success: + self.publish_control(entity_id, value) + else: + self.log("Warn: Teslemetry command failed for {} = {} (state not updated)".format(entity_id, value)) + + async def number_event(self, entity_id, value): + """Handle number changes routed from the service-hook loopback. + + Guards the int(float(value)) conversion: a number.increment/decrement service call can + forward value=None, which would otherwise raise a TypeError out of the dispatch loop. + """ + if entity_id.endswith("_backup_reserve"): + try: + percent = max(0, min(100, int(float(value)))) + except (ValueError, TypeError): + self.log("Warn: Teslemetry invalid number_event value for {}: {}".format(entity_id, value)) + return + if await self.set_backup_reserve(percent): + self.publish_control(entity_id, percent) + else: + self.log("Warn: Teslemetry backup_reserve command failed (state not updated)") + else: + self.log("Warn: Teslemetry unhandled number_event {} = {}".format(entity_id, value)) + + async def switch_event(self, entity_id, service): + """Handle switch service calls routed from the service-hook loopback.""" + if entity_id.endswith("_allow_charging_from_grid"): + allow = service != "turn_off" + if await self.set_grid_charging(allow): + self.publish_control(entity_id, "on" if allow else "off") + else: + self.log("Warn: Teslemetry grid charging command failed (state not updated)") + else: + self.log("Warn: Teslemetry unhandled switch_event {} {}".format(entity_id, service)) + + async def final(self): + """Cleanup on shutdown.""" + self.log("Info: TeslemetryAPI shutdown") diff --git a/apps/predbat/tests/test_teslemetry.py b/apps/predbat/tests/test_teslemetry.py new file mode 100644 index 000000000..067d3440c --- /dev/null +++ b/apps/predbat/tests/test_teslemetry.py @@ -0,0 +1,957 @@ +# ----------------------------------------------------------------------------- +# Predbat Home Battery System +# Copyright Trefor Southwell 2026 - All Rights Reserved +# This application maybe used for personal use only and not for commercial use +# ----------------------------------------------------------------------------- +"""Unit tests for the TeslemetryAPI component (Tesla Powerwall via Teslemetry).""" + +from unittest.mock import MagicMock, patch, AsyncMock + +from tests.test_infra import create_aiohttp_mock_response, create_aiohttp_mock_session, run_async +from teslemetry import TeslemetryAPI, OPERATION_MODES + + +class MockTeslemetryAPI(TeslemetryAPI): + """TeslemetryAPI test double that avoids ComponentBase initialisation.""" + + def __init__(self): + """Initialise the mock with captured state instead of ComponentBase wiring.""" + self.api_key = "test_token" + self.site_id = "123456" + self.base_url = "https://api.teslemetry.com" + self.prefix = "predbat" + self.api_auth_failed = False + self.last_live_poll = 0 + self.last_energy_poll = 0 + self.site_info_done = False + self.reconcile_done = False + self.dashboard_items = {} + self.log_messages = [] + self.entity_states = {} + self.mock_responses = {} + self.requests_made = [] + self._last_sent = {} + + def log(self, msg): + """Capture log messages.""" + self.log_messages.append(msg) + + def dashboard_item(self, entity, state, attributes, app=None): + """Capture dashboard items (mirrors production dashboard_item, which also calls through to + set_state_wrapper - so entity_states reflects the latest write here too).""" + self.dashboard_items[entity] = {"state": state, "attributes": attributes} + self.entity_states[entity] = state + + def set_state_wrapper(self, entity_id, state, attributes={}): + """Capture entity state updates.""" + self.entity_states[entity_id] = state + + def get_state_wrapper(self, entity_id=None, default=None, **kwargs): + """Return captured entity state.""" + return self.entity_states.get(entity_id, default) + + def update_success_timestamp(self): + """No-op for tests.""" + pass + + async def _request(self, method, path, json_body=None): + """Return canned responses instead of real HTTP.""" + self.requests_made.append((method, path, json_body)) + return self.mock_responses.get(path, None) + + +LIVE_STATUS = { + "response": { + "percentage_charged": 55.5, + "battery_power": 1200, + "solar_power": 800, + "load_power": 600, + "grid_power": -1400, + "island_status": "on_grid", + } +} + +SITE_INFO = { + "response": { + "nameplate_energy": 13500, + "default_real_mode": "self_consumption", + "backup_reserve_percent": 20, + } +} + +TARIFF_RATE_EXPORT_NOW = {"response": {"tariff_content_v2": {"version": 1, "utility": "Predbat", "code": "PREDBAT-EXPORT-NOW", "name": "Predbat (export_now)"}}} + +TARIFF_RATE_NORMAL = {"response": {"tariff_content_v2": {"version": 1, "utility": "Predbat", "code": "PREDBAT-NORMAL", "name": "Predbat (normal)"}}} + +ENERGY_HISTORY = { + "response": { + "time_series": [ + { + "solar_energy_exported": 4000, + "grid_energy_imported": 2500, + "grid_energy_exported_from_solar": 1000, + "grid_energy_exported_from_battery": 500, + "grid_energy_exported_from_generator": 0, + "consumer_energy_imported_from_grid": 2000, + "consumer_energy_imported_from_solar": 1500, + "consumer_energy_imported_from_battery": 700, + "consumer_energy_imported_from_generator": 0, + } + ] + } +} + + +def test_teslemetry_entity_names(): + """Entity helper builds prefixed ids.""" + api = MockTeslemetryAPI() + assert api.entity("soc") == "sensor.predbat_teslemetry_soc" + assert api.entity("operation_mode", domain="select") == "select.predbat_teslemetry_operation_mode" + + +def test_teslemetry_live_status_publishes_sensors(): + """live_status data is published as power/SOC sensors.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/live_status"] = LIVE_STATUS + run_async(api.fetch_live_status()) + assert api.dashboard_items["sensor.predbat_teslemetry_soc"]["state"] == 55.5 + assert api.dashboard_items["sensor.predbat_teslemetry_battery_power"]["state"] == 1200 + assert api.dashboard_items["sensor.predbat_teslemetry_grid_power"]["state"] == -1400 + assert api.dashboard_items["sensor.predbat_teslemetry_solar_power"]["state"] == 800 + assert api.dashboard_items["sensor.predbat_teslemetry_load_power"]["state"] == 600 + + +def test_teslemetry_site_info_publishes_soc_max(): + """site_info nameplate energy is published as soc_max in kWh.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 + + +def test_teslemetry_site_info_seeds_control_entity_states(): + """site_info seeds the operation_mode select and backup_reserve number entity states from the device (display only, no commands).""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "self_consumption" + assert api.entity_states["number.predbat_teslemetry_backup_reserve"] == 20 + assert api.requests_made == [("GET", "/api/1/energy_sites/123456/site_info", None)] + + +def test_teslemetry_energy_today_publishes_kwh(): + """calendar_history energy series is aggregated into daily kWh sensors.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/calendar_history?kind=energy&period=day"] = ENERGY_HISTORY + run_async(api.fetch_energy_today()) + assert api.dashboard_items["sensor.predbat_teslemetry_solar_today"]["state"] == 4.0 + assert api.dashboard_items["sensor.predbat_teslemetry_import_today"]["state"] == 2.5 + assert api.dashboard_items["sensor.predbat_teslemetry_export_today"]["state"] == 1.5 + assert api.dashboard_items["sensor.predbat_teslemetry_load_today"]["state"] == 4.2 + + +def _make_real_request_api(): + """Build a MockTeslemetryAPI with the REAL TeslemetryAPI._request bound back in place of the canned override.""" + api = MockTeslemetryAPI() + api._request = TeslemetryAPI._request.__get__(api) + return api + + +def test_teslemetry_request_auth_failure(): + """A 401 response makes the real _request return None and latch api_auth_failed.""" + + async def test(): + """Drive the real _request against a mocked 401 response.""" + api = _make_real_request_api() + mock_response = create_aiohttp_mock_response(status=401, json_data={"error": "unauthorised"}) + mock_session = create_aiohttp_mock_session(mock_response) + mock_session.request = mock_session.get + with patch("aiohttp.ClientSession") as mock_session_class: + with patch("teslemetry.asyncio.sleep", new_callable=AsyncMock): + mock_session_class.return_value = mock_session + result = await api._request("GET", "/api/1/energy_sites/123456/live_status") + assert result is None + assert api.api_auth_failed is True + assert mock_session.request.call_count == 1 # No retries on auth failure + assert any("auth failed" in msg for msg in api.log_messages) + + run_async(test()) + + +def test_teslemetry_request_rate_limit_retry(): + """A 429 response is retried with backoff and the follow-up 200 JSON is returned.""" + + async def test(): + """Drive the real _request against a mocked 429-then-200 response sequence.""" + api = _make_real_request_api() + resp_429 = create_aiohttp_mock_response(status=429, json_data={"error": "rate limited"}) + resp_200 = create_aiohttp_mock_response(status=200, json_data={"response": {"ok": 1}}) + mock_session = create_aiohttp_mock_session(resp_200) + mock_session.request = MagicMock(side_effect=[resp_429, resp_200]) + with patch("aiohttp.ClientSession") as mock_session_class: + with patch("teslemetry.asyncio.sleep", new_callable=AsyncMock) as mock_sleep: + mock_session_class.return_value = mock_session + result = await api._request("GET", "/api/1/energy_sites/123456/site_info") + assert result == {"response": {"ok": 1}} + assert mock_session.request.call_count == 2 + assert mock_sleep.await_count == 1 + assert api.api_auth_failed is False + + run_async(test()) + + +def test_teslemetry_request_success_clears_auth_flag(): + """A 200 response clears a previously latched api_auth_failed flag.""" + + async def test(): + """Drive the real _request against a mocked 200 response with the auth flag pre-latched.""" + api = _make_real_request_api() + api.api_auth_failed = True + mock_response = create_aiohttp_mock_response(status=200, json_data={"response": {"percentage_charged": 50}}) + mock_session = create_aiohttp_mock_session(mock_response) + mock_session.request = mock_session.get + with patch("aiohttp.ClientSession") as mock_session_class: + with patch("teslemetry.asyncio.sleep", new_callable=AsyncMock): + mock_session_class.return_value = mock_session + result = await api._request("GET", "/api/1/energy_sites/123456/live_status") + assert result == {"response": {"percentage_charged": 50}} + assert api.api_auth_failed is False + + run_async(test()) + + +def _make_run_api_with_fetch_capture(site_info=True, live_status=True, energy_today=True): + """Build a MockTeslemetryAPI whose fetch methods are stubbed to record calls, returning (api, calls list).""" + api = MockTeslemetryAPI() + calls = [] + + async def fake_site_info(): + """Record a site_info fetch.""" + calls.append("site_info") + return site_info + + async def fake_live_status(): + """Record a live_status fetch.""" + calls.append("live_status") + return live_status + + async def fake_energy_today(): + """Record an energy_today fetch.""" + calls.append("energy_today") + return energy_today + + api.fetch_site_info = fake_site_info + api.fetch_live_status = fake_live_status + api.fetch_energy_today = fake_energy_today + return api, calls + + +def test_teslemetry_run_unconfigured_returns_false(): + """run() returns False (not None) when key or site_id is missing, so ComponentBase treats it as a failed cycle.""" + api, calls = _make_run_api_with_fetch_capture() + api.api_key = "" + assert run_async(api.run(0, True)) is False + assert calls == [] + api = MockTeslemetryAPI() + api.site_id = "" + assert run_async(api.run(0, True)) is False + + +def test_teslemetry_run_first_success_returns_true(): + """A successful first run fetches everything and returns True; a gated-out steady-state cycle is a successful no-op.""" + api, calls = _make_run_api_with_fetch_capture() + assert run_async(api.run(0, True)) is True + assert calls == ["site_info", "live_status", "energy_today"] + # Gated-out 60s tick: nothing due yet, but the cycle itself is healthy + calls.clear() + assert run_async(api.run(60, False)) is True + assert calls == [] + # Failed live_status poll propagates as a failed cycle + api, calls = _make_run_api_with_fetch_capture(live_status=False) + assert run_async(api.run(0, True)) is False + + +def test_teslemetry_run_auth_failed_only_probes_live_status(): + """While auth-failed, run() probes ONLY live_status on the poll cadence (no site_info/energy, even when first).""" + api, calls = _make_run_api_with_fetch_capture() + api.api_auth_failed = True + # Gated-out tick: no API traffic at all, cycle reports failure to keep backoff + assert run_async(api.run(60, False)) is False + assert calls == [] + # Poll due (first=True must NOT sneak in site_info while auth-failed) + assert run_async(api.run(120, True)) is True + assert calls == ["live_status"] + + +def test_teslemetry_select_operation_mode(): + """select_event on operation_mode POSTs /operation and updates entity state on success.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert ("POST", "/api/1/energy_sites/123456/operation", {"default_real_mode": "backup"}) in api.requests_made + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "backup" + + +def test_teslemetry_select_failure_keeps_state(): + """Failed command does not update entity state (write_and_poll will retry).""" + api = MockTeslemetryAPI() + # No mock response registered -> _request returns None -> command fails + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert "select.predbat_teslemetry_operation_mode" not in api.entity_states + + +def test_teslemetry_command_success_on_low_response_code(): + """_command treats a response.code below 400 as success.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is True + + +def test_teslemetry_command_failure_on_application_error_key(): + """_command treats a body carrying an "error" key as a failure even though _request returned parsed JSON.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"error": "invalid_argument"} + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is False + assert any("failed" in msg for msg in api.log_messages) + + +def test_teslemetry_command_failure_on_high_response_code(): + """_command treats an application-level response.code >= 400 as a failure despite a 2xx transport status.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 401}} + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is False + assert any("failed" in msg for msg in api.log_messages) + + +def test_teslemetry_command_none_response_is_failure(): + """_command returns False when _request itself returns None (unchanged transport-failure behaviour).""" + api = MockTeslemetryAPI() + # No mock response registered -> _request returns None + assert run_async(api._command("operation", {"default_real_mode": "backup"})) is False + + +def test_teslemetry_number_backup_reserve(): + """number_event on backup_reserve POSTs /backup with an integer percent.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/backup"] = {"response": {"code": 201}} + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 100)) + assert ("POST", "/api/1/energy_sites/123456/backup", {"backup_reserve_percent": 100}) in api.requests_made + assert api.entity_states["number.predbat_teslemetry_backup_reserve"] == 100 + + +def test_teslemetry_switch_grid_charging(): + """switch_event maps turn_off to disallow grid charging.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.switch_event("switch.predbat_teslemetry_allow_charging_from_grid", "turn_off")) + assert ("POST", "/api/1/energy_sites/123456/grid_import_export", {"disallow_charge_from_grid_with_solar_installed": True}) in api.requests_made + assert api.entity_states["switch.predbat_teslemetry_allow_charging_from_grid"] == "off" + + +def test_teslemetry_select_export_rule(): + """select_event on allow_export POSTs the customer_preferred_export_rule.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_allow_export", "battery_ok")) + assert ("POST", "/api/1/energy_sites/123456/grid_import_export", {"customer_preferred_export_rule": "battery_ok"}) in api.requests_made + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "battery_ok" + + +def test_teslemetry_site_info_latch(): + """site_info fetch is retried on every cycle until success; latch prevents redundant calls once successful.""" + api, calls = _make_run_api_with_fetch_capture(site_info=False, live_status=True, energy_today=True) + # First cycle: site_info missing, fetch returns False, latch stays False + assert run_async(api.run(0, True)) is True + assert calls == ["site_info", "live_status", "energy_today"] + assert api.site_info_done is False + # Now provide the site_info mock and unbind fetch_site_info so the real one runs + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + api.fetch_site_info = TeslemetryAPI.fetch_site_info.__get__(api) + # Re-create the fake live_status/energy_today so they still record and return True + api.fetch_live_status = _make_run_api_with_fetch_capture(live_status=True)[0].fetch_live_status + api.fetch_energy_today = _make_run_api_with_fetch_capture(energy_today=True)[0].fetch_energy_today + # Run again: site_info_done latch should trigger fetch_site_info and succeed + assert run_async(api.run(300, False)) is True + # site_info_done should now be True and soc_max should be published + assert api.site_info_done is True + assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 + + +def test_teslemetry_build_tariff_export_now(): + """export_now tariff marks the current window ON_PEAK with a high sell rate.""" + from datetime import datetime + + api = MockTeslemetryAPI() + now = datetime(2026, 7, 2, 14, 40) + tariff = api.build_tariff("export_now", now=now) + assert tariff["version"] == 1 + sell = tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"] + assert sell["ON_PEAK"] > sell["SUPER_OFF_PEAK"] + periods = tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"] + assert periods[0]["fromHour"] == 14 and periods[0]["fromMinute"] == 30 + assert periods[0]["toHour"] == 15 and periods[0]["toMinute"] == 30 + + +def test_teslemetry_build_tariff_normal_flat(): + """normal tariff is flat (single ALL rate, no ON_PEAK windows).""" + api = MockTeslemetryAPI() + tariff = api.build_tariff("normal") + assert "ON_PEAK" not in tariff["seasons"]["AllYear"]["tou_periods"] + assert tariff["energy_charges"]["AllYear"]["rates"]["SUPER_OFF_PEAK"] >= 0 + + +def test_teslemetry_set_tariff_posts_tou_settings(): + """set_tariff wraps the tariff in tou_settings and POSTs it.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + result = run_async(api.set_tariff("normal")) + assert result is True + method, path, body = api.requests_made[-1] + assert path == "/api/1/energy_sites/123456/time_of_use_settings" + assert "tariff_content_v2" in body["tou_settings"] + + +def test_teslemetry_reconcile_on_start_device_marker_restores_normal(): + """Boot reconciliation reads the ACTUAL device tariff (not the local entity mirror); a PREDBAT-EXPORT-NOW + marker on the device restores normal tariff + never export and updates the mirror entities on success.""" + api = MockTeslemetryAPI() + # The entity mirror is always reseeded to "normal" by register_control_entities() on boot, so it must + # be irrelevant here - only the device response drives the outcome. + api.entity_states["select.predbat_teslemetry_tariff_mode"] = "normal" + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.reconcile_on_start()) + assert ("GET", "/api/1/energy_sites/123456/tariff_rate", None) in api.requests_made + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + assert any("PREDBAT-EXPORT-NOW" in msg for msg in api.log_messages) + + +def test_teslemetry_reconcile_on_start_device_normal_is_noop(): + """Boot reconciliation issues zero command requests when the device tariff is not the export_now marker.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_NORMAL + run_async(api.reconcile_on_start()) + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + + +def test_teslemetry_reconcile_on_start_read_failure_skips_without_crash(): + """Boot reconciliation skips (no writes, no exception) when the device tariff read fails, logging a read failure.""" + api = MockTeslemetryAPI() + # No mock response registered for tariff_rate -> _request returns None + run_async(api.reconcile_on_start()) + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + assert any("read failed" in msg for msg in api.log_messages) + assert not any("no tariff code" in msg for msg in api.log_messages) + + +def test_teslemetry_reconcile_on_start_no_code_in_response_skips(): + """Boot reconciliation skips with a DISTINCT log when the tariff read succeeds but the response carries no code key.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = {"response": {"tariff_content_v2": {"version": 1, "utility": "SomeUtility"}}} + run_async(api.reconcile_on_start()) + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + assert any("no tariff code" in msg for msg in api.log_messages) + assert not any("read failed" in msg for msg in api.log_messages) + + +def test_teslemetry_reconcile_on_start_read_only_mode_skips_writes(): + """When get_arg reports read-only, a device tariff still marked PREDBAT-EXPORT-NOW is logged but not written back.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.base = SimpleNamespace(set_read_only=True, get_arg=lambda arg, default=None, **kwargs: True if arg == "set_read_only" else default) + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.reconcile_on_start()) + assert [req for req in api.requests_made if req[0] == "POST"] == [] + assert api.entity_states == {} + assert any("read-only" in msg.lower() for msg in api.log_messages) + + +def test_teslemetry_reconcile_on_start_ignores_boot_default_read_only_attribute(): + """The recovery write must NOT be gated on base.set_read_only: the constructor defaults that attribute to True + and it is only refreshed from config in the fetch cycle AFTER phase-1 components start, so at reconcile time it + is always True. With get_arg (the real config source) reporting False, the writes must go ahead regardless.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + # Simulate the real boot condition: stale constructor default True on the attribute, real config value False. + api.base = SimpleNamespace(set_read_only=True, get_arg=lambda arg, default=None, **kwargs: False if arg == "set_read_only" else default) + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + run_async(api.reconcile_on_start()) + assert any(req[1].endswith("/time_of_use_settings") for req in api.requests_made if req[0] == "POST") + assert any(req[1].endswith("/grid_import_export") for req in api.requests_made if req[0] == "POST") + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + assert not any("read-only" in msg.lower() for msg in api.log_messages) + + +def _assert_tou_periods_partition_day(tou_periods): + """Assert the tou_periods cover every minute of the (circular) day exactly once — no overlaps, no gaps.""" + covered = [0] * (24 * 60) + for group in tou_periods.values(): + for period in group["periods"]: + from_min = period["fromHour"] * 60 + period["fromMinute"] + to_min = period["toHour"] * 60 + period["toMinute"] + if to_min <= from_min: + to_min += 24 * 60 + for minute in range(from_min, to_min): + covered[minute % (24 * 60)] += 1 + assert max(covered) <= 1, "tou_periods overlap" + assert min(covered) >= 1, "tou_periods leave a gap" + + +def test_teslemetry_build_tariff_export_now_midnight_wrap(): + """export_now at 23:40 wraps midnight: ON_PEAK 23:30-00:30 with a single non-overlapping off-peak complement.""" + from datetime import datetime + + api = MockTeslemetryAPI() + tariff = api.build_tariff("export_now", now=datetime(2026, 7, 2, 23, 40)) + tou_periods = tariff["seasons"]["AllYear"]["tou_periods"] + assert tou_periods["ON_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 23, "fromMinute": 30, "toHour": 0, "toMinute": 30}] + assert tou_periods["SUPER_OFF_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 30, "toHour": 23, "toMinute": 30}] + _assert_tou_periods_partition_day(tou_periods) + + +def test_teslemetry_build_tariff_export_now_midnight_exact(): + """export_now at 23:10 ends exactly at midnight: ON_PEAK 23:00-00:00 with a single off-peak complement.""" + from datetime import datetime + + api = MockTeslemetryAPI() + tariff = api.build_tariff("export_now", now=datetime(2026, 7, 2, 23, 10)) + tou_periods = tariff["seasons"]["AllYear"]["tou_periods"] + assert tou_periods["ON_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 23, "fromMinute": 0, "toHour": 0, "toMinute": 0}] + assert tou_periods["SUPER_OFF_PEAK"]["periods"] == [{"fromDayOfWeek": 0, "toDayOfWeek": 6, "fromHour": 0, "fromMinute": 0, "toHour": 23, "toMinute": 0}] + _assert_tou_periods_partition_day(tou_periods) + + +def test_teslemetry_build_tariff_sell_clamp_above_export_rate(): + """ON_PEAK sell (and buy) stay above the live export rate even when it exceeds the static high rate.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.base = SimpleNamespace(minutes_now=600, rate_import={600: 30.0}, rate_export={600: 60.0}) + tariff = api.build_tariff("export_now") + sell = tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"] + assert sell["SUPER_OFF_PEAK"] == 0.6 + assert sell["ON_PEAK"] > 0.6 + assert sell["ON_PEAK"] == 1.2 + # Buy-side ON_PEAK matches the high sell rate to discourage grid-charging during the export window + assert tariff["energy_charges"]["AllYear"]["rates"]["ON_PEAK"] == sell["ON_PEAK"] + + +def test_teslemetry_reconcile_on_start_partial_failure(): + """Partial reconcile: tariff restore succeeds but the export-rule command fails, so only tariff_mode is updated.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + # grid_import_export mock ABSENT -> _request returns None -> set_export_rule fails + run_async(api.reconcile_on_start()) + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert "select.predbat_teslemetry_allow_export" not in api.entity_states + + +def test_teslemetry_build_tariff_uses_site_local_time_not_utc(): + """build_tariff's export_now window must be expressed in the SITE'S LOCAL wall-clock time (the + same clock Predbat schedules in), not raw UTC - the Powerwall applies tou_periods in local time. + + A base whose local time is 15:40 during BST (UTC+1) must yield an ON_PEAK window starting at + 15:30 local. A naive UTC-only implementation would instead read whatever the system clock (or, + as sabotaged here, a mocked bogus UTC time) reports, producing the wrong window boundaries. + """ + import pytz + from datetime import datetime as real_datetime + from types import SimpleNamespace + + api = MockTeslemetryAPI() + london = pytz.timezone("Europe/London") + local_now = london.localize(real_datetime(2026, 7, 2, 15, 40)) + api.base = SimpleNamespace(now_utc=local_now, local_tz=london, minutes_now=None, rate_import=None, rate_export=None) + + # Sabotage a naive implementation: if build_tariff ignores the base and calls datetime.now(timezone.utc) + # directly, it will see this deliberately-wrong bogus time (03:05) instead of the real 15:40 local time above. + with patch("teslemetry.datetime") as mock_datetime: + mock_datetime.now.return_value = real_datetime(2026, 7, 2, 3, 5) + tariff = api.build_tariff("export_now") + + periods = tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"] + assert periods[0]["fromHour"] == 15 and periods[0]["fromMinute"] == 30 + + +def test_teslemetry_reconcile_latch_survives_auth_failed_first_cycle(): + """A first cycle that 401s (auth-failed, consuming `first`) must not permanently skip reconcile: + once the API recovers on a later cycle, reconcile_on_start still runs and restores a stuck export + state, because reconcile is gated on a `reconcile_done` latch rather than on `first`.""" + api, calls = _make_run_api_with_fetch_capture() + api.api_auth_failed = True # Simulate: the boot cycle already found the token dead. + assert api.reconcile_done is False + # Cycle 1 (first=True): auth-failed fast path, poll not yet due -> no HTTP traffic at all, no reconcile chance. + assert run_async(api.run(0, True)) is False + assert calls == [] + assert api.reconcile_done is False + + # Auth recovers (e.g. a live_status probe on an intervening cycle succeeded and cleared the flag). + api.api_auth_failed = False + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + # Cycle 2 (first=False): reconcile must still run even though `first` was already consumed in cycle 1. + assert run_async(api.run(300, False)) is True + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + assert api.reconcile_done is True + + +def test_teslemetry_reconcile_latch_runs_once_on_healthy_boot(): + """A healthy first cycle runs reconcile exactly once; the latch prevents a redundant reconcile + call (and its tariff_rate read) on every subsequent cycle.""" + api, calls = _make_run_api_with_fetch_capture() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_NORMAL + assert run_async(api.run(0, True)) is True + assert api.reconcile_done is True + tariff_reads = [req for req in api.requests_made if req[1].endswith("/tariff_rate")] + assert len(tariff_reads) == 1 + # A later cycle must not repeat the reconcile tariff_rate read. + assert run_async(api.run(600, False)) is True + tariff_reads = [req for req in api.requests_made if req[1].endswith("/tariff_rate")] + assert len(tariff_reads) == 1 + + +def test_teslemetry_site_info_returns_false_without_nameplate_so_soc_max_retries(): + """fetch_site_info must return False when nameplate_energy is absent/zero so run()'s + site_info_done latch stays False and soc_max is retried on a later cycle - even though other + fields (control-entity seeding) were published from the same response.""" + api = MockTeslemetryAPI() + site_info_no_nameplate = {"response": {"nameplate_energy": 0, "default_real_mode": "self_consumption", "backup_reserve_percent": 20}} + api.mock_responses["/api/1/energy_sites/123456/site_info"] = site_info_no_nameplate + assert run_async(api.fetch_site_info()) is False + assert "sensor.predbat_teslemetry_soc_max" not in api.dashboard_items + # Control-entity seeding still happens even though soc_max could not be published. + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "self_consumption" + + # Nameplate becomes available on a later cycle - now it must succeed and return True. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + assert run_async(api.fetch_site_info()) is True + assert api.dashboard_items["sensor.predbat_teslemetry_soc_max"]["state"] == 13.5 + + +def test_teslemetry_run_site_info_latch_stays_false_without_nameplate(): + """At the run() level, a site_info response without nameplate_energy must leave site_info_done + False so the fetch is retried on a later cycle, rather than latching done with no soc_max ever + published.""" + api = MockTeslemetryAPI() + api.fetch_site_info = TeslemetryAPI.fetch_site_info.__get__(api) + api.mock_responses["/api/1/energy_sites/123456/site_info"] = {"response": {"nameplate_energy": 0}} + run_async(api.run(0, True)) + assert api.site_info_done is False + + +def test_teslemetry_energy_today_requests_kind_and_period(): + """fetch_energy_today must query calendar_history with kind=energy&period=day - the Fleet/Teslemetry + endpoint requires both, and omitting them (the previous bare-path request) produces an invalid or + empty response.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/calendar_history?kind=energy&period=day"] = ENERGY_HISTORY + run_async(api.fetch_energy_today()) + assert api.requests_made == [("GET", "/api/1/energy_sites/123456/calendar_history?kind=energy&period=day", None)] + assert api.dashboard_items["sensor.predbat_teslemetry_solar_today"]["state"] == 4.0 + + +def test_teslemetry_select_event_preserves_control_attributes(): + """select_event must not wipe the options/min/max/friendly_name attributes that + register_control_entities set at init - every control write must re-apply them via + publish_control, verified here through the dashboard_items capture (which records attributes, + unlike the bare entity_states mirror).""" + api = MockTeslemetryAPI() + api.register_control_entities() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + item = api.dashboard_items["select.predbat_teslemetry_operation_mode"] + assert item["state"] == "backup" + assert item["attributes"].get("options") == OPERATION_MODES + + +def test_teslemetry_number_event_guards_null_value(): + """number_event must not crash on a null value (e.g. forwarded from a number.increment/decrement + service call without an explicit value) - it should log a warning and post no command.""" + api = MockTeslemetryAPI() + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", None)) + assert api.requests_made == [] + assert "number.predbat_teslemetry_backup_reserve" not in api.entity_states + assert any("invalid" in msg.lower() for msg in api.log_messages) + + +def test_teslemetry_dedupe_operation_mode_skips_repeat_post(): + """Sending the same operation_mode value twice via select_event posts only once (write-on-change).""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 1 + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "backup" + + +def test_teslemetry_dedupe_operation_mode_resends_on_change(): + """A changed operation_mode value must still POST (dedupe only skips an unchanged value).""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "autonomous")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "autonomous" + + +def test_teslemetry_dedupe_failed_post_not_cached_so_retries(): + """A failed POST must NOT update the dedupe cache, so the identical value is re-sent next cycle + (failure-retry survives the write-on-change dedupe).""" + api = MockTeslemetryAPI() + # No mock response registered for /operation -> _request returns None -> command fails. + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert api.requests_made == [("POST", "/api/1/energy_sites/123456/operation", {"default_real_mode": "backup"})] + assert "select.predbat_teslemetry_operation_mode" not in api.entity_states + assert "operation_mode" not in api._last_sent + # API recovers; the SAME value must still be sent - the cache was never populated on failure. + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + assert api.entity_states["select.predbat_teslemetry_operation_mode"] == "backup" + + +def test_teslemetry_dedupe_tariff_identical_body_skips_repeat_post(): + """Two normal-tariff builds with an identical body POST only once.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + run_async(api.set_tariff("normal")) + run_async(api.set_tariff("normal")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 1 + + +def test_teslemetry_dedupe_tariff_resends_when_rates_change(): + """A tariff rebuild whose sell/buy price actually changed must re-POST rather than be deduped.""" + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.base = SimpleNamespace(minutes_now=600, rate_import={600: 30.0}, rate_export={600: 15.0}, now_utc=None, local_tz=None) + run_async(api.set_tariff("normal")) + api.base.rate_export = {600: 60.0} # Export rate jumps, so the sell-side signature changes. + run_async(api.set_tariff("normal")) + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + + +def test_teslemetry_drift_correction_refreshes_cache_and_reasserts(): + """A site_info poll reporting a device mode that differs from what Predbat last sent must refresh + the dedupe cache to the ACTUAL device value, so the next assertion of Predbat's desired mode is not + skipped - this is how externally-changed (Tesla app) drift gets corrected rather than silently kept.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + assert api._last_sent["operation_mode"] == "backup" + # SITE_INFO reports default_real_mode = self_consumption, i.e. the user changed it externally. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api._last_sent["operation_mode"] == "self_consumption" + # Predbat's desired mode is still "backup" - the drifted cache means this must re-POST. + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "backup")) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/operation")] + assert len(posts) == 2 + + +def test_teslemetry_drift_correction_no_spurious_resend_when_matching(): + """When site_info reports the SAME mode Predbat last sent, the dedupe cache is unchanged and the + next assertion of that mode is still skipped - no spurious re-send from the drift-refresh itself.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/operation"] = {"response": {"code": 201}} + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "self_consumption")) + assert api._last_sent["operation_mode"] == "self_consumption" + # SITE_INFO's default_real_mode is also self_consumption - no drift. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api._last_sent["operation_mode"] == "self_consumption" + run_async(api.select_event("select.predbat_teslemetry_operation_mode", "self_consumption")) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/operation")] + assert len(posts) == 1 + + +def test_teslemetry_dedupe_tariff_resends_on_window_advance(): + """A tariff rebuild whose live rate is UNCHANGED but whose export_now ON_PEAK window has + advanced (a later base local time crosses into the next 30-minute-aligned window) must still + re-POST - the dedupe signature is the full built tariff body (which embeds the window), not + just the mode string, so a window-only move is not silently skipped. + + Would be RED under a hypothetical mode-string-keyed signature: if `set_tariff` deduped on + `_apply_command("tariff", mode, ...)` instead of the built-tariff JSON, both calls below pass + the identical mode "export_now", so `self._last_sent["tariff"] == "export_now"` would already + match on the second call and the POST would be skipped (posts == 1) even though the device's + ON_PEAK window moved from 14:30-15:30 to 15:00-16:00 - the customer would be left on a STALE, + now-expired export window. GREEN as-built: the signature is + `json.dumps(build_tariff(mode), sort_keys=True)`, which differs between the two builds purely + because fromHour/fromMinute/toHour/toMinute moved, so both calls POST. + """ + from datetime import datetime, timezone + from types import SimpleNamespace + + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + # Live import/export rates held FIXED across both calls (same minutes_now, same rate tables) so + # only the base local time - and therefore the window - differs between the two builds. + api.base = SimpleNamespace( + now_utc=datetime(2026, 7, 2, 14, 40, tzinfo=timezone.utc), + local_tz=timezone.utc, + minutes_now=600, + rate_import={600: 28.0}, + rate_export={600: 15.0}, + ) + run_async(api.set_tariff("export_now")) + + # Advance base local time into the NEXT 30-minute window; rates untouched. + api.base.now_utc = datetime(2026, 7, 2, 15, 10, tzinfo=timezone.utc) + run_async(api.set_tariff("export_now")) + + posts = [req for req in api.requests_made if req[0] == "POST"] + assert len(posts) == 2 + first_tariff = posts[0][2]["tou_settings"]["tariff_content_v2"] + second_tariff = posts[1][2]["tou_settings"]["tariff_content_v2"] + first_window = first_tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"][0] + second_window = second_tariff["seasons"]["AllYear"]["tou_periods"]["ON_PEAK"]["periods"][0] + assert (first_window["fromHour"], first_window["fromMinute"]) == (14, 30) + assert (second_window["fromHour"], second_window["fromMinute"]) == (15, 0) + # Confirm the live rate really was constant - only the window differs between the two builds. + first_on_peak_sell = first_tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"]["ON_PEAK"] + second_on_peak_sell = second_tariff["sell_tariff"]["energy_charges"]["AllYear"]["rates"]["ON_PEAK"] + assert first_on_peak_sell == second_on_peak_sell + + +def test_teslemetry_backup_reserve_drift_correction_refreshes_cache_and_reasserts(): + """A site_info poll reporting a device backup_reserve_percent that differs from what Predbat + last sent must refresh the dedupe cache to the ACTUAL device value, so the next assertion of + Predbat's desired reserve is not skipped - mirrors + test_teslemetry_drift_correction_refreshes_cache_and_reasserts (operation_mode) but exercises + the backup_reserve refresh path in fetch_site_info, which previously had no direct test.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/backup"] = {"response": {"code": 201}} + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + assert api._last_sent["backup_reserve"] == 20 + # Device now reports backup_reserve_percent = 50, i.e. the user changed it externally. + drifted_site_info = {"response": {"nameplate_energy": 13500, "default_real_mode": "self_consumption", "backup_reserve_percent": 50}} + api.mock_responses["/api/1/energy_sites/123456/site_info"] = drifted_site_info + run_async(api.fetch_site_info()) + assert api._last_sent["backup_reserve"] == 50 + # Predbat's desired reserve is still 20 - the drifted cache means this must re-POST. + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/backup")] + assert len(posts) == 2 + + +def test_teslemetry_backup_reserve_drift_correction_no_spurious_resend_when_matching(): + """When site_info reports the SAME backup_reserve Predbat last sent, the dedupe cache is + unchanged and the next assertion of that value is still skipped - no spurious re-send from the + drift-refresh itself.""" + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/backup"] = {"response": {"code": 201}} + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + assert api._last_sent["backup_reserve"] == 20 + # SITE_INFO's backup_reserve_percent is also 20 - no drift. + api.mock_responses["/api/1/energy_sites/123456/site_info"] = SITE_INFO + run_async(api.fetch_site_info()) + assert api._last_sent["backup_reserve"] == 20 + run_async(api.number_event("number.predbat_teslemetry_backup_reserve", 20)) + posts = [req for req in api.requests_made if req[0] == "POST" and req[1].endswith("/backup")] + assert len(posts) == 1 + + +def test_teslemetry_reconcile_forces_write_even_if_cache_preseeded(): + """reconcile_on_start's corrective writes must not be silently skipped by the dedupe cache even if + it already holds a value matching what recovery is about to send (e.g. a future cache-preseed + feature) - the recovery path always forces the POST regardless of cache state.""" + import json as _json + + api = MockTeslemetryAPI() + api.mock_responses["/api/1/energy_sites/123456/tariff_rate"] = TARIFF_RATE_EXPORT_NOW + api.mock_responses["/api/1/energy_sites/123456/time_of_use_settings"] = {"response": {"code": 201}} + api.mock_responses["/api/1/energy_sites/123456/grid_import_export"] = {"response": {"code": 201}} + # Pre-seed the dedupe cache as if "normal"/"never" had already been confirmed-sent - a naive + # dedupe would treat the recovery writes below as no-ops and skip them entirely. + api._last_sent["tariff"] = _json.dumps(api.build_tariff("normal"), sort_keys=True) + api._last_sent["export_rule"] = "never" + run_async(api.reconcile_on_start()) + assert any(req[1].endswith("/time_of_use_settings") for req in api.requests_made if req[0] == "POST") + assert any(req[1].endswith("/grid_import_export") for req in api.requests_made if req[0] == "POST") + assert api.entity_states["select.predbat_teslemetry_tariff_mode"] == "normal" + assert api.entity_states["select.predbat_teslemetry_allow_export"] == "never" + + +def test_teslemetry(my_predbat=None): + """Run all Teslemetry component tests (registry entry point). + + Args: + my_predbat: Unused; accepted for compatibility with the TEST_REGISTRY calling convention in unit_test.py. + """ + test_teslemetry_entity_names() + test_teslemetry_live_status_publishes_sensors() + test_teslemetry_site_info_publishes_soc_max() + test_teslemetry_site_info_seeds_control_entity_states() + test_teslemetry_energy_today_publishes_kwh() + test_teslemetry_request_auth_failure() + test_teslemetry_request_rate_limit_retry() + test_teslemetry_request_success_clears_auth_flag() + test_teslemetry_run_unconfigured_returns_false() + test_teslemetry_run_first_success_returns_true() + test_teslemetry_run_auth_failed_only_probes_live_status() + test_teslemetry_select_operation_mode() + test_teslemetry_select_failure_keeps_state() + test_teslemetry_command_success_on_low_response_code() + test_teslemetry_command_failure_on_application_error_key() + test_teslemetry_command_failure_on_high_response_code() + test_teslemetry_command_none_response_is_failure() + test_teslemetry_number_backup_reserve() + test_teslemetry_switch_grid_charging() + test_teslemetry_select_export_rule() + test_teslemetry_site_info_latch() + test_teslemetry_build_tariff_export_now() + test_teslemetry_build_tariff_normal_flat() + test_teslemetry_set_tariff_posts_tou_settings() + test_teslemetry_reconcile_on_start_device_marker_restores_normal() + test_teslemetry_reconcile_on_start_device_normal_is_noop() + test_teslemetry_reconcile_on_start_read_failure_skips_without_crash() + test_teslemetry_reconcile_on_start_no_code_in_response_skips() + test_teslemetry_reconcile_on_start_read_only_mode_skips_writes() + test_teslemetry_reconcile_on_start_ignores_boot_default_read_only_attribute() + test_teslemetry_build_tariff_export_now_midnight_wrap() + test_teslemetry_build_tariff_export_now_midnight_exact() + test_teslemetry_build_tariff_sell_clamp_above_export_rate() + test_teslemetry_reconcile_on_start_partial_failure() + test_teslemetry_build_tariff_uses_site_local_time_not_utc() + test_teslemetry_reconcile_latch_survives_auth_failed_first_cycle() + test_teslemetry_reconcile_latch_runs_once_on_healthy_boot() + test_teslemetry_site_info_returns_false_without_nameplate_so_soc_max_retries() + test_teslemetry_run_site_info_latch_stays_false_without_nameplate() + test_teslemetry_energy_today_requests_kind_and_period() + test_teslemetry_select_event_preserves_control_attributes() + test_teslemetry_number_event_guards_null_value() + test_teslemetry_dedupe_operation_mode_skips_repeat_post() + test_teslemetry_dedupe_operation_mode_resends_on_change() + test_teslemetry_dedupe_failed_post_not_cached_so_retries() + test_teslemetry_dedupe_tariff_identical_body_skips_repeat_post() + test_teslemetry_dedupe_tariff_resends_when_rates_change() + test_teslemetry_drift_correction_refreshes_cache_and_reasserts() + test_teslemetry_drift_correction_no_spurious_resend_when_matching() + test_teslemetry_dedupe_tariff_resends_on_window_advance() + test_teslemetry_backup_reserve_drift_correction_refreshes_cache_and_reasserts() + test_teslemetry_backup_reserve_drift_correction_no_spurious_resend_when_matching() + test_teslemetry_reconcile_forces_write_even_if_cache_preseeded() + print("**** Teslemetry tests passed ****") + return 0 diff --git a/apps/predbat/unit_test.py b/apps/predbat/unit_test.py index bcb03096f..8738d1230 100644 --- a/apps/predbat/unit_test.py +++ b/apps/predbat/unit_test.py @@ -50,6 +50,7 @@ from tests.test_saving_session import test_saving_session, test_saving_session_null_octopoints, test_saving_session_notify_config, test_saving_session_default_rate, test_saving_session_axle_conflict, test_saving_session_auto_join_toggle from tests.test_secrets import run_secrets_tests from tests.test_ge_cloud import test_ge_cloud +from tests.test_teslemetry import test_teslemetry from tests.test_compare import test_compare from tests.test_gateway import run_gateway_tests from tests.test_axle import test_axle @@ -289,6 +290,7 @@ def main(): ("balance_inverters", run_balance_inverters_tests, "Balance inverters tests", False), # GE Cloud unit tests ("ge_cloud", test_ge_cloud, "GE Cloud comprehensive tests (API, devices, EVC, inverter ops, events, publishing, config, downloads, cache)", False), + ("teslemetry", test_teslemetry, "Teslemetry Tesla Powerwall component tests (data path, control, tariff)", False), ("integer_config", test_integer_config_entities, "Integer config entities tests", False), ("validate_config", test_validate_config, "APPS_SCHEMA validator tests (string types, sensor boolean states)", False), ("expose_config_integer", test_expose_config_preserves_integer, "Expose config preserves integer tests", False),