Merge branch 'fixes1' of github.com:springfall2008/batpred into fixes1

Author: springfall2008

apps/predbat/execute.py

@@ -783,7 +783,7 @@ def publish_inverter_data(self):
                 "unit_of_measurement": "kW",
                 "icon": "mdi:battery",
             },
-        )        
+        )
         self.dashboard_item(
             self.prefix + ".battery_power",
             state=dp3(self.battery_power / 1000.0),
@@ -795,7 +795,6 @@ def publish_inverter_data(self):
             },
         )
 
-
     def balance_inverters(self):
         """
         Attempt to balance multiple inverters

apps/predbat/fetch.py

@@ -761,7 +761,7 @@ def minute_data(
                 last_sample_value = mdata[last_sample_minute]
                 last_but_one_minute_sample = mdata[last_sample_minute + 1] if (last_sample_minute + 1) in mdata else last_sample_value
                 step = last_sample_value - last_but_one_minute_sample
-                if last_sample_minute < 10*60 and last_sample_minute > 0:
+                if last_sample_minute < 10 * 60 and last_sample_minute > 0:
                     for minute in range(last_sample_minute):
                         if minute not in mdata:
                             mdata[minute] = dp4(newest_state + step * (last_sample_minute - minute))

apps/predbat/gecloud.py

@@ -156,7 +156,7 @@
 RETRIES = 10
 RETRY_FACTOR = 1
 MAX_THREADS = 2
-MAX_START_TIME = 10*60
+MAX_START_TIME = 10 * 60
 
 attribute_table = {
     "time": {"friendly_name": "Time", "icon": "mdi:clock", "unit_of_measurement": "Time", "state_class": "timestamp"},
@@ -696,7 +696,7 @@ async def async_automatic_config(self, devices):
 
         if "givtcp_rest" in self.base.args:
             del self.base.args["givtcp_rest"]
-    
+
         # Use the first battery serial for the ge_cloud_serial (for status)
         self.base.args["ge_cloud_serial"] = devices["battery"][0]
 
@@ -1255,7 +1255,7 @@ async def async_get_inverter_data(self, endpoint, serial="", setting_id="", post
             self.log(f"Warn: GECloud: Exception during request to {url}: {e}")
             self.failures_total += 1
             return None
-                
+
         try:
             data = response.json()
         except requests.exceptions.JSONDecodeError:

apps/predbat/unit_test.py

@@ -8832,12 +8832,13 @@ def test_minute_data(my_predbat):
     """
     failed = False
     print("**** Testing minute_data function ****")
-    
+
     # Create test datetime objects with timezone awareness
     import pytz
+
     utc = pytz.UTC
     now = datetime(2024, 10, 4, 12, 5, 0, tzinfo=utc)  # Fixed time for testing
-    
+
     # Test 1: Basic functionality with simple history data
     print("Test 1: Basic functionality")
     history = [
@@ -8846,69 +8847,46 @@ def test_minute_data(my_predbat):
         {"state": "20.0", "last_updated": "2024-10-04T11:30:00+00:00"},
         {"state": "30.0", "last_updated": "2024-10-04T12:00:00+00:00"},
     ]
-    
-    result = my_predbat.minute_data(
-        history=history,
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated",
-        backwards=True,
-        smoothing=False
-    )
+
+    result = my_predbat.minute_data(history=history, days=1, now=now, state_key="state", last_updated_key="last_updated", backwards=True, smoothing=False)
 
     points = [0, 5, 34, 35, 65]
     result_points = [result.get(p) for p in points]
     expected_points = [30.0, 20.0, 20.0, 10.0, 0.0]
 
     # Check that we have data and it's reasonable
-    if len(result) != 24*60:
+    if len(result) != 24 * 60:
         print("ERROR: Basic test failed - no data returned")
         # Print result sorted by key for easier reading
         failed = True
     elif result_points != expected_points:
-        print("ERROR: Basic test failed - values incorrect, data points were {} expected {}".format(
-            result_points, expected_points))
+        print("ERROR: Basic test failed - values incorrect, data points were {} expected {}".format(result_points, expected_points))
         for key in sorted(result.keys()):
             print("  {}: {}".format(key, result[key]))
         failed = True
-    
+
     # Test 2: Empty history
     print("Test 2: Empty history")
-    empty_result = my_predbat.minute_data(
-        history=[],
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated"
-    )
-    
+    empty_result = my_predbat.minute_data(history=[], days=1, now=now, state_key="state", last_updated_key="last_updated")
+
     if empty_result != {}:
         print("ERROR: Empty history test failed - should return empty dict")
         failed = True
     else:
         print("Empty history test passed")
