Merge branch 'dev' of https://github.com/fboundy/pv_opt into dev

fboundy · Dec 24, 2024 · 49490e5 · 49490e5
2 parents c9ebd30 + 8245f0d
commit 49490e5
Show file tree

Hide file tree

Showing 5 changed files with 36 additions and 127 deletions.
diff --git a/.test/solis_cloud_test.py b/.test/solis_cloud_test.py
@@ -63,9 +63,7 @@ def get_body(self, **params):
         return body
 
     def digest(self, body: str) -> str:
-        return base64.b64encode(hashlib.md5(body.encode("utf-8")).digest()).decode(
-            "utf-8"
-        )
+        return base64.b64encode(hashlib.md5(body.encode("utf-8")).digest()).decode("utf-8")
 
     def header(self, body: str, canonicalized_resource: str) -> dict[str, str]:
         content_md5 = self.digest(body)
@@ -74,17 +72,7 @@ def header(self, body: str, canonicalized_resource: str) -> dict[str, str]:
         now = datetime.now(timezone.utc)
         date = now.strftime("%a, %d %b %Y %H:%M:%S GMT")
 
-        encrypt_str = (
-            "POST"
-            + "\n"
-            + content_md5
-            + "\n"
-            + content_type
-            + "\n"
-            + date
-            + "\n"
-            + canonicalized_resource
-        )
+        encrypt_str = "POST" + "\n" + content_md5 + "\n" + content_type + "\n" + date + "\n" + canonicalized_resource
         hmac_obj = hmac.new(
             self.key_secret.encode("utf-8"),
             msg=encrypt_str.encode("utf-8"),
@@ -105,29 +93,23 @@ def header(self, body: str, canonicalized_resource: str) -> dict[str, str]:
     def inverter_id(self):
         body = self.get_body(stationId=self.plant_id)
         header = self.header(body, self.URLS["inverterList"])
-        response = requests.post(
-            self.URLS["root"] + self.URLS["inverterList"], data=body, headers=header
-        )
+        response = requests.post(self.URLS["root"] + self.URLS["inverterList"], data=body, headers=header)
         if response.status_code == HTTPStatus.OK:
             return response.json()["data"]["page"]["records"][0].get("id", "")
 
     @property
     def inverter_sn(self):
         body = self.get_body(stationId=self.plant_id)
         header = self.header(body, self.URLS["inverterList"])
-        response = requests.post(
-            self.URLS["root"] + self.URLS["inverterList"], data=body, headers=header
-        )
+        response = requests.post(self.URLS["root"] + self.URLS["inverterList"], data=body, headers=header)
         if response.status_code == HTTPStatus.OK:
             return response.json()["data"]["page"]["records"][0].get("sn", "")
 
     @property
     def inverter_details(self):
         body = self.get_body(id=self.inverter_id, sn=self.inverter_sn)
         header = self.header(body, self.URLS["inverterDetail"])
-        response = requests.post(
-            self.URLS["root"] + self.URLS["inverterDetail"], data=body, headers=header
-        )
+        response = requests.post(self.URLS["root"] + self.URLS["inverterDetail"], data=body, headers=header)
 
         if response.status_code == HTTPStatus.OK:
             return response.json()["data"]
@@ -148,9 +130,7 @@ def set_code(self, cid, value):
             body = self.get_body(inverterSn=self.inverter_sn, cid=cid, value=value)
             headers = self.header(body, self.URLS["control"])
             headers["token"] = self.token
-            response = requests.post(
-                self.URLS["root"] + self.URLS["control"], data=body, headers=headers
-            )
+            response = requests.post(self.URLS["root"] + self.URLS["control"], data=body, headers=headers)
             if response.status_code == HTTPStatus.OK:
                 return response.json()
 
@@ -162,18 +142,14 @@ def read_code(self, cid):
             body = self.get_body(inverterSn=self.inverter_sn, cid=cid)
             headers = self.header(body, self.URLS["atRead"])
             headers["token"] = self.token
-            response = requests.post(
-                self.URLS["root"] + self.URLS["atRead"], data=body, headers=headers
-            )
+            response = requests.post(self.URLS["root"] + self.URLS["atRead"], data=body, headers=headers)
             if response.status_code == HTTPStatus.OK:
                 return response.json()["data"]["msg"]
 
     def login(self):
         body = self.get_body(username=self.username, password=self.md5passowrd)
         header = self.header(body, self.URLS["login"])
-        response = requests.post(
-            self.URLS["root"] + self.URLS["login"], data=body, headers=header
-        )
+        response = requests.post(self.URLS["root"] + self.URLS["login"], data=body, headers=header)
         status = response.status_code
         if status == HTTPStatus.OK:
             result = response.json()

diff --git a/apps/pv_opt/.test.py b/apps/pv_opt/.test.py
@@ -38,19 +38,9 @@
     result = query_api.query(org=org, query=query)
 
     # Process the results
-    data = [
-        {"Time": record.get_time(), "Value": record.get_value()}
-        for record in result[-1].records
-    ]
+    data = [{"Time": record.get_time(), "Value": record.get_value()} for record in result[-1].records]
     series += [pd.DataFrame(data)]
-    series[-1] = (
-        series[-1]
-        .set_index("Time")
-        .resample("1min")
-        .mean()
-        .fillna(0)["Value"]
-        .rename(entity)
-    )
+    series[-1] = series[-1].set_index("Time").resample("1min").mean().fillna(0)["Value"].rename(entity)
 df = pd.concat(series, axis=1)
 df = df.resample("5min").mean()
 # Close the client

diff --git a/apps/pv_opt/pvpy.py b/apps/pv_opt/pvpy.py
@@ -1,6 +1,5 @@
 # %%
 from copy import copy
-
 # from scipy.stats import linregress
 from datetime import datetime
 

diff --git a/apps/pv_opt/solax.py b/apps/pv_opt/solax.py
@@ -71,15 +71,10 @@ def __init__(self, inverter_type, host) -> None:
         ):
             for item in defs:
                 if isinstance(defs[item], str):
