More unit tests and experimental charge state logic

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(

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):

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