-    
+
     # Test 3: Scale parameter
     print("Test 3: Scale parameter")
-    scaled_result = my_predbat.minute_data(
-        history=history,
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated",
-        backwards=True,
-        scale=2.0
-    )
-    
+    scaled_result = my_predbat.minute_data(history=history, days=1, now=now, state_key="state", last_updated_key="last_updated", backwards=True, scale=2.0)
+
     if 0 not in scaled_result:
         print("ERROR: Scale test failed - no data at minute 0")
         failed = True
     elif scaled_result[0] != result[0] * 2.0:
         print("Scale test passed - values scaled correctly from {} to {}".format(result[0], scaled_result[0]))
     else:
         print("Scale test passed")
-    
+
     # Test 4: Smoothing enabled vs disabled comparison
     print("Test 4: Smoothing")
     result = my_predbat.minute_data(
@@ -8922,20 +8900,18 @@ def test_minute_data(my_predbat):
     )
 
     points = [0, 5, 34, 35, 65]
-    expected_points = [30.0, 30.0, dp4(20.0 + 10/30), 20.0, 10.0]
+    expected_points = [30.0, 30.0, dp4(20.0 + 10 / 30), 20.0, 10.0]
     result_points = [result.get(p) for p in points]
-    if len(result) != 24*60:
+    if len(result) != 24 * 60:
         print("ERROR: Smoothing test failed - no data returned")
         failed = True
     elif result_points != expected_points:
-        print("ERROR: Smoothing test - unsmoothed values incorrect, data points were {} expected {}".format(
-            result_points, expected_points))
+        print("ERROR: Smoothing test - unsmoothed values incorrect, data points were {} expected {}".format(result_points, expected_points))
         for key in sorted(result.keys()):
             print("  {}: {}".format(key, result[key]))
         failed = True
 
     failed = True
-    
 
     # Test 4.1: Smoothing clean increment
     print("Test 4.1: Smoothing clean increment")
@@ -8951,14 +8927,13 @@ def test_minute_data(my_predbat):
         max_increment=0,
     )
     points = [0, 5, 34, 35, 65]
-    expected_points = [dp4(30.0 + 5*(10/30)), 30.0, dp4(20.0 + 10/30), 20.0, 10.0]
+    expected_points = [dp4(30.0 + 5 * (10 / 30)), 30.0, dp4(20.0 + 10 / 30), 20.0, 10.0]
     result_points = [result.get(p) for p in points]
-    if len(result) != 24*60:
+    if len(result) != 24 * 60:
         print("ERROR: Smoothing test failed - no data returned")
         failed = True
     elif result_points != expected_points:
-        print("ERROR: Smoothing test - unsmoothed values incorrect, data points were {} expected {}".format(
-            result_points, expected_points))
+        print("ERROR: Smoothing test - unsmoothed values incorrect, data points were {} expected {}".format(result_points, expected_points))
         for key in sorted(result.keys()):
             print("  {}: {}".format(key, result[key]))
         failed = True
@@ -8969,49 +8944,29 @@ def test_minute_data(my_predbat):
         {"attributes": {"power": "15.0"}, "last_updated": "2024-10-04T11:00:00+00:00"},
         {"attributes": {"power": "25.0"}, "last_updated": "2024-10-04T11:30:00+00:00"},
     ]
-    
-    attrs_result = my_predbat.minute_data(
-        history=history_with_attrs,
-        days=1,
-        now=now,
-        state_key="power",
-        last_updated_key="last_updated",
-        backwards=True,
-        attributes=True
-    )
-    
+
+    attrs_result = my_predbat.minute_data(history=history_with_attrs, days=1, now=now, state_key="power", last_updated_key="last_updated", backwards=True, attributes=True)
+
     if len(attrs_result) == 0:
         print("ERROR: Attributes test failed - no data returned")
         failed = True
     else:
         print("Attributes test passed - returned {} data points".format(len(attrs_result)))
-    
+
     # Test 6: Unit conversion
     print("Test 6: Unit conversion")
     history_with_units = [
-        {
-            "state": "1000.0", 
-            "last_updated": "2024-10-04T11:00:00+00:00",
-            "attributes": {"unit_of_measurement": "W"}
-        },
+        {"state": "1000.0", "last_updated": "2024-10-04T11:00:00+00:00", "attributes": {"unit_of_measurement": "W"}},
     ]