-                    conf[item] = defs[item].replace(
-                        "{device_name}", self.host.device_name
-                    )
+                    conf[item] = defs[item].replace("{device_name}", self.host.device_name)
                     # conf[item] = defs[item].replace("{inverter_sn}", self.host.inverter_sn)
                 elif isinstance(defs[item], list):
-                    conf[item] = [
-                        z.replace("{device_name}", self.host.device_name)
-                        for z in defs[item]
-                    ]
+                    conf[item] = [z.replace("{device_name}", self.host.device_name) for z in defs[item]]
                     # conf[item] = [z.replace("{inverter_sn}", self.host.inverter_sn) for z in defs[item]]
                 else:
                     conf[item] = defs[item]
@@ -111,9 +106,7 @@ def control_charge(self, enable, **kwargs):
             if enable:
                 self.host.set_select("use_mode", "Force Time Use")
                 time_now = pd.Timestamp.now(tz=self.tz)
-                start = (
-                    kwargs.get("start", time_now).floor("15min").strftime(TIMEFORMAT)
-                )
+                start = kwargs.get("start", time_now).floor("15min").strftime(TIMEFORMAT)
                 end = kwargs.get("end", time_now).ceil("30min").strftime(TIMEFORMAT)
                 self.host.set_select("charge_start_time_1", start)
                 self.host.set_select("charge_end_time_1", end)
@@ -126,25 +119,17 @@ def control_charge(self, enable, **kwargs):
                     voltage = self.host.get_config("battery_voltage")
                     if voltage == 0:
                         voltage = BATTERY_VOLTAGE_DEFAULT
-                        self.log(
-                            f"Read a battery voltage of zero. Assuming default of {BATTERY_VOLTAGE_DEFAULT}"
-                        )
+                        self.log(f"Read a battery voltage of zero. Assuming default of {BATTERY_VOLTAGE_DEFAULT}")
                     current = abs(round(power / voltage, 1))
-                    current = min(
-                        current, self.host.get_config("battery_current_limit_amps")
-                    )
+                    current = min(current, self.host.get_config("battery_current_limit_amps"))
 
                     # If power is exactly "1" then this is a car hold slot, to prevent sleep, set to 1A
                     if power == 1:
                         current = 1
-                        self.log(
-                            "Power = 1W (hold slot). Setting current to 1A (to prevent sleep SOC being reached)"
-                        )
+                        self.log("Power = 1W (hold slot). Setting current to 1A (to prevent sleep SOC being reached)")
                         self.log("")
                     else:
-                        self.log(
-                            f"Power {power:0.0f} = {current:0.1f}A at {self.host.get_config('battery_voltage')}V"
-                        )
+                        self.log(f"Power {power:0.0f} = {current:0.1f}A at {self.host.get_config('battery_voltage')}V")
 
                     changed, written = self.host.write_and_poll_value(
                         entity_id=entity_id, value=current, tolerance=1, verbose=True
@@ -154,9 +139,7 @@ def control_charge(self, enable, **kwargs):
                         if written:
                             self.log(f"Current {current}A written to inverter")
                         else:
-                            self.log(
-                                f"Failed to write current of {current}A to inverter"
-                            )
+                            self.log(f"Failed to write current of {current}A to inverter")
                     else:
                         self.log("Inverter already at correct current")
 
@@ -172,9 +155,7 @@ def control_charge(self, enable, **kwargs):
                         if written:
                             self.log(f"Target SOC {target_soc}% written to inverter")
                         else:
-                            self.log(
-                                f"Failed to write Target SOC of {target_soc}% to inverter"
-                            )
+                            self.log(f"Failed to write Target SOC of {target_soc}% to inverter")
                     else:
                         self.log("Inverter already at correct Target SOC")
             else:
@@ -267,17 +248,11 @@ def _solax_charge_periods(self, **kwargs):
         if self.type == "SOLAX_X1":
             return {
                 limit: pd.Timestamp(
-                    self.host.get_state_retry(
-                        entity_id=self.host.config[f"id_charge_{limit}_time_1"]
-                    ),
+                    self.host.get_state_retry(entity_id=self.host.config[f"id_charge_{limit}_time_1"]),
                     tz=self.tz,
                 )
                 for limit in LIMITS
-            } | {
-                "current": self.host.get_state_retry(
-                    entity_id=self.host.config[f"id_max_charge_current"]
-                )
-            }
+            } | {"current": self.host.get_state_retry(entity_id=self.host.config[f"id_max_charge_current"])}
 
         else:
             self._unknown_inverter()
diff --git a/apps/pv_opt/sunsynk.py b/apps/pv_opt/sunsynk.py
@@ -56,17 +56,13 @@
             "id_priority_load": "sensor.{device_name}_{inverter_sn}_priority_load",
             "id_timed_charge_start": "sensor.{device_name}_{inverter_sn}_prog1_time",
             "id_timed_charge_end": "sensor.{device_name}_{inverter_sn}_prog2_time",
-            "id_timed_charge_unused": [
-                "sensor.{device_name}_{inverter_sn}_" + f"prog{i}_time"
-                for i in range(2, 7)
-            ],
+            "id_timed_charge_unused": ["sensor.{device_name}_{inverter_sn}_" + f"prog{i}_time" for i in range(2, 7)],
             "id_timed_charge_enable": "sensor.{device_name}_{inverter_sn}_prog1_charge",
             "id_timed_charge_capacity": "sensor.{device_name}_{inverter_sn}_prog1_capacity",
             "id_timed_discharge_start": "sensor.{device_name}_{inverter_sn}_prog3_time",
             "id_timed_discharge_end": "sensor.{device_name}_{inverter_sn}_prog4_time",
             "id_timed_discharge_unused": [
-                "sensor.{device_name}_{inverter_sn}_" + f"prog{i}_time"
-                for i in [1, 2, 5, 6]
+                "sensor.{device_name}_{inverter_sn}_" + f"prog{i}_time" for i in [1, 2, 5, 6]
             ],
             "id_timed_dicharge_enable": "sensor.{device_name}_{inverter_sn}_prog3_charge",
             "id_timed_discharge_capacity": "sensor.{device_name}_{inverter_sn}_prog3_capacity",
@@ -111,21 +107,11 @@ def __init__(self, inverter_type, host) -> None:
         ):
             for item in defs:
                 if isinstance(defs[item], str):
-                    conf[item] = defs[item].replace(
-                        "{device_name}", self.host.device_name
-                    )
-                    conf[item] = defs[item].replace(
-                        "{inverter_sn}", self.host.inverter_sn
-                    )
+                    conf[item] = defs[item].replace("{device_name}", self.host.device_name)
+                    conf[item] = defs[item].replace("{inverter_sn}", self.host.inverter_sn)
                 elif isinstance(defs[item], list):
