feat: Automatically postpone scheduled charge time when percentage threshold is set (closes #339) Merge pull request #349 from cbruegg/multi-day-charge-control

psa_car_controller/psa/RemoteClient.py

@@ -272,12 +272,12 @@ class RemoteClient:
             charge_type = IMMEDIATE_CHARGE
         else:
             charge_type = DELAYED_CHARGE
-        hour, minute = self.__get_charge_hour(vin)
+        hour, minute = self.get_charge_hour(vin)
         res = self.veh_charge_request(vin, hour, minute, charge_type)
         logger.info("charge_now: %s", res)
         return True
 
-    def __get_charge_hour(self, vin):
+    def get_charge_hour(self, vin):
         hour_str = self.vehicles_list.get_car_by_vin(vin).status.get_energy('Electric').charging.next_delayed_time
         try:
             return parse_hour(hour_str)[:2]

psa_car_controller/psacc/application/charge_control.py

@@ -8,7 +8,7 @@ from time import sleep
 
 import pytz
 
-from psa_car_controller.psa.constants import DISCONNECTED, INPROGRESS, FINISHED
+from psa_car_controller.psa.constants import DISCONNECTED, INPROGRESS, FINISHED, STOPPED
 from psa_car_controller.common.utils import RateLimitException
 from .psa_client import PSAClient
 
@@ -55,7 +55,10 @@ class ChargeControl:
         self.retry_count = 0
         return True
 
-    def force_update(self, quick_refresh):
+    def force_update(self, vehicle_status):
+        charging_mode = vehicle_status.get_energy('Electric').charging.charging_mode
+        quick_refresh = isinstance(charging_mode, str) and charging_mode == "Quick"
+
         # force update if the car doesn't send info during 10 minutes
         last_update = self.psacc.vehicles_list.get_car_by_vin(self.vin).get_status().get_energy('Electric').updated_at
         if quick_refresh:
@@ -68,18 +71,24 @@ class ChargeControl:
             except RateLimitException:
                 logger.exception("force_update:")
 
+    def __is_approaching_scheduled_time(self, now: datetime):
+        scheduled_hour, scheduled_minute = self.psacc.remote_client.get_charge_hour(self.vin)
+        minutes_passed = now.hour * 60 + now.minute
+        scheduled_minute_of_day = scheduled_hour * 60 + scheduled_minute
+        return minutes_passed < scheduled_minute_of_day and scheduled_minute_of_day - minutes_passed < 30
+
     def process(self):
         now = datetime.now()
         try:
             vehicle_status = self.psacc.vehicles_list.get_car_by_vin(self.vin).get_status()
             status = vehicle_status.get_energy('Electric').charging.status
             level = vehicle_status.get_energy('Electric').level
-            if status == "InProgress":
+            has_threshold = self.percentage_threshold < 100
+            hit_threshold = level >= self.percentage_threshold
+            if status == INPROGRESS:
                 logger.info("charging status of %s is %s, battery level: %d", self.vin, status, level)
-                charging_mode = vehicle_status.get_energy('Electric').charging.charging_mode
-                quick_refresh = isinstance(charging_mode, str) and charging_mode == "Quick"
-                self.force_update(quick_refresh)
-                if level >= self.percentage_threshold and self.retry_count < 2:
+                self.force_update(vehicle_status)
+                if hit_threshold and self.retry_count < 2:
                     logger.info("Charge threshold is reached, stop the charge")
                     self.control_charge_with_ack(False)
                 elif self._next_stop_hour is not None:
@@ -94,6 +103,11 @@ class ChargeControl:
                             thread = threading.Timer(periodicity, self.process)
                             thread.setDaemon(True)
                             thread.start()
+            elif status == STOPPED and has_threshold and hit_threshold and self.__is_approaching_scheduled_time(now):
+                logger.info("Approaching scheduled charging time, but should not charge. Postponing charge hour!")
+                self.force_update(vehicle_status)
+                hour = now.hour - 1 if now.hour > 0 else 23
+                self.psacc.remote_client.change_charge_hour(self.vin, hour, 0)
             else:
                 if self._next_stop_hour is not None and self._next_stop_hour < now:
                     self._next_stop_hour += timedelta(days=1)

tests/test_chargecontrol.py

@@ -0,0 +1,117 @@
+import logging
+import unittest
+import datetime
+from time import sleep
+
+from psa_car_controller.common.mylogger import my_logger
+from psa_car_controller.psa.constants import INPROGRESS, STOPPED
+from psa_car_controller.psacc.application.charge_control import ChargeControl
+
+# This simulates a 70 % charge limit with the scheduled charge time set to currentTime + 2 minutes. The initial fake
+# status is "75 % and charging InProgress". First, ChargeControl should stop the charge.
+#
+# The fake status then changes to "75 % and charging Stopped, but still plugged". Since the scheduled charge time is
+# only 2 minutes away, ChargeControl postpones the charge to the next day. It does so by setting the scheduled charge
+# time to (currentHour - 1):00. For instance, when it's 14:21, it will change the scheduled charge time to 13:00,
+# which has to be on the next day.
+#
+# For reference, here is a correct sample output:
+# 2022-04-20 12:31:15,551 :: INFO :: charging status of 123 is InProgress, battery level: 75
+# 2022-04-20 12:31:15,552 :: INFO :: Charge threshold is reached, stop the charge
+# 2022-04-20 12:31:15,552 :: INFO :: charge_now(123, False)
+# 2022-04-20 12:31:15,552 :: INFO :: Approaching scheduled charging time, but should not charge. Postponing charge hour!
+# 2022-04-20 12:31:15,552 :: INFO :: Changing stop_hour to 11:0
+
+class MyTestCase(unittest.TestCase):
+
+    def test_something(self):
+        my_logger()
+        ChargeControl.MQTT_TIMEOUT = 0
+
+        now = datetime.datetime.now()
+
+        vin = "123"
+        percentage_threshold = 70
+
+        state = {
+            "charging": True,
+            "stop_hour": [now.hour, now.minute + 2]
+        }
+
+        expectations = {
+            "stopped_charge": False,
+            "postponed_charge": False
+        }
+
+        def met_expectations():
+            return expectations["stopped_charge"] and expectations["postponed_charge"]
+
+        class MockClient:
+            def charge_now(self, vin, now):
+                logging.getLogger().info(f"charge_now({vin}, {now})")
+                state["charging"] = now
+                if not now:
+                    expectations["stopped_charge"] = True
+
+            def get_charge_hour(self, vin):
+                return state["stop_hour"]
+
+            def change_charge_hour(self, vin, hour, minute):
+                logging.getLogger().info(f"Changing stop_hour to {hour}:{minute}")
+                state["stop_hour"] = [hour, minute]
+                expectations["postponed_charge"] = True
+
+        mock_client = MockClient()
+
+        class MockCharging:
+            def __init__(self):
+                self.status = INPROGRESS if state["charging"] else STOPPED
+                self.charging_mode = "slow"
+
+        class MockEnergy:
+            def __init__(self):
+                self.charging = MockCharging()
+                self.level = 75
+                self.updated_at = datetime.datetime.now(datetime.timezone.utc)
+
+        class MockStatus:
+            def get_energy(self, energy_type):
+                return MockEnergy()
+
+        class MockCar:
+            def get_status(self):
+                return MockStatus()
+
+            def get_energy(self):
+                return MockEnergy()
+
+        class MockVehiclesList:
+            def get_car_by_vin(self, vin):
+                return MockCar()
+
+        mock_vehicles_list = MockVehiclesList()
+
+        class MockPsa:
+            def __init__(self):
+                self.vehicles_list = mock_vehicles_list
+                self.remote_client = mock_client
+                self.info_refresh_rate = 1
+
+            def get_vehicle_info(self, vin):
+                return MockStatus()
+
+        mock_psac = MockPsa()
+        # noinspection PyTypeChecker
+        cc = ChargeControl(mock_psac, vin, percentage_threshold, state["stop_hour"])
+
+        rounds = 0
+        while not met_expectations() and rounds < 10:
+            cc.process()
+            rounds += 1
+        self.assertTrue(expectations["stopped_charge"], "charge_now(..., False) should be called to stop the charge")
+        self.assertTrue(expectations["postponed_charge"], "change_charge_hour(...) should be called to postpone "
+                                                          "the charge until the next day")
+
+
+if __name__ == '__main__':
+    unittest.main()