-    
-    converted_result = my_predbat.minute_data(
-        history=history_with_units,
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated",
-        backwards=True,
-        required_unit="kWh"
-    )
-    
+
+    converted_result = my_predbat.minute_data(history=history_with_units, days=1, now=now, state_key="state", last_updated_key="last_updated", backwards=True, required_unit="kWh")
+
     if len(converted_result) == 0:
         print("ERROR: Unit conversion test failed - no data returned")
         failed = True
     else:
         print("Unit conversion test passed - function handles unit conversion")
-    
+
     # Test 7: Invalid/unavailable data filtering
     print("Test 7: Invalid data filtering")
     history_with_invalid = [
@@ -9020,37 +8975,22 @@ def test_minute_data(my_predbat):
         {"state": "unknown", "last_updated": "2024-10-04T11:30:00+00:00"},
         {"state": "20.0", "last_updated": "2024-10-04T11:45:00+00:00"},
     ]
-    
-    filtered_result = my_predbat.minute_data(
-        history=history_with_invalid,
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated",
-        backwards=True
-    )
-    
+
+    filtered_result = my_predbat.minute_data(history=history_with_invalid, days=1, now=now, state_key="state", last_updated_key="last_updated", backwards=True)
+
     # Function should filter out invalid data
     if len(filtered_result) == 0:
         print("ERROR: Invalid data filtering test failed - no data returned")
         failed = True
     else:
         print("Invalid data filtering test passed - function filters invalid data")
-    
+
     # Test 8: Accumulate parameter
     print("Test 8: Accumulate parameter")
     accumulate_data = {0: 5.0, 30: 10.0, 60: 15.0}
-    
-    accumulated_result = my_predbat.minute_data(
-        history=history,
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated",
-        backwards=True,
-        accumulate=accumulate_data
-    )
-    
+
+    accumulated_result = my_predbat.minute_data(history=history, days=1, now=now, state_key="state", last_updated_key="last_updated", backwards=True, accumulate=accumulate_data)
+
     if 0 not in accumulated_result:
         print("ERROR: Accumulate test failed - no data at minute 0")
         failed = True
@@ -9063,62 +9003,41 @@ def test_minute_data(my_predbat):
             print("Accumulate test passed - values correctly accumulated")
         else:
             print("Accumulate test passed - function handles accumulation (original: {}, accumulated: {})".format(original, accumulated))
-    
+
     # Test 9: Forward time direction
     print("Test 9: Forward time direction")
-    forward_result = my_predbat.minute_data(
-        history=history,
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated",
-        backwards=False
-    )
-    
+    forward_result = my_predbat.minute_data(history=history, days=1, now=now, state_key="state", last_updated_key="last_updated", backwards=False)
+
     # Function should handle forward direction without error
     print("Forward time test passed - function handles forward direction")
-    
+
     # Test 10: Missing keys handling
     print("Test 10: Missing keys handling")
     history_missing_keys = [
         {"state": "10.0"},  # Missing last_updated
         {"last_updated": "2024-10-04T11:30:00+00:00"},  # Missing state
     ]
-    
-    missing_keys_result = my_predbat.minute_data(
-        history=history_missing_keys,
-        days=1,
-        now=now,
-        state_key="state",
-        last_updated_key="last_updated",
-        backwards=True
-    )
-    
+
+    missing_keys_result = my_predbat.minute_data(history=history_missing_keys, days=1, now=now, state_key="state", last_updated_key="last_updated", backwards=True)
+
     # Function should handle missing keys gracefully
     print("Missing keys test passed - function handles missing keys gracefully")
-    
+
     # Test 11: Different state key
     print("Test 11: Different state key")
     history_different_key = [
         {"value": "50.0", "last_updated": "2024-10-04T11:00:00+00:00"},
         {"value": "60.0", "last_updated": "2024-10-04T11:30:00+00:00"},
     ]
-    
-    different_key_result = my_predbat.minute_data(
-        history=history_different_key,
-        days=1,
-        now=now,
-        state_key="value",
-        last_updated_key="last_updated",
-        backwards=True
-    )
-    
+
+    different_key_result = my_predbat.minute_data(history=history_different_key, days=1, now=now, state_key="value", last_updated_key="last_updated", backwards=True)
+
     if len(different_key_result) == 0:
         print("ERROR: Different state key test failed - no data returned")
         failed = True
     else:
         print("Different state key test passed")
-    
+
     print("**** minute_data tests completed ****")
     return failed