From 2c7d64c1ea637ebe6281343c1cca2af8e8ff4136 Mon Sep 17 00:00:00 2001 From: Dan Raper Date: Sun, 31 Dec 2023 00:35:24 +0000 Subject: [PATCH] More unit tests and experimental charge state logic (#21) * Add testing for utility functions * Added experimental charge state monitoring --- custom_components/ohme/binary_sensor.py | 74 +++++++++++++++++++++++-- custom_components/ohme/utils.py | 8 +-- tests/test_utils.py | 59 ++++++++++++++++++++ 3 files changed, 132 insertions(+), 9 deletions(-) create mode 100644 tests/test_utils.py diff --git a/custom_components/ohme/binary_sensor.py b/custom_components/ohme/binary_sensor.py index d8bad2b..92af83d 100644 --- a/custom_components/ohme/binary_sensor.py +++ b/custom_components/ohme/binary_sensor.py @@ -1,6 +1,6 @@ """Platform for sensor integration.""" from __future__ import annotations - +import logging from homeassistant.components.binary_sensor import ( BinarySensorDeviceClass, BinarySensorEntity @@ -12,7 +12,9 @@ from .const import DOMAIN, DATA_COORDINATORS, COORDINATOR_CHARGESESSIONS, DATA_CLIENT from .coordinator import OhmeChargeSessionsCoordinator from .utils import charge_graph_in_slot +from time import time +_LOGGER = logging.getLogger(__name__) async def async_setup_entry( hass: core.HomeAssistant, @@ -97,6 +99,13 @@ def __init__( self._state = False self._client = client + # Cache the last power readings + self._last_reading = None + self._last_reading_in_slot = False + + # Allow a state override + self._override_until = None + self.entity_id = generate_entity_id( "binary_sensor.{}", "ohme_car_charging", hass=hass) @@ -115,13 +124,68 @@ def unique_id(self) -> str: @property def is_on(self) -> bool: - if self.coordinator.data and self.coordinator.data["power"]: - # Assume the car is actively charging if drawing over 0 watts - self._state = self.coordinator.data["power"]["watt"] > 0 + return self._state + + def _calculate_state(self) -> bool: + """Some trickery to get the charge state to update quickly.""" + # If we have overriden the state, return the current value until that time + if self._override_until and time() < self._override_until: + _LOGGER.debug("State overridden to False for 310s") + return self._state + + # We have passed override check, reset it + self._override_until = None + + power = self.coordinator.data["power"]["watt"] + + # No last reading to go off, use power draw based state only - this lags + if not self._last_reading: + _LOGGER.debug("Last reading not found, default to power > 0") + return power > 0 + + # Get power from last reading + lr_power = self._last_reading["power"]["watt"] + + # See if we are in a charge slot now and if we were for the last reading + in_charge_slot = charge_graph_in_slot( + self.coordinator.data['startTime'], self.coordinator.data['chargeGraph']['points']) + lr_in_charge_slot = self._last_reading_in_slot + + # Store this for next time + self._last_reading_in_slot = in_charge_slot + + # If: + # - Power has dropped by 40% since the last reading + # - Last reading we were in a charge slot + # - Now we are not in a charge slot + # The charge has stopped but the power reading is lagging. + if lr_power > 0 and power / lr_power < 0.6 and not in_charge_slot and lr_in_charge_slot: + _LOGGER.debug("Charge stop behaviour seen - overriding to False for 310 seconds") + self._override_until = time() + 310 # Override for 5 mins (and a bit) + return False + + # Its possible that this is the 'transitionary' reading - slots updated but not power + # Override _last_reading_in_slot and see what happens next time around + elif lr_power > 0 and not in_charge_slot and lr_in_charge_slot: + _LOGGER.debug("Possible transitionary reading. Treating as slot boundary in next tick.") + self._last_reading_in_slot = True + + # Fallback to the old way + return power > 0 + + @callback + def _handle_coordinator_update(self) -> None: + """Update data.""" + # If we have power info and the car is plugged in, calculate state. Otherwise, false + if self.coordinator.data and self.coordinator.data["power"] and self.coordinator.data['mode'] != "DISCONNECTED": + self._state = self._calculate_state() else: self._state = False - return self._state + self._last_reading = self.coordinator.data + self._last_updated = utcnow() + + self.async_write_ha_state() class PendingApprovalBinarySensor( diff --git a/custom_components/ohme/utils.py b/custom_components/ohme/utils.py index 06eed23..17159fc 100644 --- a/custom_components/ohme/utils.py +++ b/custom_components/ohme/utils.py @@ -10,10 +10,10 @@ def _format_charge_graph(charge_start, points): return [{"t": x["x"] + charge_start, "y": x["y"]} for x in points] -def charge_graph_next_slot(charge_start, points): +def charge_graph_next_slot(charge_start, points, skip_format=False): """Get the next charge slot start/end times from a list of graph points.""" now = int(time()) - data = _format_charge_graph(charge_start, points) + data = points if skip_format else _format_charge_graph(charge_start, points) # Filter to points from now onwards data = [x for x in data if x["t"] > now] @@ -47,10 +47,10 @@ def charge_graph_next_slot(charge_start, points): } -def charge_graph_in_slot(charge_start, points): +def charge_graph_in_slot(charge_start, points, skip_format=False): """Are we currently in a charge slot?""" now = int(time()) - data = _format_charge_graph(charge_start, points) + data = points if skip_format else _format_charge_graph(charge_start, points) # Loop through every value, skipping the last for idx in range(0, len(data) - 1): diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..d81a212 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,59 @@ +"""Tests for the utils.""" +from unittest import mock +import random +from time import time + +from custom_components.ohme import utils + + +async def test_format_charge_graph(hass): + """Test that the _test_format_charge_graph function adds given timestamp / 1000 to each x coordinate.""" + TEST_DATA = [{"x": 10, "y": 0}, {"x": 20, "y": 0}, + {"x": 30, "y": 0}, {"x": 40, "y": 0}] + + start_time = random.randint(1577836800, 1764547200) # 2020-2025 + start_time_ms = start_time * 1000 + + result = utils._format_charge_graph(start_time_ms, TEST_DATA) + expected = [{"t": TEST_DATA[0]['x'] + start_time, "y": mock.ANY}, + {"t": TEST_DATA[1]['x'] + start_time, "y": mock.ANY}, + {"t": TEST_DATA[2]['x'] + start_time, "y": mock.ANY}, + {"t": TEST_DATA[3]['x'] + start_time, "y": mock.ANY}] + + assert expected == result + + +async def test_charge_graph_next_slot(hass): + """Test that we correctly work out when the next slot starts and ends.""" + start_time = int(time()) + TEST_DATA = [{"t": start_time - 100, "y": 0}, + {"t": start_time + 1000, "y": 0}, + {"t": start_time + 1600, "y": 1000}, + {"t": start_time + 1800, "y": 1000}] + + result = utils.charge_graph_next_slot(0, TEST_DATA, skip_format=True) + result = { + "start": result['start'].timestamp(), + "end": result['end'].timestamp(), + } + + expected = { + "start": start_time + 1001, + "end": start_time + 1601, + } + + assert expected == result + + +async def test_charge_graph_in_slot(hass): + """Test that we correctly intepret outselves as in a slot.""" + start_time = int(time()) + TEST_DATA = [{"t": start_time - 100, "y": 0}, + {"t": start_time - 10, "y": 0}, + {"t": start_time + 200, "y": 1000}, + {"t": start_time + 300, "y": 1000}] + + result = utils.charge_graph_in_slot(0, TEST_DATA, skip_format=True) + expected = True + + assert expected == result