-                    conf[item] = [
-                        z.replace("{device_name}", self.host.device_name)
-                        for z in defs[item]
-                    ]
-                    conf[item] = [
-                        z.replace("{inverter_sn}", self.host.inverter_sn)
-                        for z in defs[item]
-                    ]
+                    conf[item] = [z.replace("{device_name}", self.host.device_name) for z in defs[item]]
+                    conf[item] = [z.replace("{inverter_sn}", self.host.inverter_sn) for z in defs[item]]
                 else:
                     conf[item] = defs[item]
 
@@ -170,18 +156,13 @@ def control_charge(self, enable, **kwargs):
                 self.enable_timed_mode()
                 params = {
                     self.config["json_work_mode"]: 2,
-                    self.config["json_timed_charge_target_soc"]: kwargs.get(
-                        "target_soc", 100
-                    ),
-                    self.config["json_timed_charge_start"]: kwargs.get(
-                        "start", time_now.strftime(TIMEFORMAT)
-                    ),
+                    self.config["json_timed_charge_target_soc"]: kwargs.get("target_soc", 100),
+                    self.config["json_timed_charge_start"]: kwargs.get("start", time_now.strftime(TIMEFORMAT)),
                     self.config["json_timed_charge_end"]: kwargs.get(
                         "end", time_now.ceil("30min").strftime(TIMEFORMAT)
                     ),
                     self.config["json_charge_current"]: min(
-                        kwargs.get("power", 0)
-                        / self.host.get_config("battery_voltage"),
+                        kwargs.get("power", 0) / self.host.get_config("battery_voltage"),
                         self.host.get_config("battery_current_limit_amps"),
                     ),
                     self.config["json_timed_charge_enable"]: True,
@@ -196,9 +177,7 @@ def control_charge(self, enable, **kwargs):
                     self.config["json_target_soc"]: 100,
                     self.config["json_timed_charge_start"]: "00:00",
                     self.config["json_timed_charge_end"]: "00:00",
-                    self.config["json_charge_current"]: self.host.get_config(
-                        "battery_current_limit_amps"
-                    ),
+                    self.config["json_charge_current"]: self.host.get_config("battery_current_limit_amps"),
                     self.config["json_timed_charge_enable"]: False,
                     self.config["json_gen_charge_enable"]: True,
                 } | {x: "00:00" for x in self.config["json_timed_charge_unused"]}
@@ -216,9 +195,7 @@ def control_discharge(self, enable, **kwargs):
                     self.config["json_timed_discharge_target_soc"]: kwargs.get(
                         "target_soc", self.host.get_config("maximum_dod_percent")
                     ),
-                    self.config["json_timed_discharge_start"]: kwargs.get(
-                        "start", time_now.strftime(TIMEFORMAT)
-                    ),
+                    self.config["json_timed_discharge_start"]: kwargs.get("start", time_now.strftime(TIMEFORMAT)),
                     self.config["json_timed_discharge_end"]: kwargs.get(
                         "end", time_now.ceil("30min").strftime(TIMEFORMAT)
                     ),
@@ -255,18 +232,10 @@ def status(self):
         time_now = pd.Timestamp.now(tz=self.tz)
 
         if self.type == "SUNSYNK_SOLARSYNK2":
-            charge_start = pd.Timestamp(
-                self.host.get_config("id_timed_charge_start"), tz=self.tz
-            )
-            charge_end = pd.Timestamp(
-                self.host.get_config("id_timed_charge_end"), tz=self.tz
-            )
-            discharge_start = pd.Timestamp(
-                self.host.get_config("id_timed_charge_start"), tz=self.tz
-            )
-            discharge_end = pd.Timestamp(
-                self.host.get_config("id_timed_charge_end"), tz=self.tz
-            )
+            charge_start = pd.Timestamp(self.host.get_config("id_timed_charge_start"), tz=self.tz)
+            charge_end = pd.Timestamp(self.host.get_config("id_timed_charge_end"), tz=self.tz)
+            discharge_start = pd.Timestamp(self.host.get_config("id_timed_charge_start"), tz=self.tz)
+            discharge_end = pd.Timestamp(self.host.get_config("id_timed_charge_end"), tz=self.tz)
 
             status = {
                 "timer mode": self.host.get_config("id_use_timer"),