diff --git a/.test/solis_cloud_test.py b/.test/solis_cloud_test.py new file mode 100644 index 0000000..2146a40 --- /dev/null +++ b/.test/solis_cloud_test.py @@ -0,0 +1,227 @@ +# %% +import hashlib +import hmac +import base64 +import json +import re +import requests +from http import HTTPStatus +from datetime import datetime, timezone +import pandas as pd + +# def getInverterList(config): +# body = getBody(stationId=config['plantId']) +# print(body) +# body = '{"stationId":"'+config['plantId']+'"}' +# print(body) +# header = prepare_header(config, body, INVERTER_URL) +# response = requests.post("https://www.soliscloud.com:13333"+INVERTER_URL, data = body, headers = header) +# inverterList = response.json() +# inverterId = "" +# for record in inverterList['data']['page']['records']: +# inverterId = record.get('id') +# return inverterList['data']['page']['records'][0] +INVERTER_DEFS = { + "SOLIS_CLOUD": { + "bits": [ + "SelfUse", + "Timed", + "OffGrid", + "BatteryWake", + "Backup", + "GridCharge", + "FeedInPriority", + ], + }, +} + + +class SolisCloud: + URLS = { + "root": "https://www.soliscloud.com:13333", + "login": "/v2/api/login", + "control": "/v2/api/control", + "inverterList": "/v1/api/inverterList", + "inverterDetail": "/v1/api/inverterDetail", + "atRead": "/v2/api/atRead", + } + + def __init__(self, username, password, key_id, key_secret, plant_id): + self.username = username + self.key_id = key_id + self.key_secret = key_secret + self.plant_id = plant_id + self.md5passowrd = hashlib.md5(password.encode("utf-8")).hexdigest() + self.token = "" + + def get_body(self, **params): + body = "{" + for key in params: + body += f'"{key}":"{params[key]}",' + body = body[:-1] + "}" + return body + + def digest(self, body: str) -> str: + 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) + content_type = "application/json" + + 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 + hmac_obj = hmac.new(self.key_secret.encode("utf-8"), msg=encrypt_str.encode("utf-8"), digestmod=hashlib.sha1) + sign = base64.b64encode(hmac_obj.digest()) + authorization = "API " + self.key_id + ":" + sign.decode("utf-8") + + header = { + "Content-MD5": content_md5, + "Content-Type": content_type, + "Date": date, + "Authorization": authorization, + } + return header + + @property + 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) + 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) + 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) + + if response.status_code == HTTPStatus.OK: + return response.json()["data"] + + @property + def is_online(self): + return self.inverter_details["state"] == 1 + + @property + def last_seen(self): + return pd.to_datetime(int(self.inverter_details["dataTimestamp"]), unit="ms") + + def set_code(self, cid, value): + if self.token == "": + self.login() + + if self.token != "": + 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) + if response.status_code == HTTPStatus.OK: + return response.json() + + def read_code(self, cid): + if self.token == "": + self.login() + + if self.token != "": + 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) + 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) + status = response.status_code + if status == HTTPStatus.OK: + result = response.json() + self.token = result["csrfToken"] + print("Logged in to SolisCloud OK") + + else: + print(status) + + def mode_switch(self): + bits = INVERTER_DEFS["SOLIS_CLOUD"]["bits"] + code = int(self.read_code("636")) + switches = {bit: (code & 2**i == 2**i) for i, bit in enumerate(bits)} + return {"code": code, "switches": switches} + + def timed_status(self, tz="GB"): + data = self.read_code("103").split(",") + return { + "charge": { + "current": float(data[0]), + "start": pd.Timestamp(data[2].split("-")[0], tz=tz), + "end": pd.Timestamp(data[2].split("-")[1], tz=tz), + }, + "discharge": { + "current": float(data[1]), + "start": pd.Timestamp(data[3].split("-")[0], tz=tz), + "end": pd.Timestamp(data[3].split("-")[1], tz=tz), + }, + } + + def read_backup_mode_soc(self): + return int(self.read_code("157")) + + def set_mode_switch(self, code): + return self.set_code("636", code) + + def get_time_string(self, time_status): + time_string = ",".join( + [ + str(int(time_status["charge"]["current"])), + str(int(time_status["discharge"]["current"])), + f'{time_status["charge"]["start"].strftime("%H:%M")}-{time_status["charge"]["end"].strftime("%H:%M")}', + f'{time_status["discharge"]["start"].strftime("%H:%M")}-{time_status["discharge"]["end"].strftime("%H:%M")}', + ] + ) + return f"{time_string},0,0,00:00-00:00,00:00-00:00,0,0,00:00-00:00,00:00-00:00" + + def set_timer(self, direction, start, end, power): + voltage = 50 + current_times = self.timed_status() + new_times = current_times.copy() + new_times[direction]["start"] = start + new_times[direction]["end"] = end + new_times[direction]["current"] = power / voltage + current_time_string = self.read_code(103) + new_time_string = self.get_time_string(new_times) + if new_time_string != current_time_string: + return self.set_code("103", new_time_string) + else: + return {"code": -1} + + +# %% +if __name__ == "__main__": + config = { + "key_secret": "735f96b6131b4691af944de80d2f1a1f", + "key_id": "1300386381676670076", + "plant_id": "1298491919448891215", + "username": "boundywindsor@gmail.com", + "password": "7y@-Ekdh&@F9", + } + + sc = SolisCloud(**config) + sc.login() + print(sc.mode_switch()) + print(sc.timed_status()) + +# %% +sc.set_timer("charge", pd.Timestamp("00:50"), pd.Timestamp("01:00"), 3000) +# %% diff --git a/README.md b/README.md index 7466d85..ebe1409 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,4 @@ -# PV Opt: Home Assistant Solar/Battery Optimiser v3.17.2 +# PV Opt: Home Assistant Solar/Battery Optimiser v3.18.0

