From 8a7a55b1245b5abc4c40b13dca467fad3cdc932b Mon Sep 17 00:00:00 2001 From: aunefyren Date: Thu, 17 Sep 2026 13:12:28 +0200 Subject: [PATCH] feature: statistics & history backfill --- README.md | 21 +- custom_components/current/api.py | 9 +- custom_components/current/coordinator.py | 55 +++ custom_components/current/manifest.json | 4 +- custom_components/current/sensor.py | 25 ++ .../current/statistics_import.py | 264 +++++++++++ dev/probe_api.py | 17 + hacs.json | 2 +- tests/conftest.py | 55 ++- tests/test_sensor.py | 6 +- tests/test_statistics_import.py | 413 ++++++++++++++++++ 11 files changed, 859 insertions(+), 12 deletions(-) create mode 100644 custom_components/current/statistics_import.py create mode 100644 tests/test_statistics_import.py diff --git a/README.md b/README.md index 12af4d4..f66515e 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,6 @@ # CURRENT EV Charging ![GitHub Release](https://img.shields.io/github/v/release/aunefyren/current?style=for-the-badge) -![GitHub Downloads (all assets, all releases)](https://img.shields.io/github/downloads/aunefyren/current/total?style=for-the-badge) ![GitHub issues](https://img.shields.io/github/issues/aunefyren/current?style=for-the-badge) ![GitHub Repo stars](https://img.shields.io/github/stars/aunefyren/current?style=for-the-badge) ![GitHub forks](https://img.shields.io/github/forks/aunefyren/current?style=for-the-badge) @@ -25,6 +24,7 @@ Must be added as a custom repository. - Live session monitoring (power, current, energy, charging duration) - Charger status: `Available`, `Charging`, `Standby` (car full/paused), `Unavailable` - Last session summary (energy and cost in account currency) +- Full charging history in the Energy dashboard, with costs - Charger controls: require authentication, permanent cable lock, restart - Multiple chargers supported — each appears as a separate device @@ -56,6 +56,25 @@ Each charger appears as its own device. The following entities are created per c
+## Energy dashboard + +The integration reads your whole charging history from CURRENT and writes it to Home Assistant's long-term statistics. Every charger gets two statistics: + +| Statistic | Unit | +|---|---| +| `current:charger__energy` | kWh | +| `current:charger__cost` | Account currency | + +This includes sessions from before the integration was installed and sessions that finished while Home Assistant was down. The history is read when Home Assistant starts and again whenever a session finishes or CURRENT revises one. + +To see the charger's usage in the Energy dashboard, add the energy statistic under **Settings → Dashboards → Energy → Individual devices**. Both statistics can also be shown with a **Statistics graph** card, for example charging costs per month. + +- CURRENT only reports a total for each session. Its energy and cost are spread evenly from when the car was plugged in until it was unplugged, so hourly values are an estimate, and so is how a session that runs past midnight is split between the days. Each session's total is exact. +- A session appears in the statistics once it has finished. +- Don't also add the **Session Energy** sensor to the Energy dashboard, or every charge is counted twice. + +
+ ## Installation 1. Add this repo to HACS as a custom repository diff --git a/custom_components/current/api.py b/custom_components/current/api.py index b761ee4..2aa77d9 100644 --- a/custom_components/current/api.py +++ b/custom_components/current/api.py @@ -173,14 +173,17 @@ async def stop_charging(self, box_id: str | int, session_id: str | int) -> dict: "GET", f"Commands/RemoteStop/{box_id}/{session_id}" ) - async def get_history(self, count: int = 5) -> dict: - """Return the most recent charging sessions and account totals.""" + async def get_history(self, count: int = 5, start_index: int = 0) -> dict: + """Return completed charging sessions, newest first, and account totals. + + `start_index` skips that many sessions, for paging further back. + """ data = await self._request_with_refresh( "GET", f"ChargingHistory/customers/{self._customer_id}", params={ "number": count, - "startIndex": 0, + "startIndex": start_index, "fromDateTimestamp": 0, "toDateTimestamp": 0, "calculateTotalPrice": "true", diff --git a/custom_components/current/coordinator.py b/custom_components/current/coordinator.py index 021e33a..913f745 100644 --- a/custom_components/current/coordinator.py +++ b/custom_components/current/coordinator.py @@ -1,5 +1,6 @@ """Data update coordinator for CURRENT.""" +import asyncio import logging import time from collections.abc import Awaitable @@ -13,6 +14,7 @@ from .api import AuthError, CannotConnectError, CurrentApiClient from .const import DOMAIN, SCAN_INTERVAL_ACTIVE, SCAN_INTERVAL_IDLE +from .statistics_import import async_fetch_all_sessions, async_import_statistics _LOGGER = logging.getLogger(__name__) @@ -37,6 +39,10 @@ def __init__( ) self.client = client self._fast_poll_until: float = 0 + self._statistics_lock = asyncio.Lock() + # What the latest history looked like when statistics were last + # imported, so the full history is only read again when it changes. + self._imported_history: tuple | None = None def start_fast_polling(self, duration: int = 120) -> None: """Poll faster for a while, so a start or stop shows up quickly.""" @@ -80,8 +86,57 @@ async def _async_update_data(self) -> dict[str, Any]: seconds=SCAN_INTERVAL_ACTIVE if ongoing else SCAN_INTERVAL_IDLE ) + self._schedule_statistics(history, chargers) + return { "ongoing": ongoing, "chargers": chargers, "history": history, } + + def _schedule_statistics(self, history: dict, chargers: list[dict]) -> None: + """Import statistics in the background when the history has changed. + + Every poll fetches the latest few sessions. Reading the whole history + takes several requests, so that only happens at startup and when one + of those sessions is new or has been revised. + """ + fingerprint = tuple( + ( + (item.get("Session") or {}).get("PK_ServiceSessionID"), + (item.get("Session") or {}).get("SessionEnd"), + item.get("TotalkWH"), + item.get("TotalPrice"), + ) + for item in (history or {}).get("List") or [] + ) + if fingerprint == self._imported_history: + return + if self._statistics_lock.locked(): + _LOGGER.debug("Statistics import still running, skipping this cycle") + return + + charger_names = { + c["FK_ChargePointID"]: c["Name"] + for c in chargers + if c.get("FK_ChargePointID") is not None and c.get("Name") + } + self.config_entry.async_create_background_task( + self.hass, + self._async_import_statistics(fingerprint, charger_names), + name=f"{DOMAIN}_statistics", + ) + + async def _async_import_statistics( + self, fingerprint: tuple, charger_names: dict[int, str] + ) -> None: + """Read the whole charging history and write it to statistics.""" + async with self._statistics_lock: + try: + sessions = await async_fetch_all_sessions(self.client) + except (AuthError, CannotConnectError) as err: + # The regular poll reports these; try again on the next one. + _LOGGER.warning("Could not read charging history: %s", err) + return + async_import_statistics(self.hass, sessions, charger_names) + self._imported_history = fingerprint diff --git a/custom_components/current/manifest.json b/custom_components/current/manifest.json index 3aa80a7..a438320 100644 --- a/custom_components/current/manifest.json +++ b/custom_components/current/manifest.json @@ -3,10 +3,10 @@ "name": "CURRENT EV Charging", "codeowners": ["@aunefyren"], "config_flow": true, - "dependencies": [], + "dependencies": ["recorder"], "documentation": "https://github.com/aunefyren/current", "iot_class": "cloud_polling", "issue_tracker": "https://github.com/aunefyren/current/issues", "requirements": [], - "version": "1.2.0" + "version": "2.0.0" } diff --git a/custom_components/current/sensor.py b/custom_components/current/sensor.py index 63b232d..05afead 100644 --- a/custom_components/current/sensor.py +++ b/custom_components/current/sensor.py @@ -3,6 +3,7 @@ import logging from collections.abc import Callable from dataclasses import dataclass +from datetime import datetime from typing import Any from homeassistant.components.sensor import ( @@ -26,6 +27,7 @@ from .const import DOMAIN from .coordinator import CurrentCoordinator +from .statistics_import import parse_time _LOGGER = logging.getLogger(__name__) @@ -49,6 +51,19 @@ def _get_history_sessions(data: dict) -> list: return (data.get("history") or {}).get("List") or [] +def _get_last_session_end(data: dict) -> datetime | None: + """Return when the last completed session ended. + + The last session sensors are totals that start over with every session, so + this is their last_reset. Without it, the recorder would read a smaller + session following a bigger one as negative energy. + """ + sessions = _get_history_sessions(data) + if not sessions: + return None + return parse_time((sessions[0].get("Session") or {}).get("SessionEnd")) + + def _get_live(data: dict) -> dict: """Return the charger's live readings. @@ -74,6 +89,7 @@ class CurrentSensorEntityDescription(SensorEntityDescription): value_fn: Callable[[dict[str, Any]], Any] unit_fn: Callable[[dict[str, Any]], str | None] | None = None attributes_fn: Callable[[dict[str, Any]], dict[str, Any]] | None = None + last_reset_fn: Callable[[dict[str, Any]], datetime | None] | None = None SENSOR_DESCRIPTIONS: tuple[CurrentSensorEntityDescription, ...] = ( @@ -143,6 +159,7 @@ class CurrentSensorEntityDescription(SensorEntityDescription): "TotalPrice" ), unit_fn=lambda data: (data.get("chargers") or [{}])[0].get("Currency"), + last_reset_fn=_get_last_session_end, ), CurrentSensorEntityDescription( key="last_session_energy", @@ -151,6 +168,7 @@ class CurrentSensorEntityDescription(SensorEntityDescription): device_class=SensorDeviceClass.ENERGY, state_class=SensorStateClass.TOTAL, value_fn=lambda data: (_get_history_sessions(data) or [{}])[0].get("TotalkWH"), + last_reset_fn=_get_last_session_end, ), ) @@ -234,6 +252,13 @@ def native_value(self) -> Any: """Return the value for this charger.""" return self.entity_description.value_fn(self._filtered_data()) + @property + def last_reset(self) -> datetime | None: + """Return when a per-session total last started over.""" + if self.entity_description.last_reset_fn is None: + return None + return self.entity_description.last_reset_fn(self._filtered_data()) + @property def extra_state_attributes(self) -> dict[str, Any] | None: """Return extra detail for this charger, where the sensor has any.""" diff --git a/custom_components/current/statistics_import.py b/custom_components/current/statistics_import.py new file mode 100644 index 0000000..3224fb8 --- /dev/null +++ b/custom_components/current/statistics_import.py @@ -0,0 +1,264 @@ +"""Write completed charging sessions into Home Assistant long-term statistics. + +The live session sensors only know about a session while Home Assistant is +watching it, and record energy at the moment it was polled. CURRENT's charging +history knows every session the account ever had, with its start, end, energy +and cost. External statistics let us write those against the hours they +happened in, including sessions from before the integration was installed or +while Home Assistant was down. + +The history only reports a total per session, not how it was spread over the +session, so each session's energy and cost are divided evenly over the time +from its start to its end. +""" + +from __future__ import annotations + +import logging +import re +from collections import defaultdict +from dataclasses import dataclass +from datetime import datetime, timedelta +from typing import Any + +from homeassistant.components.recorder.models import ( + StatisticData, + StatisticMeanType, + StatisticMetaData, +) +from homeassistant.components.recorder.statistics import ( + async_add_external_statistics, +) +from homeassistant.const import UnitOfEnergy +from homeassistant.core import HomeAssistant +from homeassistant.util import dt as dt_util + +from .api import CurrentApiClient +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + +# CURRENT hands out at most 50 sessions a page, however many are asked for. +HISTORY_PAGE_SIZE = 50 +# A safety stop, should CURRENT ignore startIndex and keep returning new ids. +MAX_HISTORY_PAGES = 100 + +ENERGY = "energy" +COST = "cost" + +ONE_HOUR = timedelta(hours=1) + + +@dataclass(frozen=True, kw_only=True) +class CompletedSession: + """The parts of a finished session that statistics are built from.""" + + session_id: Any + charge_point_id: int + start: datetime + end: datetime + energy: float | None + cost: float | None + currency: str | None + + +def statistic_id_for(charge_point_id: int | str, kind: str) -> str: + """Build the external statistic id for one charger's energy or cost. + + Recorder's VALID_STATISTIC_ID rejects anything outside [a-z0-9_], any + double underscore, and leading or trailing underscores. + """ + slug = re.sub(r"[^a-z0-9]+", "_", str(charge_point_id).lower()).strip("_") + return f"{DOMAIN}:charger_{slug or 'unknown'}_{kind}" + + +def _as_float(value: object) -> float | None: + """Coerce an API number, tolerating nulls and unexpected types.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + return float(value) + + +def parse_time(value: object) -> datetime | None: + """Parse a CURRENT timestamp into UTC. + + The history marks its timestamps as UTC. The active sessions endpoint + leaves the zone off and means local time: a session probed at 13:08 CEST + reported starting at 11:53 after charging for 74 minutes. Timestamps + without a zone are therefore read in Home Assistant's time zone. + """ + if not isinstance(value, str): + return None + parsed = dt_util.parse_datetime(value) + if parsed is None: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=dt_util.get_default_time_zone()) + return dt_util.as_utc(parsed) + + +def parse_session(item: dict[str, Any]) -> CompletedSession | None: + """Read one history item, or return None if it is not a finished session.""" + session = item.get("Session") or {} + charge_point_id = item.get("ChargePointID") or session.get("ChargingPointID") + start = parse_time(session.get("SessionStart")) + end = parse_time(session.get("SessionEnd")) + if charge_point_id is None or start is None or end is None: + return None + + energy = _as_float(item.get("TotalkWH")) + if energy is None: + energy = _as_float(session.get("TotalkWh")) + + return CompletedSession( + session_id=session.get("PK_ServiceSessionID") or (charge_point_id, start), + charge_point_id=charge_point_id, + start=start, + end=end, + energy=energy, + cost=_as_float(item.get("TotalPrice")), + currency=item.get("Currency"), + ) + + +def spread_over_hours( + start: datetime, end: datetime, amount: float +) -> dict[datetime, float]: + """Divide an amount over the hours between start and end. + + Each hour gets the share of the amount that its overlap with the session + is of the whole session. A session with no length lands in its start hour. + """ + first_hour = start.replace(minute=0, second=0, microsecond=0) + if end <= start: + return {first_hour: amount} + + length = (end - start).total_seconds() + shares: dict[datetime, float] = {} + hour = first_hour + while hour < end: + overlap = (min(hour + ONE_HOUR, end) - max(hour, start)).total_seconds() + shares[hour] = amount * overlap / length + hour += ONE_HOUR + return shares + + +async def async_fetch_all_sessions(client: CurrentApiClient) -> list[CompletedSession]: + """Page through the whole charging history, oldest session first.""" + sessions: dict[Any, CompletedSession] = {} + offset = 0 + + for _ in range(MAX_HISTORY_PAGES): + result = await client.get_history(count=HISTORY_PAGE_SIZE, start_index=offset) + result = result or {} + items = result.get("List") or [] + offset += len(items) + + added = 0 + for item in items: + parsed = parse_session(item) + if parsed is not None and parsed.session_id not in sessions: + sessions[parsed.session_id] = parsed + added += 1 + + if not added: + break + # TotalOrders counts every session, while the other totals only cover + # the page returned. Without it, a short page is taken as the last, + # though CURRENT may also cap how many sessions a page holds. + total = result.get("TotalOrders") + if isinstance(total, int) and not isinstance(total, bool): + if offset >= total: + break + elif len(items) < HISTORY_PAGE_SIZE: + break + else: + _LOGGER.warning( + "Stopped reading charging history after %d pages", MAX_HISTORY_PAGES + ) + + return sorted(sessions.values(), key=lambda s: s.start) + + +def _build_statistics( + sessions: list[CompletedSession], value_of: str +) -> list[StatisticData]: + """Turn sessions into hourly buckets carrying a running total.""" + per_hour: dict[datetime, float] = defaultdict(float) + for session in sessions: + amount = getattr(session, value_of) + if amount is None: + continue + for hour, share in spread_over_hours( + session.start, session.end, amount + ).items(): + per_hour[hour] += share + + total = 0.0 + statistics: list[StatisticData] = [] + for hour in sorted(per_hour): + total += per_hour[hour] + statistics.append(StatisticData(start=hour, state=total, sum=total)) + return statistics + + +def async_import_statistics( + hass: HomeAssistant, + sessions: list[CompletedSession], + charger_names: dict[int, str], +) -> None: + """Write energy and cost statistics for every charger in the history. + + The whole history is rewritten each time. Writing a bucket again replaces + it, so a revised price or energy reading is corrected instead of counted + twice. + """ + by_charger: dict[int, list[CompletedSession]] = defaultdict(list) + for session in sessions: + by_charger[session.charge_point_id].append(session) + + for charge_point_id, charger_sessions in by_charger.items(): + name = charger_names.get(charge_point_id) or f"Charger {charge_point_id}" + + energy = _build_statistics(charger_sessions, ENERGY) + if energy: + async_add_external_statistics( + hass, + StatisticMetaData( + mean_type=StatisticMeanType.NONE, + has_sum=True, + name=f"{name} energy", + source=DOMAIN, + statistic_id=statistic_id_for(charge_point_id, ENERGY), + unit_class="energy", + unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR, + ), + energy, + ) + + # Costs are in the account's currency; take the latest one given. + currency = next( + (s.currency for s in reversed(charger_sessions) if s.currency), None + ) + cost = _build_statistics(charger_sessions, COST) + if cost: + async_add_external_statistics( + hass, + StatisticMetaData( + mean_type=StatisticMeanType.NONE, + has_sum=True, + name=f"{name} cost", + source=DOMAIN, + statistic_id=statistic_id_for(charge_point_id, COST), + unit_class=None, + unit_of_measurement=currency, + ), + cost, + ) + + _LOGGER.debug( + "Imported %d sessions for charger %s into %d hourly statistics", + len(charger_sessions), + charge_point_id, + len(energy), + ) diff --git a/dev/probe_api.py b/dev/probe_api.py index b5ea21e..470f42b 100644 --- a/dev/probe_api.py +++ b/dev/probe_api.py @@ -383,6 +383,23 @@ def call(label: str, method: str, path: str, **kwargs) -> dict: token=access_token, params={**history_params, "number": 20}, ) + # The statistics import pages through the whole history. startIndex should + # skip that many sessions, so this page should repeat sessions 6-10 of + # history.20; and a large page shows whether CURRENT caps its size. + call( + "history.page2", + "GET", + f"ChargingHistory/customers/{customer_id}", + token=access_token, + params={**history_params, "startIndex": 5}, + ) + call( + "history.100", + "GET", + f"ChargingHistory/customers/{customer_id}", + token=access_token, + params={**history_params, "number": 100}, + ) # What a rejected token looks like: 401 or 403, and with what body. This # decides when the client refreshes versus gives up. diff --git a/hacs.json b/hacs.json index ec3fac1..bd7f607 100644 --- a/hacs.json +++ b/hacs.json @@ -3,5 +3,5 @@ "content_in_root": false, "country": ["NO"], "render_readme": true, - "homeassistant": "2024.11.0" + "homeassistant": "2026.2.0" } diff --git a/tests/conftest.py b/tests/conftest.py index 0bb5e44..8add9ea 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -9,8 +9,10 @@ import copy import json +import logging import pathlib from collections.abc import Iterator +from datetime import datetime, timedelta from typing import Any from unittest.mock import patch @@ -45,10 +47,19 @@ FIXTURES = pathlib.Path(__file__).parent / "fixtures" API_PREFIX = f"{API_BASE_URL}/v2/" +# pytest-homeassistant-custom-component turns SQLAlchemy's statement logging on +# at INFO, which buries the test results under the recorder's inserts. +logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING) + @pytest.fixture(autouse=True) -def auto_enable_custom_integrations(enable_custom_integrations): - """Let every test load this integration.""" +def auto_enable_custom_integrations(recorder_mock, enable_custom_integrations): + """Let every test load this integration. + + `recorder_mock` is requested first on purpose: the integration depends on + the recorder, and the recorder's database fixture asserts that it is built + before `hass` exists. + """ return @@ -107,6 +118,8 @@ def __init__(self, responses: dict[str, Any]) -> None: "datas" ] self.history: dict[str, Any] = responses["history"]["Result"] + # Largest page the endpoint hands out, whatever `number` asks for. + self.history_page_limit: int | None = None # Tokens the server currently accepts. self.valid_tokens = {MOCK_ACCESS_TOKEN} @@ -162,6 +175,38 @@ def add_second_charger(self) -> None: ) self.history = {**self.history, "List": [item, *self.history["List"]]} + def set_history_length(self, count: int) -> None: + """Replace the history with `count` sessions, one a day, newest first. + + Each is a copy of the first captured session, moved back a day at a + time, with its own id. + """ + template = self.history["List"][0] + start = datetime.fromisoformat(template["Session"]["SessionStart"]) + end = datetime.fromisoformat(template["Session"]["SessionEnd"]) + items = [] + for n in range(count): + item = copy.deepcopy(template) + item["Session"].update( + PK_ServiceSessionID=100000 + n, + SessionStart=(start - timedelta(days=n)).isoformat(), + SessionEnd=(end - timedelta(days=n)).isoformat(), + ) + items.append(item) + self.history = {**self.history, "List": items, "TotalOrders": count} + + def history_requests(self) -> list[dict[str, Any]]: + """Return the history requests that were made.""" + return [r for r in self.requests if r["path"].startswith("ChargingHistory/")] + + def history_response(self, params: dict[str, Any]) -> dict[str, Any]: + """Return one page of history, as `number` and `startIndex` ask.""" + number = int(params.get("number", 5)) + if self.history_page_limit is not None: + number = min(number, self.history_page_limit) + start = int(params.get("startIndex", 0)) + return {**self.history, "List": self.history["List"][start : start + number]} + def commands(self) -> list[dict[str, Any]]: """Return the charger commands that were sent.""" return [r for r in self.requests if r["path"].startswith("Commands/")] @@ -206,7 +251,9 @@ def request(self, method: str, url: str, **kwargs: Any) -> FakeResponse: if method == "GET" and path == f"sessions/user/{MOCK_USER_ID}/active": return FakeResponse(200, {"Result": self.sessions}) if method == "GET" and path == f"ChargingHistory/customers/{MOCK_CUSTOMER_ID}": - return FakeResponse(200, {"Result": self.history}) + return FakeResponse( + 200, {"Result": self.history_response(kwargs.get("params") or {})} + ) if path.startswith("Commands/"): if self.command_status is not None: return FakeResponse(self.command_status, {}) @@ -290,7 +337,7 @@ async def setup_entry(hass: HomeAssistant, entry: MockConfigEntry) -> None: """Add the entry to hass and set it up.""" entry.add_to_hass(hass) assert await hass.config_entries.async_setup(entry.entry_id) - await hass.async_block_till_done() + await hass.async_block_till_done(wait_background_tasks=True) def entity_id( diff --git a/tests/test_sensor.py b/tests/test_sensor.py index 0f3dbbd..e913162 100644 --- a/tests/test_sensor.py +++ b/tests/test_sensor.py @@ -45,7 +45,11 @@ async def test_idle( cost = state(hass, "last_session_cost") assert float(cost.state) == 82.18 assert cost.attributes["unit_of_measurement"] == "NOK" - assert float(state(hass, "last_session_energy").state) == 58.702 + energy = state(hass, "last_session_energy") + assert float(energy.state) == 58.702 + # Totals start over with each session, when the last one ended. + for sensor in (cost, energy): + assert sensor.attributes["last_reset"] == "2026-09-15T05:29:39.780000+00:00" async def test_charging( diff --git a/tests/test_statistics_import.py b/tests/test_statistics_import.py new file mode 100644 index 0000000..f6e4c02 --- /dev/null +++ b/tests/test_statistics_import.py @@ -0,0 +1,413 @@ +"""Tests for the long-term statistics import.""" + +from __future__ import annotations + +import re +from datetime import UTC, datetime +from unittest.mock import patch + +import pytest +from homeassistant.components.recorder import get_instance +from homeassistant.components.recorder.statistics import ( + get_metadata, + statistics_during_period, +) +from homeassistant.core import HomeAssistant +from pytest_homeassistant_custom_component.common import MockConfigEntry +from pytest_homeassistant_custom_component.components.recorder.common import ( + async_wait_recording_done, +) + +from custom_components.current.api import CannotConnectError, CurrentApiClient +from custom_components.current.const import DOMAIN +from custom_components.current.statistics_import import ( + COST, + ENERGY, + HISTORY_PAGE_SIZE, + async_fetch_all_sessions, + async_import_statistics, + parse_session, + parse_time, + spread_over_hours, + statistic_id_for, +) + +from .conftest import CurrentApiMock, FakeSession, setup_entry +from .const import ( + MOCK_ACCESS_TOKEN, + MOCK_CHARGE_POINT_ID, + MOCK_CUSTOMER_ID, + MOCK_REFRESH_TOKEN, + MOCK_USER_ID, + SECOND_CHARGE_POINT_ID, +) + +# The recorder's own rule for external statistic ids. +VALID_STATISTIC_ID = re.compile(r"^(?!.+__)(?!_)[\da-z_]+(? datetime: + """Build a UTC datetime.""" + return datetime(*args, tzinfo=UTC) + + +# -- ids and parsing ------------------------------------------------------ + + +@pytest.mark.parametrize("charge_point_id", [3001, "3001", "AB-12", "--", ""]) +@pytest.mark.parametrize("kind", [ENERGY, COST]) +def test_statistic_ids_are_accepted_by_recorder( + charge_point_id: int | str, kind: str +) -> None: + """Every id we can generate must satisfy the recorder's pattern.""" + statistic_id = statistic_id_for(charge_point_id, kind) + assert VALID_STATISTIC_ID.match(statistic_id), statistic_id + assert statistic_id.startswith(f"{DOMAIN}:") + + +def test_statistic_ids_are_distinct() -> None: + """Energy and cost, on different chargers, must not collide.""" + ids = { + statistic_id_for(charger, kind) + for charger in (MOCK_CHARGE_POINT_ID, SECOND_CHARGE_POINT_ID) + for kind in (ENERGY, COST) + } + assert len(ids) == 4 + + +def test_parse_session(api: CurrentApiMock) -> None: + """A captured history item is read into a completed session.""" + session = parse_session(api.history["List"][0]) + + assert session is not None + assert session.charge_point_id == MOCK_CHARGE_POINT_ID + assert session.start == utc(2026, 9, 14, 9, 9, 15, 103000) + assert session.end == utc(2026, 9, 15, 5, 29, 39, 780000) + assert session.energy == 58.702 + assert session.cost == 82.18 + assert session.currency == "NOK" + + +async def test_parse_time_without_zone_is_local(hass: HomeAssistant) -> None: + """A timestamp without a zone is local time, one with a zone is kept.""" + await hass.config.async_set_time_zone("Europe/Oslo") + assert parse_time("2026-09-17T11:53:41.903") == utc(2026, 9, 17, 9, 53, 41, 903000) + assert parse_time("2026-09-17T11:53:41.903Z") == utc( + 2026, 9, 17, 11, 53, 41, 903000 + ) + assert parse_time(None) is None + assert parse_time("not a time") is None + + +def test_parse_session_skips_unfinished(api: CurrentApiMock) -> None: + """A session without an end is still going, and is left out.""" + item = api.history["List"][0] + item["Session"]["SessionEnd"] = None + assert parse_session(item) is None + + +def test_parse_session_falls_back_to_session_energy(api: CurrentApiMock) -> None: + """The session's own energy is used when the item has none.""" + item = api.history["List"][0] + item["TotalkWH"] = None + session = parse_session(item) + assert session is not None + assert session.energy == item["Session"]["TotalkWh"] + + +# -- spreading over hours ------------------------------------------------- + + +def test_spread_within_one_hour() -> None: + """A session inside one hour puts everything in that hour.""" + assert spread_over_hours(utc(2026, 9, 1, 10, 5), utc(2026, 9, 1, 10, 50), 7.0) == { + utc(2026, 9, 1, 10): 7.0 + } + + +def test_spread_by_overlap() -> None: + """Each hour gets its share of the time the session covered.""" + # 30 minutes, 60 minutes, 30 minutes. + shares = spread_over_hours(utc(2026, 9, 1, 10, 30), utc(2026, 9, 1, 12, 30), 20.0) + assert shares == pytest.approx( + { + utc(2026, 9, 1, 10): 5.0, + utc(2026, 9, 1, 11): 10.0, + utc(2026, 9, 1, 12): 5.0, + } + ) + + +def test_spread_ending_on_the_hour() -> None: + """A session ending exactly on the hour does not touch the next hour.""" + shares = spread_over_hours(utc(2026, 9, 1, 10), utc(2026, 9, 1, 12), 4.0) + assert shares == pytest.approx({utc(2026, 9, 1, 10): 2.0, utc(2026, 9, 1, 11): 2.0}) + + +def test_spread_without_length() -> None: + """A session that ends when it starts lands in its start hour.""" + moment = utc(2026, 9, 1, 10, 15) + assert spread_over_hours(moment, moment, 3.0) == {utc(2026, 9, 1, 10): 3.0} + + +def test_spread_keeps_the_total(api: CurrentApiMock) -> None: + """Spreading never adds or loses energy.""" + for item in api.history["List"]: + session = parse_session(item) + assert session is not None + shares = spread_over_hours(session.start, session.end, session.energy) + assert sum(shares.values()) == pytest.approx(session.energy) + + +# -- reading the history -------------------------------------------------- + + +@pytest.fixture +def client(session: FakeSession) -> CurrentApiClient: + """Return a client over the fake endpoint.""" + return CurrentApiClient( + session=session, + access_token=MOCK_ACCESS_TOKEN, + refresh_token=MOCK_REFRESH_TOKEN, + customer_id=MOCK_CUSTOMER_ID, + user_id=MOCK_USER_ID, + ) + + +async def test_fetch_all_pages(client: CurrentApiClient, api: CurrentApiMock) -> None: + """The history is paged through until every session has been read.""" + api.set_history_length(2 * HISTORY_PAGE_SIZE + 10) + + sessions = await async_fetch_all_sessions(client) + + assert len(sessions) == 2 * HISTORY_PAGE_SIZE + 10 + assert [r["params"]["startIndex"] for r in api.history_requests()] == [ + 0, + HISTORY_PAGE_SIZE, + 2 * HISTORY_PAGE_SIZE, + ] + # Oldest first, which is what running totals are built in. + assert [s.start for s in sessions] == sorted(s.start for s in sessions) + + +async def test_fetch_with_capped_pages( + client: CurrentApiClient, api: CurrentApiMock +) -> None: + """A server returning fewer sessions than asked for is paged by what it gave.""" + api.set_history_length(45) + api.history_page_limit = 20 + + sessions = await async_fetch_all_sessions(client) + + assert len(sessions) == 45 + assert [r["params"]["startIndex"] for r in api.history_requests()] == [0, 20, 40] + + +async def test_fetch_like_the_real_account( + client: CurrentApiClient, api: CurrentApiMock +) -> None: + """123 sessions, 50 to a page as CURRENT caps it, take three requests.""" + api.set_history_length(123) + api.history_page_limit = 50 + + sessions = await async_fetch_all_sessions(client) + + assert len(sessions) == 123 + assert [r["params"]["startIndex"] for r in api.history_requests()] == [0, 50, 100] + + +async def test_fetch_stops_when_paging_is_ignored( + client: CurrentApiClient, api: CurrentApiMock +) -> None: + """If every page is the same, reading stops instead of looping.""" + api.set_history_length(HISTORY_PAGE_SIZE) + api.history["TotalOrders"] = 10 * HISTORY_PAGE_SIZE + api.history_response = lambda params: api.history # ignores startIndex + + sessions = await async_fetch_all_sessions(client) + + assert len(sessions) == HISTORY_PAGE_SIZE + assert len(api.history_requests()) == 2 + + +# -- writing statistics --------------------------------------------------- + + +async def read_rows(hass: HomeAssistant, statistic_id: str) -> list[dict]: + """Return every stored hourly bucket, oldest first.""" + await async_wait_recording_done(hass) + stats = await get_instance(hass).async_add_executor_job( + statistics_during_period, + hass, + utc(2020, 1, 1), + None, + {statistic_id}, + "hour", + None, + {"state", "sum"}, + ) + return stats.get(statistic_id, []) + + +async def read_metadata(hass: HomeAssistant, statistic_id: str) -> dict: + """Return the stored metadata of one statistic.""" + await async_wait_recording_done(hass) + found = await get_instance(hass).async_add_executor_job( + lambda: get_metadata(hass, statistic_ids={statistic_id}) + ) + return found[statistic_id][1] + + +def history_sessions(api: CurrentApiMock): + """Parse the fake endpoint's history, oldest first.""" + parsed = [parse_session(item) for item in api.history["List"]] + return sorted((s for s in parsed if s is not None), key=lambda s: s.start) + + +async def test_import_writes_energy_and_cost( + hass: HomeAssistant, api: CurrentApiMock +) -> None: + """Energy and cost end on the history's totals, in the right units.""" + sessions = history_sessions(api) + + async_import_statistics(hass, sessions, {MOCK_CHARGE_POINT_ID: "Test Charger"}) + + energy = await read_rows(hass, ENERGY_ID) + cost = await read_rows(hass, COST_ID) + assert energy[-1]["sum"] == pytest.approx(sum(s.energy for s in sessions)) + assert cost[-1]["sum"] == pytest.approx(sum(s.cost for s in sessions)) + + # Buckets start with the first session's hour and never run backwards. + assert ( + energy[0]["start"] + == sessions[0].start.replace(minute=0, second=0, microsecond=0).timestamp() + ) + sums = [row["sum"] for row in energy] + assert sums == sorted(sums) + + energy_meta = await read_metadata(hass, ENERGY_ID) + assert energy_meta["unit_of_measurement"] == "kWh" + assert energy_meta["name"] == "Test Charger energy" + assert energy_meta["source"] == DOMAIN + cost_meta = await read_metadata(hass, COST_ID) + assert cost_meta["unit_of_measurement"] == "NOK" + assert cost_meta["name"] == "Test Charger cost" + + +async def test_import_again_does_not_double_count( + hass: HomeAssistant, api: CurrentApiMock +) -> None: + """Importing the same history again leaves the totals as they were.""" + sessions = history_sessions(api) + + async_import_statistics(hass, sessions, {}) + first = await read_rows(hass, ENERGY_ID) + async_import_statistics(hass, sessions, {}) + second = await read_rows(hass, ENERGY_ID) + + assert [r["sum"] for r in second] == pytest.approx([r["sum"] for r in first]) + + +async def test_import_picks_up_revisions( + hass: HomeAssistant, api: CurrentApiMock +) -> None: + """A revised session replaces its old value rather than adding to it.""" + async_import_statistics(hass, history_sessions(api), {}) + before = (await read_rows(hass, COST_ID))[-1]["sum"] + + api.history["List"][2]["TotalPrice"] += 10.0 + async_import_statistics(hass, history_sessions(api), {}) + + assert (await read_rows(hass, COST_ID))[-1]["sum"] == pytest.approx(before + 10.0) + + +async def test_import_per_charger(hass: HomeAssistant, api: CurrentApiMock) -> None: + """Each charger gets statistics of its own sessions only.""" + api.add_second_charger() + sessions = history_sessions(api) + + async_import_statistics(hass, sessions, {}) + + for charger in (MOCK_CHARGE_POINT_ID, SECOND_CHARGE_POINT_ID): + rows = await read_rows(hass, statistic_id_for(charger, ENERGY)) + expected = sum(s.energy for s in sessions if s.charge_point_id == charger) + assert rows[-1]["sum"] == pytest.approx(expected) + # A charger with no name given still gets a readable one. + meta = await read_metadata(hass, statistic_id_for(SECOND_CHARGE_POINT_ID, ENERGY)) + assert meta["name"] == f"Charger {SECOND_CHARGE_POINT_ID} energy" + + +# -- the coordinator ------------------------------------------------------ + + +async def refresh(hass: HomeAssistant, entry: MockConfigEntry) -> None: + """Poll again and wait for any import that starts.""" + await hass.data[DOMAIN][entry.entry_id].async_refresh() + await hass.async_block_till_done(wait_background_tasks=True) + + +async def test_setup_imports_statistics( + hass: HomeAssistant, + patched_session: FakeSession, + api: CurrentApiMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Setting up reads the whole history into statistics.""" + await setup_entry(hass, mock_config_entry) + + rows = await read_rows(hass, ENERGY_ID) + assert rows[-1]["sum"] == pytest.approx( + sum(item["TotalkWH"] for item in api.history["List"]) + ) + meta = await read_metadata(hass, ENERGY_ID) + assert meta["name"] == "Test Charger energy" + + +async def test_history_only_reread_when_it_changes( + hass: HomeAssistant, + patched_session: FakeSession, + api: CurrentApiMock, + mock_config_entry: MockConfigEntry, +) -> None: + """Polls fetch the latest sessions, but only a change reads them all.""" + await setup_entry(hass, mock_config_entry) + after_setup = len(api.history_requests()) + # The poll's own request, and at least one page for the import. + assert after_setup >= 2 + + await refresh(hass, mock_config_entry) + assert len(api.history_requests()) == after_setup + 1 + + # A new session finishes. + item = api.history["List"][0] + new = {**item, "Session": {**item["Session"], "PK_ServiceSessionID": 99999}} + new["Session"]["SessionStart"] = "2026-09-16T10:00:00Z" + new["Session"]["SessionEnd"] = "2026-09-16T12:00:00Z" + api.history = {**api.history, "List": [new, *api.history["List"]]} + + await refresh(hass, mock_config_entry) + assert len(api.history_requests()) > after_setup + 2 + + rows = await read_rows(hass, ENERGY_ID) + assert rows[-1]["start"] == utc(2026, 9, 16, 11).timestamp() + + +async def test_failed_import_is_retried( + hass: HomeAssistant, + patched_session: FakeSession, + api: CurrentApiMock, + mock_config_entry: MockConfigEntry, +) -> None: + """If the history cannot be read, the next poll tries again.""" + with patch( + "custom_components.current.coordinator.async_fetch_all_sessions", + side_effect=CannotConnectError("fake failure"), + ): + await setup_entry(hass, mock_config_entry) + assert await read_rows(hass, ENERGY_ID) == [] + + await refresh(hass, mock_config_entry) + assert await read_rows(hass, ENERGY_ID)