This documentation needs updating!

@@ -9,8 +9,7 @@ The application will integrate fully with Solis inverters which are controlled u - [Home Assistant Solax Modbus Integration](https://github.com/wills106/homeassistant-solax-modbus) - [Home Assistant Core Modbus Integration](https://github.com/fboundy/ha_solis_modbus) - [Home Assistant Solarman Integration](https://github.com/StephanJoubert/home_assistant_solarman) -- [Home Assistant Solis Sensor Integration](https://github.com/hultenvp/solis-sensor) (read-only mode) -- [Home Assistant Solis Control Integration](https://github.com/stevegal/solis_control) (allows inverter control via solis_cloud and HA automations) +- [Home Assistant Solis Sensor Integration](https://github.com/hultenvp/solis-sensor) Once installed it should require miminal configuration. Other inverters/integrations can be added if required or can be controlled indirectly using automations. @@ -324,126 +323,6 @@ Restarts between Home Assistant and Add-Ons are not synchronised so it is helpfu addon: a0d7b954_appdaemon mode: single -

14. For Solis-Control: Add Automation to Control Inverter

- -If you're using the solis-sensor and solis_control integrations through Solis Cloud, you'll need to add the following automation which will send the messages to Solis Cloud in order to control your inverter. N.B: It's important that you've set up the solis_control integration correctly and requested API access via Solis Cloud Technical Support. - -``` -alias: "Solis: Use PV_Opt" -description: "Use the output of pv_opt to control your Solis inverter via Solis Cloud." -trigger: - - platform: state - entity_id: - - sensor.pvopt_status - to: Idle (Read Only) - for: - hours: 0 - minutes: 0 - seconds: 5 - enabled: false - - platform: time_pattern - hours: /1 - minutes: "00" - seconds: "05" - - platform: time_pattern - hours: /1 - minutes: "30" - seconds: "05" - - platform: state - entity_id: - - sensor.pvopt_charge_start -condition: [] -action: - - if: - - condition: template - value_template: >- - {{ states('sensor.pvopt_charge_start') | as_datetime | as_local <= - today_at("23:59") }} - then: - - data: - days: - - chargeCurrent: >- - {% set direction = float(states('sensor.pvopt_charge_current'), - 0.0) %} {% set chargeAmps = min((max(direction, 0.0) | - round(method='floor')), 50)%} {{ chargeAmps }} - dischargeCurrent: >- - {% set direction = float(states('sensor.pvopt_charge_current'), - 0.0) %} {% set dischargeAmps = min((min(direction, 0.0) | abs | - round(method='floor')), 50) %} {{ dischargeAmps }} - chargeStartTime: >- - {% set direction = float(states('sensor.pvopt_charge_current'), - 0.0) %} {% set startChargeTime = '00:00' %} {% if direction >= - 0.0 -%} - {% set startChargeTime = (as_local(as_datetime(states('sensor.pvopt_charge_start')))|string)[11:16] %} - {%- endif %} {{ startChargeTime }} - chargeEndTime: >- - {% set direction = float(states('sensor.pvopt_charge_current'), - 0.0) %} {% set endChargeTime = '00:00' %} {% if direction >= 0.0 - -%} - {% set endChargeTime = (as_local(as_datetime(states('sensor.pvopt_charge_end')))|string)[11:16] %} - {%- endif %} {{ endChargeTime }} - dischargeStartTime: >- - {% set direction = float(states('sensor.pvopt_charge_current'), - 0.0) %} {% set startDischargeTime = '00:00' %} {% if direction < - 0.0 -%} - {% set startDischargeTime = (as_local(as_datetime(states('sensor.pvopt_charge_start')))|string)[11:16] %} - {%- endif %} {{ startDischargeTime }} - dischargeEndTime: >- - {% set direction = float(states('sensor.pvopt_charge_current'), - 0.0) %} {% set endDischargeTime = '00:00' %} {% if direction < - 0.0 -%} - {% set endDischargeTime = (as_local(as_datetime(states('sensor.pvopt_charge_end')))|string)[11:16] %} - {%- endif %} {{ endDischargeTime }} - - chargeCurrent: "0" - dischargeCurrent: "0" - chargeStartTime: "00:00" - chargeEndTime: "00:00" - dischargeStartTime: "00:00" - dischargeEndTime: "00:00" - - chargeCurrent: "0" - dischargeCurrent: "0" - chargeStartTime: "00:00" - chargeEndTime: "00:00" - dischargeStartTime: "00:00" - dischargeEndTime: "00:00" - config: - secret: <> - key_id: "<>" - username: <> - password: <> - plantId: "<>" - action: pyscript.solis_control - else: - - data: - days: - - chargeCurrent: "0" - dischargeCurrent: "0" - chargeStartTime: "00:00" - chargeEndTime: "00:00" - dischargeStartTime: "00:00" - dischargeEndTime: "00:00" - - chargeCurrent: "0" - dischargeCurrent: "0" - chargeStartTime: "00:00" - chargeEndTime: "00:00" - dischargeStartTime: "00:00" - dischargeEndTime: "00:00" - - chargeCurrent: "0" - dischargeCurrent: "0" - chargeStartTime: "00:00" - chargeEndTime: "00:00" - dischargeStartTime: "00:00" - dischargeEndTime: "00:00" - config: - secret: <> - key_id: "<>" - username: <> - password: <> - plantId: "<>" - action: pyscript.solis_control -mode: single -``` -

Configuration

diff --git a/apps/pv_opt/config/config.yaml b/apps/pv_opt/config/config.yaml index 23d6b51..e7276c6 100644 --- a/apps/pv_opt/config/config.yaml +++ b/apps/pv_opt/config/config.yaml @@ -318,39 +318,26 @@ pv_opt: inverter_type: SOLIS_CLOUD device_name: soliscloud + + soliscloud_username: !secret soliscloud_username + soliscloud_password: !secret soliscloud_password + soliscloud_key_id: !secret soliscloud_key_id + soliscloud_key_secret: !secret soliscloud_key_secret + soliscloud_plant_id: !secret soliscloud_plant_id battery_voltage: sensor.{device_name}_battery_voltage - # update_cycle_seconds: 15 + update_cycle_seconds: 0 maximum_dod_percent: sensor.{device_name}_force_discharge_soc id_consumption_today: sensor.{device_name}_daily_grid_energy_used # id_consumption: # - sensor.{device_name}_total_consumption_power - # - sensor.{device_name}_backup_load_power id_grid_import_today: sensor.{device_name}_daily_grid_energy_purchased id_grid_export_today: sensor.{device_name}_daily_on_grid_energy id_battery_soc: sensor.{device_name}_remaining_battery_capacity - # id_grid_import_today: sensor.{device_name}_grid_import_today - # id_grid_export_today: sensor.{device_name}_grid_export_today - - # id_battery_soc: sensor.{device_name}_battery_soc - # id_timed_charge_start_hours: number.{device_name}_timed_charge_start_hours - # id_timed_charge_start_minutes: number.{device_name}_timed_charge_start_minutes - # id_timed_charge_end_hours: number.{device_name}_timed_charge_end_hours - # id_timed_charge_end_minutes: number.{device_name}_timed_charge_end_minutes - # id_timed_charge_current: number.{device_name}_timed_charge_current - - # id_timed_discharge_start_hours: number.{device_name}_timed_discharge_start_hours - # id_timed_discharge_start_minutes: number.{device_name}_timed_discharge_start_minutes - # id_timed_discharge_end_hours: number.{device_name}_timed_discharge_end_hours - # id_timed_discharge_end_minutes: number.{device_name}_timed_discharge_end_minutes - # id_timed_discharge_current: number.{device_name}_timed_discharge_current - - # id_timed_charge_discharge_button: button.{device_name}_update_charge_discharge_times - # id_inverter_mode: select.{device_name}_energy_storage_control_switch # Tariff comparison # id_daily_solar: sensor.{device_name}_power_generation_today diff --git a/apps/pv_opt/pv_opt.py b/apps/pv_opt/pv_opt.py index 160b1f2..640201e 100644 --- a/apps/pv_opt/pv_opt.py +++ b/apps/pv_opt/pv_opt.py @@ -12,7 +12,7 @@ from numpy import nan import re -VERSION = "3.17.2" +VERSION = "3.18.0" OCTOPUS_PRODUCT_URL = r"https://api.octopus.energy/v1/products/" @@ -459,6 +459,8 @@ class PVOpt(hass.Hass): @ad.app_lock def initialize(self): self.config = {} + self.change_items = {} + self.config_state = {} self.log("") self.log(f"******************* PV Opt v{VERSION} *******************") self.log("") @@ -509,8 +511,6 @@ def initialize(self): if self.debug or self.args.get("list_entities", True): self._list_entities() - self.change_items = {} - self.config_state = {} self.timer_handle = None self.handles = {} self.mqtt_handles = {} diff --git a/apps/pv_opt/solis.py b/apps/pv_opt/solis.py index db7fd51..cbba11f 100644 --- a/apps/pv_opt/solis.py +++ b/apps/pv_opt/solis.py @@ -1,5 +1,22 @@ import pandas as pd import time +import hashlib +import hmac +import base64 +import json +import re +import requests +from http import HTTPStatus +from datetime import datetime, timezone + +URLS = { + "root": "https://www.soliscloud.com:13333", + "login": "/v2/api/login", + "control": "/v2/api/control", + "inverterList": "/v1/api/inverterList", + "atRead": "/v2/api/atRead", +} + TIMEFORMAT = "%H:%M" INVERTER_DEFS = { @@ -205,16 +222,24 @@ }, }, "SOLIS_CLOUD": { - "online": "sensor.{device_name}_temperature", + "bits": [ + "SelfUse", + "Timed", + "OffGrid", + "BatteryWake", + "Backup", + "GridCharge", + "FeedInPriority", + ], "default_config": { "maximum_dod_percent": "sensor.{device_name}_force_discharge_soc", "id_consumption_today": "sensor.{device_name}_daily_grid_energy_used", "id_grid_import_today": "sensor.{device_name}_daily_grid_energy_purchased", "id_grid_export_today": "sensor.{device_name}_daily_on_grid_energy", "id_battery_soc": "sensor.{device_name}_remaining_battery_capacity", - "supports_hold_soc": False, + "supports_hold_soc": True, "supports_forced_discharge": True, - "update_cycle_seconds": 300, + "update_cycle_seconds": 0, }, "brand_config": { "battery_voltage": "sensor.{device_name}_battery_voltage", @@ -223,6 +248,177 @@ } +class SolisCloud: + URLS = { + "root": "https://www.soliscloud.com:13333", + "login": "/v2/api/login", + "control": "/v2/api/control", + "inverterList": "/v1/api/inverterList", + "inverterDetail": "/v1/api/inverterDetail", + "atRead": "/v2/api/atRead", + } + + def __init__(self, username, password, key_id, key_secret, plant_id, **kwargs): + self.username = username + self.key_id = key_id + self.key_secret = key_secret + self.plant_id = plant_id + self.md5password = hashlib.md5(password.encode("utf-8")).hexdigest() + self.token = "" + self.log = kwargs.get("log", print) + + def get_body(self, **params): + body = "{" + for key in params: + body += f'"{key}":"{params[key]}",' + body = body[:-1] + "}" + return body + + def digest(self, body: str) -> str: + 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) + content_type = "application/json" + + 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 + hmac_obj = hmac.new(self.key_secret.encode("utf-8"), msg=encrypt_str.encode("utf-8"), digestmod=hashlib.sha1) + sign = base64.b64encode(hmac_obj.digest()) + authorization = "API " + str(self.key_id) + ":" + sign.decode("utf-8") + + header = { + "Content-MD5": content_md5, + "Content-Type": content_type, + "Date": date, + "Authorization": authorization, + } + return header + + @property + 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) + 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) + 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) + + if response.status_code == HTTPStatus.OK: + return response.json()["data"] + + @property + def is_online(self): + return self.inverter_details["state"] == 1 + + @property + def last_seen(self): + return pd.to_datetime(int(self.inverter_details["dataTimestamp"]), unit="ms") + + def read_code(self, cid): + if self.token == "": + self.login() + 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) + if response.status_code == HTTPStatus.OK: + return response.json()["data"]["msg"] + + def set_code(self, cid, value): + if self.token == "": + self.login() + + if self.token != "": + 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) + if response.status_code == HTTPStatus.OK: + return response.json() + + def login(self): + body = self.get_body(username=self.username, password=self.md5password) + header = self.header(body, self.URLS["login"]) + response = requests.post(self.URLS["root"] + self.URLS["login"], data=body, headers=header) + status = response.status_code + if status == HTTPStatus.OK: + result = response.json() + self.token = result["csrfToken"] + print("Logged in to SolisCloud OK") + + else: + print(status) + + def read_mode_switch(self): + bits = INVERTER_DEFS["SOLIS_CLOUD"]["bits"] + code = int(self.read_code("636")) + switches = {bit: (code & 2**i == 2**i) for i, bit in enumerate(bits)} + return {"code": code, "switches": switches} + + def timed_status(self, tz="GB"): + data = self.read_code("103").split(",") + return { + "charge": { + "current": float(data[0]), + "start": pd.Timestamp(data[2].split("-")[0], tz=tz), + "end": pd.Timestamp(data[2].split("-")[1], tz=tz), + }, + "discharge": { + "current": float(data[1]), + "start": pd.Timestamp(data[3].split("-")[0], tz=tz), + "end": pd.Timestamp(data[3].split("-")[1], tz=tz), + }, + } + + def read_backup_mode_soc(self): + return int(self.read_code("157")) + + def set_mode_switch(self, code): + return self.set_code("636", code) + + def get_time_string(self, time_status): + time_string = ",".join( + [ + str(int(time_status["charge"]["current"])), + str(int(time_status["discharge"]["current"])), + f'{time_status["charge"]["start"].strftime("%H:%M")}-{time_status["charge"]["end"].strftime("%H:%M")}', + f'{time_status["discharge"]["start"].strftime("%H:%M")}-{time_status["discharge"]["end"].strftime("%H:%M")}', + ] + ) + return f"{time_string},0,0,00:00-00:00,00:00-00:00,0,0,00:00-00:00,00:00-00:00" + + def set_timer(self, direction, start, end, current): + current_times = self.timed_status() + new_times = current_times.copy() + if start is not None: + new_times[direction]["start"] = start + if end is not None: + new_times[direction]["end"] = end + new_times[direction]["current"] = current + current_time_string = self.read_code(103) + new_time_string = self.get_time_string(new_times) + if new_time_string != current_time_string: + return self.set_code("103", new_time_string) + else: + return {"code": -1} + + class InverterController: def __init__(self, inverter_type, host) -> None: self.host = host @@ -243,20 +439,30 @@ def __init__(self, inverter_type, host) -> None: conf[item] = [z.replace("{device_name}", self.host.device_name) for z in defs[item]] else: conf[item] = defs[item] + if self.type == "SOLIS_CLOUD": + params = { + item: host.args.get(f"soliscloud_{item}") + for item in ["username", "password", "key_id", "key_secret", "plant_id"] + } + if all([x is not None for x in params.values()]): + self.cloud = SolisCloud(**params, log=self.log) + else: + raise Exception("Unable to create Solis Cloud controller") def is_online(self): - entity_id = INVERTER_DEFS[self.type].get("online", (None, None)) - if entity_id is not None: - entity_id = entity_id.replace("{device_name}", self.host.device_name) - return self.host.get_state_retry(entity_id) not in ["unknown", "unavailable"] + if self.type == "SOLIS_CLOUD": + return self.cloud.is_online else: - return True + entity_id = INVERTER_DEFS[self.type].get("online", (None, None)) + if entity_id is not None: + entity_id = entity_id.replace("{device_name}", self.host.device_name) + return self.host.get_state_retry(entity_id) not in ["unknown", "unavailable"] + else: + return True def enable_timed_mode(self): - if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": + if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_CORE_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CLOUD"]: self._solis_set_mode_switch(SelfUse=True, Timed=True, GridCharge=True, Backup=False) - elif self.type == "SOLIS_CLOUD": - pass else: self._unknown_inverter() @@ -271,7 +477,7 @@ def control_discharge(self, enable, **kwargs): self._control_charge_discharge("discharge", enable, **kwargs) def hold_soc(self, enable, soc=None, **kwargs): - if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": + if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_CORE_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CLOUD"]: start = kwargs.get("start", pd.Timestamp.now(tz=self.tz).floor("1min")) end = kwargs.get("end", pd.Timestamp.now(tz=self.tz).ceil("30min")) self._solis_control_charge_discharge( @@ -281,8 +487,6 @@ def hold_soc(self, enable, soc=None, **kwargs): end=end, power=0, ) - elif self.type == "SOLIS_CLOUD": - pass else: self._unknown_inverter() @@ -325,9 +529,7 @@ def hold_soc_old(self, enable, soc=None): @property def status(self): status = None - if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": - status = self._solis_state() - elif self.type == "SOLIS_CLOUD": + if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_CORE_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CLOUD"]: status = self._solis_state() return status @@ -335,10 +537,8 @@ def _monitor_target_soc(self, target_soc, mode="charge"): pass def _control_charge_discharge(self, direction, enable, **kwargs): - if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": + if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_CORE_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CLOUD"]: self._solis_control_charge_discharge(direction, enable, **kwargs) - elif self.type == "SOLIS_CLOUD": - pass def _solis_control_charge_discharge(self, direction, enable, **kwargs): status = self._solis_state() @@ -379,132 +579,146 @@ def _solis_control_charge_discharge(self, direction, enable, **kwargs): write_flag = True value_changed = False - for limit in times: - if times[limit] is not None: - for unit in ["hours", "minutes"]: - entity_id = self.host.config[f"id_timed_{direction}_{limit}_{unit}"] - if unit == "hours": - value = times[limit].hour - else: - value = times[limit].minute + if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CORE_MODBUS"]: + for limit in times: + if times[limit] is not None: + for unit in ["hours", "minutes"]: + entity_id = self.host.config[f"id_timed_{direction}_{limit}_{unit}"] + if unit == "hours": + value = times[limit].hour + else: + value = times[limit].minute - if self.type == "SOLIS_SOLAX_MODBUS": - changed, written = self.host.write_and_poll_value( - entity_id=entity_id, value=value, verbose=True - ) - elif self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": - changed, written = self._solis_write_time_register(direction, limit, unit, value) + if self.type == "SOLIS_SOLAX_MODBUS": + changed, written = self.host.write_and_poll_value( + entity_id=entity_id, value=value, verbose=True + ) + elif self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": + changed, written = self._solis_write_time_register(direction, limit, unit, value) + + else: + e = "Unknown inverter type" + self.log(e, level="ERROR") + raise Exception(e) + + if changed: + if written: + self.log(f"Wrote {direction} {limit} {unit} of {value} to inverter") + value_changed = True + else: + self.log( + f"Failed to write {direction} {limit} {unit} to inverter", + level="ERROR", + ) + write_flag = False + + if value_changed: + if self.type == "SOLIS_SOLAX_MODBUS" and write_flag: + entity_id = self.host.config["id_timed_charge_discharge_button"] + self.host.call_service("button/press", entity_id=entity_id) + time.sleep(0.5) + try: + time_pressed = pd.Timestamp(self.host.get_state_retry(entity_id)) + + dt = (pd.Timestamp.now(self.host.tz) - time_pressed).total_seconds() + if dt < 10: + self.log(f"Successfully pressed button {entity_id}") - else: - e = "Unknown inverter type" - self.log(e, level="ERROR") - raise Exception(e) - - if changed: - if written: - self.log(f"Wrote {direction} {limit} {unit} of {value} to inverter") - value_changed = True else: self.log( - f"Failed to write {direction} {limit} {unit} to inverter", - level="ERROR", + f"Failed to press button {entity_id}. Last pressed at {time_pressed.strftime(TIMEFORMAT)} ({dt:0.2f} seconds ago)" ) - write_flag = False + except: + self.log(f"Failed to press button {entity_id}: it appears to never have been pressed.") - if value_changed: - if self.type == "SOLIS_SOLAX_MODBUS" and write_flag: - entity_id = self.host.config["id_timed_charge_discharge_button"] - self.host.call_service("button/press", entity_id=entity_id) - time.sleep(0.5) - try: - time_pressed = pd.Timestamp(self.host.get_state_retry(entity_id)) - - dt = (pd.Timestamp.now(self.host.tz) - time_pressed).total_seconds() - if dt < 10: - self.log(f"Successfully pressed button {entity_id}") + else: + self.log("Inverter already at correct time settings") + + if power is not None: + entity_id = self.host.config[f"id_timed_{direction}_current"] + + current = abs(round(power / self.host.get_config("battery_voltage"), 1)) + current = min(current, self.host.get_config("battery_current_limit_amps")) + self.log(f"Power {power:0.0f} = {current:0.1f}A at {self.host.get_config('battery_voltage')}V") + if self.type == "SOLIS_SOLAX_MODBUS": + changed, written = self.host.write_and_poll_value(entity_id=entity_id, value=current, tolerance=1) + elif self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": + changed, written = self._solis_write_current_register(direction, current, tolerance=1) + else: + e = "Unknown inverter type" + self.log(e, level="ERROR") + raise Exception(e) + if changed: + if written: + self.log(f"Current {current}A written to inverter") else: - self.log( - f"Failed to press button {entity_id}. Last pressed at {time_pressed.strftime(TIMEFORMAT)} ({dt:0.2f} seconds ago)" - ) - except: - self.log(f"Failed to press button {entity_id}: it appears to never have been pressed.") - - else: - self.log("Inverter already at correct time settings") - - if power is not None: - entity_id = self.host.config[f"id_timed_{direction}_current"] + self.log(f"Failed to write {current} to inverter") + else: + self.log("Inverter already at correct current") - current = abs(round(power / self.host.get_config("battery_voltage"), 1)) + elif self.type == "SOLIS_CLOUD": + current = abs(round(power / self.host.get_config("battery_voltage"), 0)) current = min(current, self.host.get_config("battery_current_limit_amps")) - self.log(f"Power {power:0.0f} = {current:0.1f}A at {self.host.get_config('battery_voltage')}V") - if self.type == "SOLIS_SOLAX_MODBUS": - changed, written = self.host.write_and_poll_value(entity_id=entity_id, value=current, tolerance=1) - elif self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": - changed, written = self._solis_write_current_register(direction, current, tolerance=1) - else: - e = "Unknown inverter type" - self.log(e, level="ERROR") - raise Exception(e) - - if changed: - if written: - self.log(f"Current {current} written to inverter") - else: - self.log(f"Failed to write {current} to inverter") - else: - self.log("Inverter already at correct current") + self.log(f"Power {power:0.0f} = {current:0.0f}A at {self.host.get_config('battery_voltage')}V") + response = self.cloud.set_timer(direction, times["start"], times["end"], current) + if response["code"] == -1: + self.log("Inverter already at correct time and current settings") + elif response["code"] == 0: + self.log( + f"Wrote {direction} time of {times['start'].strftime('%H:%M')}-{times['end'].strftime('%H:%M')} to inverter" + ) + self.log(f"Current {current}A written to inverter") def _solis_set_mode_switch(self, **kwargs): # Read the mode switch - if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CORE_MODBUS"]: - if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_SOLARMAN": - status = self._solis_solax_solarman_mode_switch() + if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_SOLARMAN": + status = self._solis_solax_solarman_mode_switch() - elif self.type == "SOLIS_CORE_MODBUS": - status = self._solis_core_mode_switch() + elif self.type == "SOLIS_CORE_MODBUS": + status = self._solis_core_mode_switch() - switches = status["switches"] - if self.host.debug: - self.log(f">>> kwargs: {kwargs}") - self.log(">>> Solis switch status:") + elif self.type == "SOLIS_CLOUD": + status = self.cloud.read_mode_switch() - for switch in switches: - if switch in kwargs: - if self.host.debug: - self.log(f">>> {switch}: {kwargs[switch]}") - switches[switch] = kwargs[switch] + switches = status["switches"] + if self.host.debug: + self.log(f">>> kwargs: {kwargs}") + self.log(">>> Solis switch status:") - elif self.type == "SOLIS_CLOUD": - pass + for switch in switches: + if switch in kwargs: + if self.host.debug: + self.log(f">>> {switch}: {kwargs[switch]}") + switches[switch] = kwargs[switch] # Set the mode switch - if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CORE_MODBUS"]: - bits = INVERTER_DEFS[self.type]["bits"] - bin_list = [2**i * switches[bit] for i, bit in enumerate(bits)] - code = sum(bin_list) + bits = INVERTER_DEFS[self.type]["bits"] + bin_list = [2**i * switches[bit] for i, bit in enumerate(bits)] + code = sum(bin_list) + + if self.type != "SOLIS_CLOUD": entity_id = self.host.config["id_inverter_mode"] - if self.type == "SOLIS_SOLAX_MODBUS": - entity_modes = self.host.get_state_retry(entity_id, attribute="options") - modes = {INVERTER_DEFS[self.type]["codes"].get(mode): mode for mode in entity_modes} - # mode = INVERTER_DEFS[self.type]["modes"].get(code) - mode = modes.get(code) - if self.host.debug: - self.log(f">>> Inverter Code: {code}") - self.log(f">>> Entity modes: {entity_modes}") - self.log(f">>> Modes: {modes}") - self.log(f">>> Inverter Mode: {mode}") + if self.type == "SOLIS_SOLAX_MODBUS": + entity_modes = self.host.get_state_retry(entity_id, attribute="options") + modes = {INVERTER_DEFS[self.type]["codes"].get(mode): mode for mode in entity_modes} + # mode = INVERTER_DEFS[self.type]["modes"].get(code) + mode = modes.get(code) + if self.host.debug: + self.log(f">>> Inverter Code: {code}") + self.log(f">>> Entity modes: {entity_modes}") + self.log(f">>> Modes: {modes}") + self.log(f">>> Inverter Mode: {mode}") - self.host.set_select("inverter_mode", mode) + self.host.set_select("inverter_mode", mode) - elif self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": - address = INVERTER_DEFS[self.type]["registers"]["storage_control_switch"] - self._solis_write_holding_register(address=address, value=code, entity_id=entity_id) + elif self.type == "SOLIS_CORE_MODBUS" or self.type == "SOLIS_SOLARMAN": + address = INVERTER_DEFS[self.type]["registers"]["storage_control_switch"] + self._solis_write_holding_register(address=address, value=code, entity_id=entity_id) elif self.type == "SOLIS_CLOUD": - pass + self.cloud.set_mode_switch(code) def _solis_solax_solarman_mode_switch(self): inverter_mode = self.host.get_state_retry(entity_id=self.host.config["id_inverter_mode"]) @@ -529,28 +743,35 @@ def _solis_core_mode_switch(self): def _solis_state(self): limits = ["start", "end"] + if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_SOLARMAN": status = self._solis_solax_solarman_mode_switch() elif self.type == "SOLIS_CORE_MODBUS": status = self._solis_core_mode_switch() - else: - status = {} + elif self.type == "SOLIS_CLOUD": + status = self.cloud.read_mode_switch() + + if self.type in ["SOLIS_SOLAX_MODBUS", "SOLIS_SOLARMAN", "SOLIS_CORE_MODBUS"]: + for direction in ["charge", "discharge"]: + status[direction] = {} + for limit in limits: + states = {} + for unit in ["hours", "minutes"]: + entity_id = self.host.config[f"id_timed_{direction}_{limit}_{unit}"] + states[unit] = int(float(self.host.get_state_retry(entity_id=entity_id))) + status[direction][limit] = pd.Timestamp( + f"{states['hours']:02d}:{states['minutes']:02d}", tz=self.host.tz + ) - for direction in ["charge", "discharge"]: - status[direction] = {} - for limit in limits: - states = {} - for unit in ["hours", "minutes"]: - entity_id = self.host.config[f"id_timed_{direction}_{limit}_{unit}"] - states[unit] = int(float(self.host.get_state_retry(entity_id=entity_id))) - status[direction][limit] = pd.Timestamp( - f"{states['hours']:02d}:{states['minutes']:02d}", tz=self.host.tz - ) - time_now = pd.Timestamp.now(tz=self.tz) status[direction]["current"] = float( self.host.get_state_retry(self.host.config[f"id_timed_{direction}_current"]) ) + elif self.type == "SOLIS_CLOUD": + status = status | self.cloud.timed_status(tz=self.host.tz) + + time_now = pd.Timestamp.now(tz=self.tz) + for direction in ["charge", "discharge"]: status[direction]["active"] = ( time_now >= status[direction]["start"] and time_now < status[direction]["end"] @@ -562,6 +783,8 @@ def _solis_state(self): status["hold_soc"] = {"active": status["switches"]["Backup"]} if self.type == "SOLIS_SOLAX_MODBUS" or self.type == "SOLIS_CORE_MODBUS": status["hold_soc"]["soc"] = self.host.get_config("id_backup_mode_soc") + elif self.type == "SOLIS_CLOUD": + status["hold_soc"]["soc"] = self.cloud.read_backup_mode_soc() else: status["hold_soc"]["soc"] = None