diff --git a/.ruff.toml b/.ruff.toml index 0163d6c..159e67e 100644 --- a/.ruff.toml +++ b/.ruff.toml @@ -23,6 +23,7 @@ ignore = [ "T201", # Prints are allowed in examples "PLR0913", "FBT002", + "N999", ] [lint.flake8-pytest-style] diff --git a/docs/openapiv1.md b/docs/openapiv1.md index da21723..866302d 100644 --- a/docs/openapiv1.md +++ b/docs/openapiv1.md @@ -2,7 +2,7 @@ This version of the API follows the newer [OpenAPI V1 API](https://www.showdoc.com.cn/262556420217021/0) Growatt has made available. -It extends our ["Legacy" ShinePhone](./shinephone.md) so methods from [there](./shinephone.md#methods) should be available, but it's safer to rely on the functions described in this file where possible. +It extends our ["Legacy" ShinePhone](./shinephone.md) so methods from [there](./shinephone.md#methods) should be available, but it's safer to rely on the methods described in this file where possible. ## Usage @@ -33,6 +33,36 @@ Methods that work across all device types. | `api.plant_energy_history(plant_id, start_date, end_date, time_unit, page, perpage)` | plant_id: String, start_date: Date, end_date: Date, time_unit: String, page: Int, perpage: Int | Get historical energy data for a plant for multiple days/months/years. | | `api.device_list(plant_id)` | plant_id: String | Get a list of devices in specified plant. | +#### Devices + +Devices offer a generic way to interact with your device using the V1 API without needing to provide your S/N every time. And can be used instead of the more specific device methods in the API class. + +```python +import growattServer +from growattServer.open_api_v1.devices import Sph, Min + +api = growattServer.OpenApiV1(token="YOUR_API_TOKEN") + +my_inverter = Sph(api, 'YOUR_DEVICE_SERIAL_NUMBER') # or Min(api, 'YOUR_DEVICE_SERIAL_NUMBER') +my_inverter.detail() +my_inverter.energy() +my_inverter.energy_history() +my_inverter.read_parameter() +my_inverter.write_parameter() +``` + +| Method | Arguments | Description | +|:---|:---|:---| +| `device.energy()` | None | Get current energy data for any inverter, including power and energy values. | +| `device.detail()` | None | Get detailed data for any inverter. | +| `device.energy_history(start_date=None, end_date=None, timezone=None, page=None, limit=None)` | start_date: Date, end_date: Date, timezone: String, page: Int, limit: Int | Get energy history data for any inverter (7-day max range). | +| `device.read_parameter(parameter_id, start_address=None, end_address=None)` | parameter_id: String, start_address: Int, end_address: Int | Read a specific setting for any inverter. | +| `device.write_parameter(parameter_id, parameter_values)` | parameter_id: String, parameter_values: Dict/Array | Set parameters on any inverter. Parameter values can be a single value, a list, or a dictionary. | + +For more details see: [OpenApiV1 Devices](./openapiv1/devices.md) + +The remaining methods below all actually use these device methods. + #### MIN Methods Methods for MIN devices (type 7). @@ -68,12 +98,12 @@ Convenience methods that wrap the core SPH methods above for common use cases. |:---|:---|:---| | `api.sph_write_ac_charge_times(...)` | device_sn, charge_power, charge_stop_soc, mains_enabled, periods | Helper: wraps `sph_write_parameter()` with type `mix_ac_charge_time_period`. see: [details](./openapiv1/sph_settings.md) | | `api.sph_write_ac_discharge_times(...)` | device_sn, discharge_power, discharge_stop_soc, periods | Helper: wraps `sph_write_parameter()` with type `mix_ac_discharge_time_period`. see: [details](./openapiv1/sph_settings.md) | -| `api.sph_read_ac_charge_times(...)` | device_sn (optional), settings_data (optional) | Helper: parses charge config from `sph_detail()` response. see: [details](./openapiv1/sph_settings.md) | -| `api.sph_read_ac_discharge_times(...)` | device_sn (optional), settings_data (optional) | Helper: parses discharge config from `sph_detail()` response. see: [details](./openapiv1/sph_settings.md) | +| `api.sph_read_ac_charge_times(...)` | device_sn, settings_data (optional) | Helper: parses charge config from `sph_detail()` response. see: [details](./openapiv1/sph_settings.md) | +| `api.sph_read_ac_discharge_times(...)` | device_sn, settings_data (optional) | Helper: parses discharge config from `sph_detail()` response. see: [details](./openapiv1/sph_settings.md) | #### Classic methods -Methods from [classic API](./shinephone.md#methods) should be available, but it's safer to rely on the functions described in this section where possible. There is no guarantee that the classic API methods will work, or remain stable through updates. +Methods from [classic API](./shinephone.md#methods) should be available, but it's safer to rely on the methods described in this section where possible. There is no guarantee that the classic API methods will work, or remain stable through updates. ### Variables diff --git a/docs/openapiv1/devices.md b/docs/openapiv1/devices.md new file mode 100644 index 0000000..89d1231 --- /dev/null +++ b/docs/openapiv1/devices.md @@ -0,0 +1,46 @@ +# OpenAPI V1 - Devices + +Devices offer a generic way to interact with your device using the V1 API without needing to provide your S/N every time. And can be used instead of the more specific device functions in the API class. + +```python +import growattServer +from growattServer.open_api_v1.devices import Sph, Min + +api = growattServer.OpenApiV1(token="YOUR_API_TOKEN") + +device = Sph(api, 'YOUR_DEVICE_SERIAL_NUMBER') # or Min(api, 'YOUR_DEVICE_SERIAL_NUMBER') +device.detail() +device.energy() +device.energy_history() +device.read_parameter() +device.write_parameter() +``` + +If you do not know your devices type, but do have their type id this method will provide you with the correct device class to use + +``` +import growattServer + +api = growattServer.OpenApiV1(token="YOUR_API_TOKEN") +device = api.get_device(device_sn, device_type) +if device is not None: + device.detail() + device.energy() + device.energy_history() + device.read_parameter() + device.write_parameter() +``` + +The basic methods are described here + +| Method | Arguments | Description | +|:---|:---|:---| +| `device.energy()` | None | Get current energy data for any inverter, including power and energy values. | +| `device.detail()` | None | Get detailed data for any inverter. | +| `device.energy_history(start_date=None, end_date=None, timezone=None, page=None, limit=None)` | start_date: Date, end_date: Date, timezone: String, page: Int, limit: Int | Get energy history data for any inverter (7-day max range). | +| `device.read_parameter(parameter_id, start_address=None, end_address=None)` | parameter_id: String, start_address: Int, end_address: Int | Read a specific setting for any inverter. | +| `device.write_parameter(parameter_id, parameter_values)` | parameter_id: String, parameter_values: Dict/Array | Set parameters on any inverter. Parameter values can be a single value, a list, or a dictionary. | + +However some device classes harbor more methods, which will be described in their respective readmes: +- [SPH/MIX](./devices/sph.md) +- [Min/TLX](./devices/min.md) \ No newline at end of file diff --git a/docs/openapiv1/devices/min.md b/docs/openapiv1/devices/min.md new file mode 100644 index 0000000..0152150 --- /dev/null +++ b/docs/openapiv1/devices/min.md @@ -0,0 +1,14 @@ +# OpenAPI V1 - Min/TLX Device (device type 7) + +The Min device offers the following methods + +| Method | Arguments | Description | +|:---|:---|:---| +| `device.energy()` | None | Get current energy data for a min inverter, including power and energy values. | +| `device.detail()` | None | Get detailed data for a min inverter. | +| `device.energy_history(start_date=None, end_date=None, timezone=None, page=None, limit=None)` | start_date: Date, end_date: Date, timezone: String, page: Int, limit: Int | Get energy history data for a min inverter (7-day max range). | +| `device.settings()` | None | Get all settings for a min inverter. | +| `device.read_parameter(parameter_id, start_address=None, end_address=None)` | parameter_id: String, start_address: Int, end_address: Int | Read a specific setting for a min inverter. see: [details](../min_tlx_settings.md) | +| `device.write_parameter(parameter_id, parameter_values)` | parameter_id: String, parameter_values: Dict/Array | Set parameters on a min inverter. Parameter values can be a single value, a list, or a dictionary. see: [details](../min_tlx_settings.md) | +| `device.write_time_segment(segment_id, batt_mode, start_time, end_time, enabled=True)` | segment_id: Int, batt_mode: Int <0=load priority, 1=battery priority, 2=grid priority>, start_time: datetime.time, end_time: datetime.time, enabled: Bool | Update a specific time segment for a min inverter. see: [details](../min_tlx_settings.md) | +| `device.read_time_segments(settings_data=None)` | settings_data: Dict | Read all time segments from a MIN inverter. Optionally pass settings_data to avoid redundant API calls. see: [details](../min_tlx_settings.md) | \ No newline at end of file diff --git a/docs/openapiv1/devices/sph.md b/docs/openapiv1/devices/sph.md new file mode 100644 index 0000000..670d45f --- /dev/null +++ b/docs/openapiv1/devices/sph.md @@ -0,0 +1,22 @@ +# OpenAPI V1 - SPH/MIX Device (device type 5) + +The SPH device offers the following methods + +| Method | Arguments | Description | +|:---|:---|:---| +| `device.detail()` | None | Get detailed data and settings for an SPH hybrid inverter. see: [details](../sph_settings.md) | +| `device.energy()` | None | Get current energy data for an SPH inverter, including power and energy values. | +| `device.energy_history(start_date=None, end_date=None, timezone=None, page=None, limit=None)` | start_date: Date, end_date: Date, timezone: String, page: Int, limit: Int | Get energy history data for an SPH inverter (7-day max range). | +| `device.read_parameter(parameter_id=None, start_address=None, end_address=None)` | parameter_id: String (optional), start_address: Int (optional), end_address: Int (optional) | Read a specific parameter (only pv_on_off supported). see: [details](../sph_settings.md) | +| `device.write_parameter(parameter_id, parameter_values)` | parameter_id: String, parameter_values: Dict/Array | Set parameters on an SPH inverter. see: [details](../sph_settings.md) | + +#### SPH Helper Methods + +Convenience methods that wrap the core SPH methods above for common use cases. + +| Method | Arguments | Description | +|:---|:---|:---| +| `device.write_ac_charge_times(...)` | device_sn, charge_power, charge_stop_soc, mains_enabled, periods | Helper: wraps `sph_write_parameter()` with type `mix_ac_charge_time_period`. see: [details](../sph_settings.md) | +| `device.write_ac_discharge_times(...)` | device_sn, discharge_power, discharge_stop_soc, periods | Helper: wraps `sph_write_parameter()` with type `mix_ac_discharge_time_period`. see: [details](../sph_settings.md) | +| `device.read_ac_charge_times(...)` | device_sn (optional), settings_data (optional) | Helper: parses charge config from `sph_detail()` response. see: [details](../sph_settings.md) | +| `device.read_ac_discharge_times(...)` | device_sn (optional), settings_data (optional) | Helper: parses discharge config from `sph_detail()` response. see: [details](../sph_settings.md) | \ No newline at end of file diff --git a/docs/openapiv1/sph_settings.md b/docs/openapiv1/sph_settings.md index 6d46df1..4ec37b2 100644 --- a/docs/openapiv1/sph_settings.md +++ b/docs/openapiv1/sph_settings.md @@ -184,7 +184,7 @@ For SPH (hybrid inverter) systems, the public V1 API provides methods to read an ### Read: `api.sph_read_ac_charge_times` * parameters: - * `device_sn`: The device serial number (not used if settings_data is provided) + * `device_sn`: The device serial number * `settings_data`: Settings data from sph_detail() (not used if device_sn is provided) * note: Either `device_sn` or `settings_data` must be provided * returns: Dict with `charge_power`, `charge_stop_soc`, `mains_enabled`, and `periods` list diff --git a/examples/sph_example.py b/examples/sph_example.py index 03fe716..9f110a0 100644 --- a/examples/sph_example.py +++ b/examples/sph_example.py @@ -76,6 +76,7 @@ # Read AC charge time periods using helper function and inverter_data to avoid rate limiting charge_config = api.sph_read_ac_charge_times( + device_sn=inverter_sn, settings_data=inverter_data, ) print("AC Charge Configuration:") @@ -86,6 +87,7 @@ # Read AC discharge time periods using helper function and inverter_data to avoid rate limiting discharge_config = api.sph_read_ac_discharge_times( + device_sn=inverter_sn, settings_data=inverter_data, ) print("AC Discharge Configuration:") diff --git a/growattServer/__init__.py b/growattServer/__init__.py index b057305..41d839a 100755 --- a/growattServer/__init__.py +++ b/growattServer/__init__.py @@ -1,7 +1,6 @@ #!/usr/bin/env python3 """growattServer package exports.""" -# ruff: noqa: N999 from .base_api import GrowattApi, Timespan, hash_password from .exceptions import GrowattError, GrowattParameterError, GrowattV1ApiError from .open_api_v1 import DeviceType, OpenApiV1 diff --git a/growattServer/open_api_v1.py b/growattServer/open_api_v1/__init__.py similarity index 53% rename from growattServer/open_api_v1.py rename to growattServer/open_api_v1/__init__.py index 19a49bc..6ec2bd4 100644 --- a/growattServer/open_api_v1.py +++ b/growattServer/open_api_v1/__init__.py @@ -1,12 +1,13 @@ """OpenApi V1 extensions for Growatt API client.""" - import platform import warnings -from datetime import UTC, date, datetime, timedelta +from datetime import UTC, date, datetime from enum import Enum -from . import GrowattApi -from .exceptions import GrowattParameterError, GrowattV1ApiError +from growattServer import GrowattApi +from growattServer.exceptions import GrowattV1ApiError + +from .devices import AbstractDevice, Min, Sph class DeviceType(Enum): @@ -16,9 +17,9 @@ class DeviceType(Enum): STORAGE = 2 OTHER = 3 MAX = 4 - SPH = 5 # (MIX) + SPH = Sph.DEVICE_TYPE_ID # (MIX) SPA = 6 - MIN = 7 + MIN = Min.DEVICE_TYPE_ID PCS = 8 HPS = 9 PBD = 10 @@ -57,7 +58,7 @@ def __init__(self, token) -> None: # Set up authentication for V1 API using the provided token self.session.headers.update({"token": token}) - def _process_response(self, response, operation_name="API operation"): + def process_response(self, response, operation_name="API operation"): """ Process API response and handle errors. @@ -81,7 +82,7 @@ def _process_response(self, response, operation_name="API operation"): ) return response.get("data") - def _get_url(self, page): + def get_url(self, page): """Return the page URL for the v1 API.""" return self.api_url + page @@ -107,11 +108,11 @@ def plant_list(self): # Make the request response = self.session.get( - url=self._get_url("plant/list"), + url=self.get_url("plant/list"), data=request_data ) - return self._process_response(response.json(), "getting plant list") + return self.process_response(response.json(), "getting plant list") def plant_details(self, plant_id): """ @@ -129,11 +130,11 @@ def plant_details(self, plant_id): """ response = self.session.get( - self._get_url("plant/details"), + self.get_url("plant/details"), params={"plant_id": plant_id} ) - return self._process_response(response.json(), "getting plant details") + return self.process_response(response.json(), "getting plant details") def plant_energy_overview(self, plant_id): """ @@ -151,11 +152,11 @@ def plant_energy_overview(self, plant_id): """ response = self.session.get( - self._get_url("plant/data"), + self.get_url("plant/data"), params={"plant_id": plant_id} ) - return self._process_response(response.json(), "getting plant energy overview") + return self.process_response(response.json(), "getting plant energy overview") def plant_power_overview(self, plant_id: int, day: str | date | None = None) -> dict: """ @@ -189,14 +190,14 @@ def plant_power_overview(self, plant_id: int, day: str | date | None = None) -> day = datetime.now(UTC).date() response = self.session.get( - self._get_url("plant/power"), + self.get_url("plant/power"), params={ "plant_id": plant_id, "date": day, } ) - return self._process_response(response.json(), "getting plant power overview") + return self.process_response(response.json(), "getting plant power overview") def plant_energy_history(self, plant_id, start_date=None, end_date=None, time_unit="day", page=None, perpage=None): """ @@ -247,7 +248,7 @@ def plant_energy_history(self, plant_id, start_date=None, end_date=None, time_un "Date interval must not exceed 20 years in 'year' mode.", RuntimeWarning, stacklevel=2) response = self.session.get( - self._get_url("plant/energy"), + self.get_url("plant/energy"), params={ "plant_id": plant_id, "start_date": start_date.strftime("%Y-%m-%d"), @@ -258,7 +259,7 @@ def plant_energy_history(self, plant_id, start_date=None, end_date=None, time_un } ) - return self._process_response(response.json(), "getting plant energy history") + return self.process_response(response.json(), "getting plant energy history") def device_list(self, plant_id): """ @@ -306,14 +307,25 @@ def device_list(self, plant_id): """ response = self.session.get( - url=self._get_url("device/list"), + url=self.get_url("device/list"), params={ "plant_id": plant_id, "page": "", "perpage": "", }, ) - return self._process_response(response.json(), "getting device list") + return self.process_response(response.json(), "getting device list") + + def get_device(self, device_sn: str, device_type: int) -> AbstractDevice|None: + """Get the device class by serial number and device_type id.""" + match device_type: + case Sph.DEVICE_TYPE_ID: + return Sph(self, device_sn) + case Min.DEVICE_TYPE_ID: + return Min(self, device_sn) + case _: + warnings.warn(f"Device for type id: {device_type} has not been implemented yet.", stacklevel=2) + return None def min_detail(self, device_sn): """ @@ -330,14 +342,7 @@ def min_detail(self, device_sn): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - response = self.session.get( - self._get_url("device/tlx/tlx_data_info"), - params={ - "device_sn": device_sn - } - ) - - return self._process_response(response.json(), "getting MIN inverter details") + return Min(self, device_sn).detail() def min_energy(self, device_sn): """ @@ -354,14 +359,7 @@ def min_energy(self, device_sn): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - response = self.session.post( - url=self._get_url("device/tlx/tlx_last_data"), - data={ - "tlx_sn": device_sn, - }, - ) - - return self._process_response(response.json(), "getting MIN inverter energy data") + return Min(self, device_sn).energy() def min_energy_history(self, device_sn, start_date=None, end_date=None, timezone=None, page=None, limit=None): """ @@ -384,31 +382,7 @@ def min_energy_history(self, device_sn, start_date=None, end_date=None, timezone requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if start_date is None and end_date is None: - start_date = datetime.now(timezone.utc).date() - end_date = datetime.now(timezone.utc).date() - elif start_date is None: - start_date = end_date - elif end_date is None: - end_date = start_date - - # check interval validity - if end_date - start_date > timedelta(days=7): - raise GrowattParameterError("date interval must not exceed 7 days") - - response = self.session.post( - url=self._get_url("device/tlx/tlx_data"), - data={ - "tlx_sn": device_sn, - "start_date": start_date.strftime("%Y-%m-%d"), - "end_date": end_date.strftime("%Y-%m-%d"), - "timezone_id": timezone, - "page": page, - "perpage": limit, - } - ) - - return self._process_response(response.json(), "getting MIN inverter energy history") + return Min(self, device_sn).energy_history(start_date, end_date, timezone, page, limit) def min_settings(self, device_sn): """ @@ -425,14 +399,7 @@ def min_settings(self, device_sn): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - response = self.session.get( - self._get_url("device/tlx/tlx_set_info"), - params={ - "device_sn": device_sn - } - ) - - return self._process_response(response.json(), "getting MIN inverter settings") + return Min(self, device_sn).settings(device_sn) def min_read_parameter(self, device_sn, parameter_id, start_address=None, end_address=None): """ @@ -453,36 +420,7 @@ def min_read_parameter(self, device_sn, parameter_id, start_address=None, end_ad requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if parameter_id is None and start_address is None: - raise GrowattParameterError( - "specify either parameter_id or start_address/end_address") - if parameter_id is not None and start_address is not None: - raise GrowattParameterError( - "specify either parameter_id or start_address/end_address - not both." - ) - if parameter_id is not None: - # named parameter - start_address = 0 - end_address = 0 - else: - # using register-number mode - parameter_id = "set_any_reg" - if start_address is None: - start_address = end_address - if end_address is None: - end_address = start_address - - response = self.session.post( - self._get_url("readMinParam"), - data={ - "device_sn": device_sn, - "paramId": parameter_id, - "startAddr": start_address, - "endAddr": end_address, - } - ) - - return self._process_response(response.json(), f"reading parameter {parameter_id}") + return Min(self, device_sn).read_parameter(parameter_id, start_address, end_address) def min_write_parameter(self, device_sn, parameter_id, parameter_values=None): """ @@ -504,44 +442,7 @@ def min_write_parameter(self, device_sn, parameter_id, parameter_values=None): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - # Initialize all parameters as empty strings - max_min_params = 19 - parameters = dict.fromkeys(range(1, max_min_params + 1), "") - - # Process parameter values based on type - if parameter_values is not None: - if isinstance(parameter_values, (str, int, float, bool)): - # Single value goes to param1 - parameters[1] = str(parameter_values) - elif isinstance(parameter_values, list): - # List of values go to sequential params - for i, value in enumerate(parameter_values, 1): - if i <= max_min_params: # Only use up to max_min_params parameters - parameters[i] = str(value) - elif isinstance(parameter_values, dict): - # Dict maps param positions to values - for pos_raw, value in parameter_values.items(): - pos = int(pos_raw) if not isinstance(pos_raw, int) else pos_raw - if 1 <= pos <= max_min_params: # Validate parameter positions - parameters[pos] = str(value) - - # IMPORTANT: Create a data dictionary with ALL parameters explicitly included - request_data = { - "tlx_sn": device_sn, - "type": parameter_id - } - - # Add all MIN parameters to the request - for i in range(1, max_min_params + 1): - request_data[f"param{i}"] = str(parameters[i]) - - # Send the request - response = self.session.post( - self._get_url("tlxSet"), - data=request_data - ) - - return self._process_response(response.json(), f"writing parameter {parameter_id}") + return Min(self, device_sn).write_parameter(parameter_id, parameter_values) def min_write_time_segment(self, device_sn, segment_id, batt_mode, start_time, end_time, enabled=True): """ @@ -564,43 +465,7 @@ def min_write_time_segment(self, device_sn, segment_id, batt_mode, start_time, e requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - max_min_params = 19 - max_min_segments = 9 - max_batt_mode = 2 - - if not 1 <= segment_id <= max_min_segments: - msg = f"segment_id must be between 1 and {max_min_segments}" - raise GrowattParameterError(msg) - - if not 0 <= batt_mode <= max_batt_mode: - msg = f"batt_mode must be between 0 and {max_batt_mode}" - raise GrowattParameterError(msg) - - # Initialize ALL 19 parameters as empty strings, not just the ones we need - all_params = { - "tlx_sn": device_sn, - "type": f"time_segment{segment_id}" - } - - # Add param1 through param19, setting the values we need - all_params["param1"] = str(batt_mode) - all_params["param2"] = str(start_time.hour) - all_params["param3"] = str(start_time.minute) - all_params["param4"] = str(end_time.hour) - all_params["param5"] = str(end_time.minute) - all_params["param6"] = "1" if enabled else "0" - - # Add empty strings for all unused parameters - for i in range(7, max_min_params + 1): - all_params[f"param{i}"] = "" - - # Send the request - response = self.session.post( - self._get_url("tlxSet"), - data=all_params - ) - - return self._process_response(response.json(), f"writing time segment {segment_id}") + return Min(self, device_sn).write_time_segment(segment_id, batt_mode, start_time, end_time, enabled) def min_read_time_segments(self, device_sn, settings_data=None): """ @@ -640,81 +505,7 @@ def min_read_time_segments(self, device_sn, settings_data=None): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - # Process the settings data - if settings_data is None: - # Fetch settings if not provided - settings_data = self.min_settings(device_sn=device_sn) - - # Define mode names - mode_names = { - 0: "Load First", - 1: "Battery First", - 2: "Grid First" - } - - segments = [] - - # Process each time segment - for i in range(1, 10): # Segments 1-9 - # Get raw time values - start_time_raw = settings_data.get(f"forcedTimeStart{i}", "0:0") - end_time_raw = settings_data.get(f"forcedTimeStop{i}", "0:0") - - # Handle 'null' string values - if start_time_raw == "null" or not start_time_raw: - start_time_raw = "0:0" - if end_time_raw == "null" or not end_time_raw: - end_time_raw = "0:0" - - # Format times with leading zeros (HH:MM) - try: - start_parts = start_time_raw.split(":") - start_hour = int(start_parts[0]) - start_min = int(start_parts[1]) - start_time = f"{start_hour:02d}:{start_min:02d}" - except (ValueError, IndexError): - start_time = "00:00" - - try: - end_parts = end_time_raw.split(":") - end_hour = int(end_parts[0]) - end_min = int(end_parts[1]) - end_time = f"{end_hour:02d}:{end_min:02d}" - except (ValueError, IndexError): - end_time = "00:00" - - # Get the mode value safely - mode_raw = settings_data.get(f"time{i}Mode") - if mode_raw == "null" or mode_raw is None: - batt_mode = None - else: - try: - batt_mode = int(mode_raw) - except (ValueError, TypeError): - batt_mode = None - - # Get the enabled status safely - enabled_raw = settings_data.get(f"forcedStopSwitch{i}", 0) - if enabled_raw == "null" or enabled_raw is None: - enabled = False - else: - try: - enabled = int(enabled_raw) == 1 - except (ValueError, TypeError): - enabled = False - - segment = { - "segment_id": i, - "batt_mode": batt_mode, - "mode_name": mode_names.get(batt_mode, "Unknown"), - "start_time": start_time, - "end_time": end_time, - "enabled": enabled - } - - segments.append(segment) - - return segments + return Min(self, device_sn).read_time_segments(settings_data) # SPH Device Methods (Device Type 5) @@ -733,15 +524,7 @@ def sph_detail(self, device_sn): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - # API: https://www.showdoc.com.cn/262556420217021/6129763571291058 - response = self.session.get( - self._get_url("device/mix/mix_data_info"), - params={ - "device_sn": device_sn - } - ) - - return self._process_response(response.json(), "getting SPH inverter details") + return Sph(self, device_sn).detail() def sph_energy(self, device_sn): """ @@ -758,15 +541,7 @@ def sph_energy(self, device_sn): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - # API: https://www.showdoc.com.cn/262556420217021/6129764475556048 - response = self.session.post( - url=self._get_url("device/mix/mix_last_data"), - data={ - "mix_sn": device_sn, - }, - ) - - return self._process_response(response.json(), "getting SPH inverter energy data") + return Sph(self, device_sn).energy() def sph_energy_history(self, device_sn, start_date=None, end_date=None, timezone=None, page=None, limit=None): """ @@ -789,32 +564,7 @@ def sph_energy_history(self, device_sn, start_date=None, end_date=None, timezone requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if start_date is None and end_date is None: - start_date = datetime.now(timezone.utc).date() - end_date = datetime.now(timezone.utc).date() - elif start_date is None: - start_date = end_date - elif end_date is None: - end_date = start_date - - # check interval validity - if end_date - start_date > timedelta(days=7): - raise GrowattParameterError("date interval must not exceed 7 days") - - # API: https://www.showdoc.com.cn/262556420217021/6129765461123058 - response = self.session.post( - url=self._get_url("device/mix/mix_data"), - data={ - "mix_sn": device_sn, - "start_date": start_date.strftime("%Y-%m-%d"), - "end_date": end_date.strftime("%Y-%m-%d"), - "timezone_id": timezone, - "page": page, - "perpage": limit, - } - ) - - return self._process_response(response.json(), "getting SPH inverter energy history") + return Sph(self, device_sn).energy_history(start_date, end_date, timezone, page, limit) def sph_read_parameter(self, device_sn, parameter_id=None, start_address=None, end_address=None): """ @@ -835,33 +585,7 @@ def sph_read_parameter(self, device_sn, parameter_id=None, start_address=None, e requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if parameter_id is None and start_address is None: - raise GrowattParameterError( - "specify either parameter_id or start_address/end_address") - if parameter_id is not None and start_address is not None: - raise GrowattParameterError( - "specify either parameter_id or start_address/end_address - not both." - ) - if parameter_id is not None: - # named parameter - start_address = 0 - end_address = 0 - else: - # address range - parameter_id = "set_any_reg" - - # API: https://www.showdoc.com.cn/262556420217021/6129766954561259 - response = self.session.post( - self._get_url("readMixParam"), - data={ - "device_sn": device_sn, - "paramId": parameter_id, - "startAddr": start_address, - "endAddr": end_address - } - ) - - return self._process_response(response.json(), f"reading parameter {parameter_id}") + return Sph(self, device_sn).read_parameter(parameter_id, start_address, end_address) def sph_write_parameter(self, device_sn, parameter_id, parameter_values=None): """ @@ -883,44 +607,7 @@ def sph_write_parameter(self, device_sn, parameter_id, parameter_values=None): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - # Initialize all parameters as empty strings (API uses param1-param18) - max_sph_params = 18 - parameters = dict.fromkeys(range(1, max_sph_params + 1), "") - - # Process parameter values based on type - if parameter_values is not None: - if isinstance(parameter_values, (str, int, float, bool)): - # Single value goes to param1 - parameters[1] = str(parameter_values) - elif isinstance(parameter_values, list): - # List of values go to sequential params - for i, value in enumerate(parameter_values, 1): - if i <= max_sph_params: # Only use up to max_sph_params parameters - parameters[i] = str(value) - elif isinstance(parameter_values, dict): - # Dict maps param positions to values - for pos_raw, value in parameter_values.items(): - pos = int(pos_raw) if not isinstance(pos_raw, int) else pos_raw - if 1 <= pos <= max_sph_params: # Validate parameter positions - parameters[pos] = str(value) - - # Create a data dictionary with ALL parameters explicitly included - request_data = { - "mix_sn": device_sn, - "type": parameter_id - } - - # Add all SPH parameters to the request - for i in range(1, max_sph_params + 1): - request_data[f"param{i}"] = str(parameters[i]) - - # API: https://www.showdoc.com.cn/262556420217021/6129761750718760 - response = self.session.post( - self._get_url("mixSet"), - data=request_data - ) - - return self._process_response(response.json(), f"writing parameter {parameter_id}") + return Sph(self, device_sn).write_parameter(parameter_id, parameter_values) def sph_write_ac_charge_times(self, device_sn, charge_power, charge_stop_soc, mains_enabled, periods): """ @@ -960,40 +647,7 @@ def sph_write_ac_charge_times(self, device_sn, charge_power, charge_stop_soc, ma requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if not 0 <= charge_power <= 100: # noqa: PLR2004 - raise GrowattParameterError("charge_power must be between 0 and 100") - - if not 0 <= charge_stop_soc <= 100: # noqa: PLR2004 - raise GrowattParameterError("charge_stop_soc must be between 0 and 100") - - if len(periods) != 3: # noqa: PLR2004 - raise GrowattParameterError("periods must contain exactly 3 period definitions") - - # Build request data - request_data = { - "mix_sn": device_sn, - "type": "mix_ac_charge_time_period", - "param1": str(charge_power), - "param2": str(charge_stop_soc), - "param3": "1" if mains_enabled else "0", - } - - # Add period parameters (param4-18) - for i, period in enumerate(periods): - base = i * 5 + 4 - request_data[f"param{base}"] = str(period["start_time"].hour) - request_data[f"param{base + 1}"] = str(period["start_time"].minute) - request_data[f"param{base + 2}"] = str(period["end_time"].hour) - request_data[f"param{base + 3}"] = str(period["end_time"].minute) - request_data[f"param{base + 4}"] = "1" if period["enabled"] else "0" - - # API: https://www.showdoc.com.cn/262556420217021/6129761750718760 - response = self.session.post( - self._get_url("mixSet"), - data=request_data - ) - - return self._process_response(response.json(), "writing AC charge time periods") + return Sph(self, device_sn).write_ac_charge_times(charge_power, charge_stop_soc, mains_enabled, periods) def sph_write_ac_discharge_times(self, device_sn, discharge_power, discharge_stop_soc, periods): """ @@ -1031,111 +685,9 @@ def sph_write_ac_discharge_times(self, device_sn, discharge_power, discharge_sto requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if not 0 <= discharge_power <= 100: # noqa: PLR2004 - raise GrowattParameterError("discharge_power must be between 0 and 100") - - if not 0 <= discharge_stop_soc <= 100: # noqa: PLR2004 - raise GrowattParameterError("discharge_stop_soc must be between 0 and 100") - - if len(periods) != 3: # noqa: PLR2004 - raise GrowattParameterError("periods must contain exactly 3 period definitions") - - # Build request data - request_data = { - "mix_sn": device_sn, - "type": "mix_ac_discharge_time_period", - "param1": str(discharge_power), - "param2": str(discharge_stop_soc), - } + return Sph(self, device_sn).write_ac_discharge_times(discharge_power, discharge_stop_soc, periods) - # Add period parameters (param3-17) - for i, period in enumerate(periods): - base = i * 5 + 3 - request_data[f"param{base}"] = str(period["start_time"].hour) - request_data[f"param{base + 1}"] = str(period["start_time"].minute) - request_data[f"param{base + 2}"] = str(period["end_time"].hour) - request_data[f"param{base + 3}"] = str(period["end_time"].minute) - request_data[f"param{base + 4}"] = "1" if period["enabled"] else "0" - - # API: https://www.showdoc.com.cn/262556420217021/6129761750718760 - response = self.session.post( - self._get_url("mixSet"), - data=request_data - ) - - return self._process_response(response.json(), "writing AC discharge time periods") - - def _parse_time_periods(self, settings_data, time_type): - """ - Parse time periods from settings data. - - Internal helper method to extract and format time period data from SPH settings. - - Args: - settings_data (dict): Settings data from sph_detail call. - time_type (str): Either "Charge" or "Discharge" to specify which periods to parse. - - Returns: - list: A list of dictionaries, each containing details for one time period: - - period_id (int): The period number (1-3) - - start_time (str): Start time in format "HH:MM" - - end_time (str): End time in format "HH:MM" - - enabled (bool): Whether the period is enabled - - """ - periods = [] - - # Process each time period (1-3 for SPH) - for i in range(1, 4): - # Get raw time values - start_time_raw = settings_data.get(f"forced{time_type}TimeStart{i}", "0:0") - end_time_raw = settings_data.get(f"forced{time_type}TimeStop{i}", "0:0") - enabled_raw = settings_data.get(f"forced{time_type}StopSwitch{i}", 0) - - # Handle 'null' string values - if start_time_raw == "null" or not start_time_raw: - start_time_raw = "0:0" - if end_time_raw == "null" or not end_time_raw: - end_time_raw = "0:0" - - # Format times with leading zeros (HH:MM) - try: - start_parts = start_time_raw.split(":") - start_hour = int(start_parts[0]) - start_min = int(start_parts[1]) - start_time = f"{start_hour:02d}:{start_min:02d}" - except (ValueError, IndexError): - start_time = "00:00" - - try: - end_parts = end_time_raw.split(":") - end_hour = int(end_parts[0]) - end_min = int(end_parts[1]) - end_time = f"{end_hour:02d}:{end_min:02d}" - except (ValueError, IndexError): - end_time = "00:00" - - # Get the enabled status - if enabled_raw == "null" or enabled_raw is None: - enabled = False - else: - try: - enabled = int(enabled_raw) == 1 - except (ValueError, TypeError): - enabled = False - - period = { - "period_id": i, - "start_time": start_time, - "end_time": end_time, - "enabled": enabled - } - - periods.append(period) - - return periods - - def sph_read_ac_charge_times(self, device_sn=None, settings_data=None): + def sph_read_ac_charge_times(self, device_sn, settings_data=None): """ Read AC charge time periods and settings from an SPH inverter. @@ -1147,10 +699,8 @@ def sph_read_ac_charge_times(self, device_sn=None, settings_data=None): with the data returned from sph_detail(). Args: - device_sn (str, optional): The device serial number of the inverter. - Required if settings_data is not provided. + device_sn (str): The device serial number of the inverter. settings_data (dict, optional): Settings data from sph_detail call to avoid repeated API calls. - If provided, device_sn is not required. Returns: dict: A dictionary containing: @@ -1169,9 +719,9 @@ def sph_read_ac_charge_times(self, device_sn=None, settings_data=None): print(f"Charge power: {charge_config['charge_power']}%") print(f"Periods: {charge_config['periods']}") - # Option 2: Reuse existing settings data (no device_sn needed) + # Option 2: Reuse existing settings data settings_response = api.sph_detail("DEVICE_SERIAL_NUMBER") - charge_config = api.sph_read_ac_charge_times(settings_data=settings_response) + charge_config = api.sph_read_ac_charge_times(device_sn="DEVICE_SERIAL_NUMBER", settings_data=settings_response) Raises: GrowattParameterError: If neither device_sn nor settings_data is provided. @@ -1179,34 +729,9 @@ def sph_read_ac_charge_times(self, device_sn=None, settings_data=None): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if settings_data is None: - if device_sn is None: - raise GrowattParameterError("Either device_sn or settings_data must be provided") - settings_data = self.sph_detail(device_sn=device_sn) - - # Extract global charge settings - charge_power = settings_data.get("chargePowerCommand", 0) - charge_stop_soc = settings_data.get("wchargeSOCLowLimit", 100) - mains_enabled_raw = settings_data.get("acChargeEnable", 0) - - # Handle null/empty values - if charge_power == "null" or charge_power is None or charge_power == "": - charge_power = 0 - if charge_stop_soc == "null" or charge_stop_soc is None or charge_stop_soc == "": - charge_stop_soc = 100 - if mains_enabled_raw == "null" or mains_enabled_raw is None or mains_enabled_raw == "": - mains_enabled = False - else: - mains_enabled = int(mains_enabled_raw) == 1 - - return { - "charge_power": int(charge_power), - "charge_stop_soc": int(charge_stop_soc), - "mains_enabled": mains_enabled, - "periods": self._parse_time_periods(settings_data, "Charge") - } + return Sph(self, device_sn).read_ac_charge_times(settings_data) - def sph_read_ac_discharge_times(self, device_sn=None, settings_data=None): + def sph_read_ac_discharge_times(self, device_sn, settings_data=None): """ Read AC discharge time periods and settings from an SPH inverter. @@ -1219,9 +744,7 @@ def sph_read_ac_discharge_times(self, device_sn=None, settings_data=None): Args: device_sn (str, optional): The device serial number of the inverter. - Required if settings_data is not provided. settings_data (dict, optional): Settings data from sph_detail call to avoid repeated API calls. - If provided, device_sn is not required. Returns: dict: A dictionary containing: @@ -1239,9 +762,9 @@ def sph_read_ac_discharge_times(self, device_sn=None, settings_data=None): print(f"Discharge power: {discharge_config['discharge_power']}%") print(f"Stop SOC: {discharge_config['discharge_stop_soc']}%") - # Option 2: Reuse existing settings data (no device_sn needed) + # Option 2: Reuse existing settings data settings_response = api.sph_detail("DEVICE_SERIAL_NUMBER") - discharge_config = api.sph_read_ac_discharge_times(settings_data=settings_response) + discharge_config = api.sph_read_ac_discharge_times(device_sn="DEVICE_SERIAL_NUMBER", settings_data=settings_response) Raises: GrowattParameterError: If neither device_sn nor settings_data is provided. @@ -1249,23 +772,4 @@ def sph_read_ac_discharge_times(self, device_sn=None, settings_data=None): requests.exceptions.RequestException: If there is an issue with the HTTP request. """ - if settings_data is None: - if device_sn is None: - raise GrowattParameterError("Either device_sn or settings_data must be provided") - settings_data = self.sph_detail(device_sn=device_sn) - - # Extract global discharge settings - discharge_power = settings_data.get("disChargePowerCommand", 0) - discharge_stop_soc = settings_data.get("wdisChargeSOCLowLimit", 10) - - # Handle null/empty values - if discharge_power == "null" or discharge_power is None or discharge_power == "": - discharge_power = 0 - if discharge_stop_soc == "null" or discharge_stop_soc is None or discharge_stop_soc == "": - discharge_stop_soc = 10 - - return { - "discharge_power": int(discharge_power), - "discharge_stop_soc": int(discharge_stop_soc), - "periods": self._parse_time_periods(settings_data, "Discharge") - } + return Sph(self, device_sn).read_ac_discharge_times(settings_data) diff --git a/growattServer/open_api_v1/devices/__init__.py b/growattServer/open_api_v1/devices/__init__.py new file mode 100644 index 0000000..d2c0fe0 --- /dev/null +++ b/growattServer/open_api_v1/devices/__init__.py @@ -0,0 +1,4 @@ +# noqa: D104 +from .abstract_device import AbstractDevice # noqa: F401 +from .min import Min # noqa: F401 +from .sph import Sph # noqa: F401 diff --git a/growattServer/open_api_v1/devices/abstract_device.py b/growattServer/open_api_v1/devices/abstract_device.py new file mode 100644 index 0000000..9e02c9a --- /dev/null +++ b/growattServer/open_api_v1/devices/abstract_device.py @@ -0,0 +1,51 @@ +"""Abstract device file for centralising shared device logic.""" +from typing import TYPE_CHECKING, TypedDict + +from growattServer.exceptions import GrowattParameterError + +if TYPE_CHECKING: + from growattServer.open_api_v1 import OpenApiV1 + + +class ReadParamResponse(TypedDict): + """Response type for ReadParam endpoints.""" + + data: str + error_code: str + error_msg: str + +class AbstractDevice: + """Abstract device type. Must not be used directly.""" + + def __init__(self, api: "OpenApiV1", device_sn: str) -> None: + """ + Initialize the device with the bare minimum being the device_sn. + + Args: + api (OpenApiV1): API used for all API calls. + device_sn (str): Device serial number used for all API calls. + + """ + self.api = api + self.device_sn = device_sn + + def validate_read_parameter_input(self, parameter_id: str | None, start_address: int | None, end_address: int | None): # noqa: ARG002 + """ + Validate read parameter input and throws an error if it is invalid. + + Args: + parameter_id (str): Parameter ID to read. Don't use start_address and end_address if this is set. + start_address (int, optional): Register start address (for set_any_reg). Don't use parameter_id if this is set. + end_address (int, optional): Register end address (for set_any_reg). Don't use parameter_id if this is set. + + Raises: + GrowattParameterError: If parameters are invalid. + + """ + if parameter_id is None and start_address is None: + raise GrowattParameterError( + "specify either parameter_id or start_address/end_address") + if parameter_id is not None and start_address is not None: + raise GrowattParameterError( + "specify either parameter_id or start_address/end_address - not both." + ) diff --git a/growattServer/open_api_v1/devices/min.py b/growattServer/open_api_v1/devices/min.py new file mode 100644 index 0000000..ca1737b --- /dev/null +++ b/growattServer/open_api_v1/devices/min.py @@ -0,0 +1,411 @@ +"""Min/TLX device file.""" +from datetime import UTC, datetime, timedelta +from typing import Any + +from growattServer.exceptions import GrowattParameterError + +from .abstract_device import AbstractDevice + + +class Min(AbstractDevice): + """Min/TLX device type.""" + + DEVICE_TYPE_ID = 7 + + def detail(self) -> dict: + """ + Get detailed data for a MIN inverter. + + See the API doc: https://www.showdoc.com.cn/262556420217021/6129816412127075. + + Args: + device_sn (str): The serial number of the MIN inverter. + + Returns: + dict: A dictionary containing the MIN inverter details. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + response = self.api.session.get( + self.api.get_url("device/tlx/tlx_data_info"), + params={ + "device_sn": self.device_sn + } + ) + + return self.api.process_response(response.json(), "getting MIN inverter details") + + def energy(self) -> dict: + """ + Get energy data for a MIN inverter. + + Args: + device_sn (str): The serial number of the MIN inverter. + + Returns: + dict: A dictionary containing the MIN inverter energy data. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + response = self.api.session.post( + url=self.api.get_url("device/tlx/tlx_last_data"), + data={ + "tlx_sn": self.device_sn, + }, + ) + + return self.api.process_response(response.json(), "getting MIN inverter energy data") + + def energy_history(self, start_date=None, end_date=None, timezone=None, page=None, limit=None) -> dict: + """ + Get MIN inverter data history. + + Args: + device_sn (str): The ID of the MIN inverter. + start_date (date, optional): Start date. Defaults to today. + end_date (date, optional): End date. Defaults to today. + timezone (str, optional): Timezone ID. + page (int, optional): Page number. + limit (int, optional): Results per page. + + Returns: + dict: A dictionary containing the MIN inverter history data. + + Raises: + GrowattParameterError: If date interval is invalid (exceeds 7 days). + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + if start_date is None and end_date is None: + start_date = datetime.now(UTC).date() + end_date = datetime.now(UTC).date() + elif start_date is None: + start_date = end_date + elif end_date is None: + end_date = start_date + + # check interval validity + if end_date - start_date > timedelta(days=7): + raise GrowattParameterError("date interval must not exceed 7 days") + + response = self.api.session.post( + url=self.api.get_url("device/tlx/tlx_data"), + data={ + "tlx_sn": self.device_sn, + "start_date": start_date.strftime("%Y-%m-%d"), + "end_date": end_date.strftime("%Y-%m-%d"), + "timezone_id": timezone, + "page": page, + "perpage": limit, + } + ) + + return self.api.process_response(response.json(), "getting MIN inverter energy history") + + def settings(self) -> dict: + """ + Get settings for a MIN inverter. + + Args: + device_sn (str): The serial number of the MIN inverter. + + Returns: + dict: A dictionary containing the MIN inverter settings. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + response = self.api.session.get( + self.api.get_url("device/tlx/tlx_set_info"), + params={ + "device_sn": self.device_sn + } + ) + + return self.api.process_response(response.json(), "getting MIN inverter settings") + + def read_parameter(self, parameter_id, start_address=None, end_address=None) -> dict: + """ + Read setting from MIN inverter. + + Args: + device_sn (str): The ID of the TLX inverter. + parameter_id (str): Parameter ID to read. Don't use start_address and end_address if this is set. + start_address (int, optional): Register start address (for set_any_reg). Don't use parameter_id if this is set. + end_address (int, optional): Register end address (for set_any_reg). Don't use parameter_id if this is set. + + Returns: + dict: A dictionary containing the setting value. + + Raises: + GrowattParameterError: If parameters are invalid. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + self.validate_read_parameter_input(parameter_id, start_address, end_address) + + if parameter_id is not None: + # named parameter + start_address = 0 + end_address = 0 + else: + # using register-number mode + parameter_id = "set_any_reg" + if start_address is None: + start_address = end_address + if end_address is None: + end_address = start_address + + response = self.api.session.post( + self.api.get_url("readMinParam"), + data={ + "device_sn": self.device_sn, + "paramId": parameter_id, + "startAddr": start_address, + "endAddr": end_address, + } + ) + + return self.api.process_response(response.json(), f"reading parameter {parameter_id}") + + def write_parameter(self, parameter_id, parameter_values=None) -> dict: + """ + Set parameters on a MIN inverter. + + Args: + device_sn (str): Serial number of the inverter + parameter_id (str): Setting type to be configured + parameter_values: Parameter values to be sent to the system. + Can be a single string (for param1 only), + a list of strings (for sequential params), + or a dictionary mapping param positions to values + + Returns: + dict: JSON response from the server + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + # Initialize all parameters as empty strings + max_min_params = 19 + parameters = dict.fromkeys(range(1, max_min_params + 1), "") + + # Process parameter values based on type + if parameter_values is not None: + if isinstance(parameter_values, (str, int, float, bool)): + # Single value goes to param1 + parameters[1] = str(parameter_values) + elif isinstance(parameter_values, list): + # List of values go to sequential params + for i, value in enumerate(parameter_values, 1): + if i <= max_min_params: # Only use up to max_min_params parameters + parameters[i] = str(value) + elif isinstance(parameter_values, dict): + # Dict maps param positions to values + for pos_raw, value in parameter_values.items(): + pos = int(pos_raw) if not isinstance(pos_raw, int) else pos_raw + if 1 <= pos <= max_min_params: # Validate parameter positions + parameters[pos] = str(value) + + # IMPORTANT: Create a data dictionary with ALL parameters explicitly included + request_data = { + "tlx_sn": self.device_sn, + "type": parameter_id + } + + # Add all MIN parameters to the request + for i in range(1, max_min_params + 1): + request_data[f"param{i}"] = str(parameters[i]) + + # Send the request + response = self.api.session.post( + self.api.get_url("tlxSet"), + data=request_data + ) + + return self.api.process_response(response.json(), f"writing parameter {parameter_id}") + + def write_time_segment(self, segment_id, batt_mode, start_time, end_time, enabled=True) -> dict: + """ + Set a time segment for a MIN inverter. + + Args: + device_sn (str): The serial number of the inverter. + segment_id (int): Time segment ID (1-9). + batt_mode (int): 0=load priority, 1=battery priority, 2=grid priority. + start_time (datetime.time): Start time for the segment. + end_time (datetime.time): End time for the segment. + enabled (bool): Whether this segment is enabled. + + Returns: + dict: The server response. + + Raises: + GrowattParameterError: If parameters are invalid. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + max_min_params = 19 + max_min_segments = 9 + max_batt_mode = 2 + + if not 1 <= segment_id <= max_min_segments: + msg = f"segment_id must be between 1 and {max_min_segments}" + raise GrowattParameterError(msg) + + if not 0 <= batt_mode <= max_batt_mode: + msg = f"batt_mode must be between 0 and {max_batt_mode}" + raise GrowattParameterError(msg) + + # Initialize ALL 19 parameters as empty strings, not just the ones we need + all_params = { + "tlx_sn": self.device_sn, + "type": f"time_segment{segment_id}" + } + + # Add param1 through param19, setting the values we need + all_params["param1"] = str(batt_mode) + all_params["param2"] = str(start_time.hour) + all_params["param3"] = str(start_time.minute) + all_params["param4"] = str(end_time.hour) + all_params["param5"] = str(end_time.minute) + all_params["param6"] = "1" if enabled else "0" + + # Add empty strings for all unused parameters + for i in range(7, max_min_params + 1): + all_params[f"param{i}"] = "" + + # Send the request + response = self.api.session.post( + self.api.get_url("tlxSet"), + data=all_params + ) + + return self.api.process_response(response.json(), f"writing time segment {segment_id}") + + def read_time_segments(self, settings_data=None) -> list[dict[str, Any]]: + """ + Read Time-of-Use (TOU) settings from a Growatt MIN/TLX inverter. + + Retrieves all 9 time segments from a Growatt MIN/TLX inverter and + parses them into a structured format. + + Note that this function uses min_settings() internally to get the settings data, + To avoid endpoint rate limit, you can pass the settings_data parameter + with the data returned from min_settings(). + + Args: + device_sn (str): The device serial number of the inverter + settings_data (dict, optional): Settings data from min_settings call to avoid repeated API calls. + Can be either the complete response or just the data portion. + + Returns: + list: A list of dictionaries, each containing details for one time segment: + - segment_id (int): The segment number (1-9) + - batt_mode (int): 0=Load First, 1=Battery First, 2=Grid First + - mode_name (str): String representation of the mode + - start_time (str): Start time in format "HH:MM" + - end_time (str): End time in format "HH:MM" + - enabled (bool): Whether the segment is enabled + + Example: + # Option 1: Make a single call + tou_settings = api.min_read_time_segments("DEVICE_SERIAL_NUMBER") + + # Option 2: Reuse existing settings data + settings_response = api.min_settings("DEVICE_SERIAL_NUMBER") + tou_settings = api.min_read_time_segments("DEVICE_SERIAL_NUMBER", settings_response) + + Raises: + GrowattV1ApiError: If the API request fails + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + # Process the settings data + if settings_data is None: + # Fetch settings if not provided + settings_data = self.settings() + + # Define mode names + mode_names = { + 0: "Load First", + 1: "Battery First", + 2: "Grid First" + } + + segments = [] + + # Process each time segment + for i in range(1, 10): # Segments 1-9 + # Get raw time values + start_time_raw = settings_data.get(f"forcedTimeStart{i}", "0:0") + end_time_raw = settings_data.get(f"forcedTimeStop{i}", "0:0") + + # Handle 'null' string values + if start_time_raw == "null" or not start_time_raw: + start_time_raw = "0:0" + if end_time_raw == "null" or not end_time_raw: + end_time_raw = "0:0" + + # Format times with leading zeros (HH:MM) + try: + start_parts = start_time_raw.split(":") + start_hour = int(start_parts[0]) + start_min = int(start_parts[1]) + start_time = f"{start_hour:02d}:{start_min:02d}" + except (ValueError, IndexError): + start_time = "00:00" + + try: + end_parts = end_time_raw.split(":") + end_hour = int(end_parts[0]) + end_min = int(end_parts[1]) + end_time = f"{end_hour:02d}:{end_min:02d}" + except (ValueError, IndexError): + end_time = "00:00" + + # Get the mode value safely + mode_raw = settings_data.get(f"time{i}Mode") + if mode_raw == "null" or mode_raw is None: + batt_mode = None + else: + try: + batt_mode = int(mode_raw) + except (ValueError, TypeError): + batt_mode = None + + # Get the enabled status safely + enabled_raw = settings_data.get(f"forcedStopSwitch{i}", 0) + if enabled_raw == "null" or enabled_raw is None: + enabled = False + else: + try: + enabled = int(enabled_raw) == 1 + except (ValueError, TypeError): + enabled = False + + segment = { + "segment_id": i, + "batt_mode": batt_mode, + "mode_name": mode_names.get(batt_mode, "Unknown"), + "start_time": start_time, + "end_time": end_time, + "enabled": enabled + } + + segments.append(segment) + + return segments diff --git a/growattServer/open_api_v1/devices/sph.py b/growattServer/open_api_v1/devices/sph.py new file mode 100644 index 0000000..5bdf24e --- /dev/null +++ b/growattServer/open_api_v1/devices/sph.py @@ -0,0 +1,556 @@ +"""SPH/MIX device file.""" +from datetime import UTC, datetime, timedelta + +from growattServer.exceptions import GrowattParameterError + +from .abstract_device import AbstractDevice + + +class Sph(AbstractDevice): + """SPH/MIX device type.""" + + DEVICE_TYPE_ID = 5 + + def detail(self): + """ + Get detailed data for an SPH inverter. + + Args: + device_sn (str): The serial number of the SPH inverter. + + Returns: + dict: A dictionary containing the SPH inverter details. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + # API: https://www.showdoc.com.cn/262556420217021/6129763571291058 + response = self.api.session.get( + self.api.get_url("device/mix/mix_data_info"), + params={ + "device_sn": self.device_sn + } + ) + + return self.api.process_response(response.json(), "getting SPH inverter details") + + def energy(self): + """ + Get energy data for an SPH inverter. + + Args: + device_sn (str): The serial number of the SPH inverter. + + Returns: + dict: A dictionary containing the SPH inverter energy data. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + # API: https://www.showdoc.com.cn/262556420217021/6129764475556048 + response = self.api.session.post( + url=self.api.get_url("device/mix/mix_last_data"), + data={ + "mix_sn": self.device_sn, + }, + ) + + return self.api.process_response(response.json(), "getting SPH inverter energy data") + + def energy_history(self, start_date=None, end_date=None, timezone=None, page=None, limit=None): + """ + Get SPH inverter data history. + + Args: + device_sn (str): The ID of the SPH inverter. + start_date (date, optional): Start date. Defaults to today. + end_date (date, optional): End date. Defaults to today. + timezone (str, optional): Timezone ID. + page (int, optional): Page number. + limit (int, optional): Results per page. + + Returns: + dict: A dictionary containing the SPH inverter history data. + + Raises: + GrowattParameterError: If date interval is invalid (exceeds 7 days). + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + if start_date is None and end_date is None: + start_date = datetime.now(UTC).date() + end_date = datetime.now(UTC).date() + elif start_date is None: + start_date = end_date + elif end_date is None: + end_date = start_date + + # check interval validity + if end_date - start_date > timedelta(days=7): + raise GrowattParameterError("date interval must not exceed 7 days") + + # API: https://www.showdoc.com.cn/262556420217021/6129765461123058 + response = self.api.session.post( + url=self.api.get_url("device/mix/mix_data"), + data={ + "mix_sn": self.device_sn, + "start_date": start_date.strftime("%Y-%m-%d"), + "end_date": end_date.strftime("%Y-%m-%d"), + "timezone_id": timezone, + "page": page, + "perpage": limit, + } + ) + + return self.api.process_response(response.json(), "getting SPH inverter energy history") + + def read_parameter(self, parameter_id=None, start_address=None, end_address=None): + """ + Read setting from SPH inverter. + + Args: + device_sn (str): The ID of the SPH inverter. + parameter_id (str, optional): Parameter ID to read. Don't use start_address and end_address if this is set. + start_address (int, optional): Register start address (for set_any_reg). Don't use parameter_id if this is set. + end_address (int, optional): Register end address (for set_any_reg). Don't use parameter_id if this is set. + + Returns: + dict: A dictionary containing the setting value. + + Raises: + GrowattParameterError: If parameters are invalid. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + if parameter_id is None and start_address is None: + raise GrowattParameterError( + "specify either parameter_id or start_address/end_address") + if parameter_id is not None and start_address is not None: + raise GrowattParameterError( + "specify either parameter_id or start_address/end_address - not both." + ) + if parameter_id is not None: + # named parameter + start_address = 0 + end_address = 0 + else: + # address range + parameter_id = "set_any_reg" + + # API: https://www.showdoc.com.cn/262556420217021/6129766954561259 + response = self.api.session.post( + self.api.get_url("readMixParam"), + data={ + "device_sn": self.device_sn, + "paramId": parameter_id, + "startAddr": start_address, + "endAddr": end_address + } + ) + + return self.api.process_response(response.json(), f"reading parameter {parameter_id}") + + def write_parameter(self, parameter_id, parameter_values=None): + """ + Set parameters on an SPH inverter. + + Args: + device_sn (str): Serial number of the inverter + parameter_id (str): Setting type to be configured + parameter_values: Parameter values to be sent to the system. + Can be a single string (for param1 only), + a list of strings (for sequential params), + or a dictionary mapping param positions to values + + Returns: + dict: JSON response from the server + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + # Initialize all parameters as empty strings (API uses param1-param18) + max_sph_params = 18 + parameters = dict.fromkeys(range(1, max_sph_params + 1), "") + + # Process parameter values based on type + if parameter_values is not None: + if isinstance(parameter_values, (str, int, float, bool)): + # Single value goes to param1 + parameters[1] = str(parameter_values) + elif isinstance(parameter_values, list): + # List of values go to sequential params + for i, value in enumerate(parameter_values, 1): + if i <= max_sph_params: # Only use up to max_sph_params parameters + parameters[i] = str(value) + elif isinstance(parameter_values, dict): + # Dict maps param positions to values + for pos_raw, value in parameter_values.items(): + pos = int(pos_raw) if not isinstance(pos_raw, int) else pos_raw + if 1 <= pos <= max_sph_params: # Validate parameter positions + parameters[pos] = str(value) + + # Create a data dictionary with ALL parameters explicitly included + request_data = { + "mix_sn": self.device_sn, + "type": parameter_id + } + + # Add all SPH parameters to the request + for i in range(1, max_sph_params + 1): + request_data[f"param{i}"] = str(parameters[i]) + + # API: https://www.showdoc.com.cn/262556420217021/6129761750718760 + response = self.api.session.post( + self.api.get_url("mixSet"), + data=request_data + ) + + return self.api.process_response(response.json(), f"writing parameter {parameter_id}") + + def write_ac_charge_times(self, charge_power, charge_stop_soc, mains_enabled, periods): + """ + Set AC charge time periods for an SPH inverter. + + Args: + device_sn (str): The serial number of the inverter. + charge_power (int): Charging power percentage (0-100). + charge_stop_soc (int): Stop charging at this SOC percentage (0-100). + mains_enabled (bool): Enable grid charging. + periods (list): List of 3 period dicts, each with keys: + - start_time (datetime.time): Start time for the period + - end_time (datetime.time): End time for the period + - enabled (bool): Whether this period is enabled + + Returns: + dict: The server response. + + Example: + from datetime import time + + api.sph_write_ac_charge_times( + device_sn="ABC123", + charge_power=100, + charge_stop_soc=100, + mains_enabled=True, + periods=[ + {"start_time": time(1, 0), "end_time": time(5, 0), "enabled": True}, + {"start_time": time(0, 0), "end_time": time(0, 0), "enabled": False}, + {"start_time": time(0, 0), "end_time": time(0, 0), "enabled": False}, + ] + ) + + Raises: + GrowattParameterError: If parameters are invalid. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + if not 0 <= charge_power <= 100: # noqa: PLR2004 + raise GrowattParameterError("charge_power must be between 0 and 100") + + if not 0 <= charge_stop_soc <= 100: # noqa: PLR2004 + raise GrowattParameterError("charge_stop_soc must be between 0 and 100") + + if len(periods) != 3: # noqa: PLR2004 + raise GrowattParameterError("periods must contain exactly 3 period definitions") + + # Build request data + request_data = { + "mix_sn": self.device_sn, + "type": "mix_ac_charge_time_period", + "param1": str(charge_power), + "param2": str(charge_stop_soc), + "param3": "1" if mains_enabled else "0", + } + + # Add period parameters (param4-18) + for i, period in enumerate(periods): + base = i * 5 + 4 + request_data[f"param{base}"] = str(period["start_time"].hour) + request_data[f"param{base + 1}"] = str(period["start_time"].minute) + request_data[f"param{base + 2}"] = str(period["end_time"].hour) + request_data[f"param{base + 3}"] = str(period["end_time"].minute) + request_data[f"param{base + 4}"] = "1" if period["enabled"] else "0" + + # API: https://www.showdoc.com.cn/262556420217021/6129761750718760 + response = self.api.session.post( + self.api.get_url("mixSet"), + data=request_data + ) + + return self.api.process_response(response.json(), "writing AC charge time periods") + + def write_ac_discharge_times(self, discharge_power, discharge_stop_soc, periods): + """ + Set AC discharge time periods for an SPH inverter. + + Args: + device_sn (str): The serial number of the inverter. + discharge_power (int): Discharging power percentage (0-100). + discharge_stop_soc (int): Stop discharging at this SOC percentage (0-100). + periods (list): List of 3 period dicts, each with keys: + - start_time (datetime.time): Start time for the period + - end_time (datetime.time): End time for the period + - enabled (bool): Whether this period is enabled + + Returns: + dict: The server response. + + Example: + from datetime import time + + api.sph_write_ac_discharge_times( + device_sn="ABC123", + discharge_power=100, + discharge_stop_soc=10, + periods=[ + {"start_time": time(17, 0), "end_time": time(21, 0), "enabled": True}, + {"start_time": time(0, 0), "end_time": time(0, 0), "enabled": False}, + {"start_time": time(0, 0), "end_time": time(0, 0), "enabled": False}, + ] + ) + + Raises: + GrowattParameterError: If parameters are invalid. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + if not 0 <= discharge_power <= 100: # noqa: PLR2004 + raise GrowattParameterError("discharge_power must be between 0 and 100") + + if not 0 <= discharge_stop_soc <= 100: # noqa: PLR2004 + raise GrowattParameterError("discharge_stop_soc must be between 0 and 100") + + if len(periods) != 3: # noqa: PLR2004 + raise GrowattParameterError("periods must contain exactly 3 period definitions") + + # Build request data + request_data = { + "mix_sn": self.device_sn, + "type": "mix_ac_discharge_time_period", + "param1": str(discharge_power), + "param2": str(discharge_stop_soc), + } + + # Add period parameters (param3-17) + for i, period in enumerate(periods): + base = i * 5 + 3 + request_data[f"param{base}"] = str(period["start_time"].hour) + request_data[f"param{base + 1}"] = str(period["start_time"].minute) + request_data[f"param{base + 2}"] = str(period["end_time"].hour) + request_data[f"param{base + 3}"] = str(period["end_time"].minute) + request_data[f"param{base + 4}"] = "1" if period["enabled"] else "0" + + # API: https://www.showdoc.com.cn/262556420217021/6129761750718760 + response = self.api.session.post( + self.api.get_url("mixSet"), + data=request_data + ) + + return self.api.process_response(response.json(), "writing AC discharge time periods") + + def _parse_time_periods(self, settings_data, time_type): + """ + Parse time periods from settings data. + + Internal helper method to extract and format time period data from SPH settings. + + Args: + settings_data (dict): Settings data from sph_detail call. + time_type (str): Either "Charge" or "Discharge" to specify which periods to parse. + + Returns: + list: A list of dictionaries, each containing details for one time period: + - period_id (int): The period number (1-3) + - start_time (str): Start time in format "HH:MM" + - end_time (str): End time in format "HH:MM" + - enabled (bool): Whether the period is enabled + + """ + periods = [] + + # Process each time period (1-3 for SPH) + for i in range(1, 4): + # Get raw time values + start_time_raw = settings_data.get(f"forced{time_type}TimeStart{i}", "0:0") + end_time_raw = settings_data.get(f"forced{time_type}TimeStop{i}", "0:0") + enabled_raw = settings_data.get(f"forced{time_type}StopSwitch{i}", 0) + + # Handle 'null' string values + if start_time_raw == "null" or not start_time_raw: + start_time_raw = "0:0" + if end_time_raw == "null" or not end_time_raw: + end_time_raw = "0:0" + + # Format times with leading zeros (HH:MM) + try: + start_parts = start_time_raw.split(":") + start_hour = int(start_parts[0]) + start_min = int(start_parts[1]) + start_time = f"{start_hour:02d}:{start_min:02d}" + except (ValueError, IndexError): + start_time = "00:00" + + try: + end_parts = end_time_raw.split(":") + end_hour = int(end_parts[0]) + end_min = int(end_parts[1]) + end_time = f"{end_hour:02d}:{end_min:02d}" + except (ValueError, IndexError): + end_time = "00:00" + + # Get the enabled status + if enabled_raw == "null" or enabled_raw is None: + enabled = False + else: + try: + enabled = int(enabled_raw) == 1 + except (ValueError, TypeError): + enabled = False + + period = { + "period_id": i, + "start_time": start_time, + "end_time": end_time, + "enabled": enabled + } + + periods.append(period) + + return periods + + def read_ac_charge_times(self, settings_data=None): + """ + Read AC charge time periods and settings from an SPH inverter. + + Retrieves all 3 AC charge time periods plus global charge settings + (power, stop SOC, mains enabled) from an SPH inverter. + + Note that this function uses sph_detail() internally to get the settings data. + To avoid endpoint rate limit, you can pass the settings_data parameter + with the data returned from sph_detail(). + + Args: + device_sn (str): The device serial number of the inverter. + settings_data (dict, optional): Settings data from sph_detail call to avoid repeated API calls. + + Returns: + dict: A dictionary containing: + - charge_power (int): Charging power percentage (0-100) + - charge_stop_soc (int): Stop charging at this SOC percentage (0-100) + - mains_enabled (bool): Whether grid/mains charging is enabled + - periods (list): List of 3 period dicts, each with: + - period_id (int): The period number (1-3) + - start_time (str): Start time in format "HH:MM" + - end_time (str): End time in format "HH:MM" + - enabled (bool): Whether the period is enabled + + Example: + # Option 1: Fetch settings automatically + charge_config = api.sph_read_ac_charge_times(device_sn="DEVICE_SERIAL_NUMBER") + print(f"Charge power: {charge_config['charge_power']}%") + print(f"Periods: {charge_config['periods']}") + + # Option 2: Reuse existing settings data + settings_response = api.sph_detail("DEVICE_SERIAL_NUMBER") + charge_config = api.sph_read_ac_charge_times(device_sn="DEVICE_SERIAL_NUMBER", settings_data=settings_response) + + Raises: + GrowattParameterError: If neither device_sn nor settings_data is provided. + GrowattV1ApiError: If the API request fails. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + if settings_data is None: + settings_data = self.detail() + + # Extract global charge settings + charge_power = settings_data.get("chargePowerCommand", 0) + charge_stop_soc = settings_data.get("wchargeSOCLowLimit", 100) + mains_enabled_raw = settings_data.get("acChargeEnable", 0) + + # Handle null/empty values + if charge_power == "null" or charge_power is None or charge_power == "": + charge_power = 0 + if charge_stop_soc == "null" or charge_stop_soc is None or charge_stop_soc == "": + charge_stop_soc = 100 + if mains_enabled_raw == "null" or mains_enabled_raw is None or mains_enabled_raw == "": + mains_enabled = False + else: + mains_enabled = int(mains_enabled_raw) == 1 + + return { + "charge_power": int(charge_power), + "charge_stop_soc": int(charge_stop_soc), + "mains_enabled": mains_enabled, + "periods": self._parse_time_periods(settings_data, "Charge") + } + + def read_ac_discharge_times(self, settings_data=None): + """ + Read AC discharge time periods and settings from an SPH inverter. + + Retrieves all 3 AC discharge time periods plus global discharge settings + (power, stop SOC) from an SPH inverter. + + Note that this function uses sph_detail() internally to get the settings data. + To avoid endpoint rate limit, you can pass the settings_data parameter + with the data returned from sph_detail(). + + Args: + device_sn (str, optional): The device serial number of the inverter. + settings_data (dict, optional): Settings data from sph_detail call to avoid repeated API calls. + + Returns: + dict: A dictionary containing: + - discharge_power (int): Discharging power percentage (0-100) + - discharge_stop_soc (int): Stop discharging at this SOC percentage (0-100) + - periods (list): List of 3 period dicts, each with: + - period_id (int): The period number (1-3) + - start_time (str): Start time in format "HH:MM" + - end_time (str): End time in format "HH:MM" + - enabled (bool): Whether the period is enabled + + Example: + # Option 1: Fetch settings automatically + discharge_config = api.sph_read_ac_discharge_times(device_sn="DEVICE_SERIAL_NUMBER") + print(f"Discharge power: {discharge_config['discharge_power']}%") + print(f"Stop SOC: {discharge_config['discharge_stop_soc']}%") + + # Option 2: Reuse existing settings data + settings_response = api.sph_detail("DEVICE_SERIAL_NUMBER") + discharge_config = api.sph_read_ac_discharge_times(device_sn="DEVICE_SERIAL_NUMBER", settings_data=settings_response) + + Raises: + GrowattParameterError: If neither device_sn nor settings_data is provided. + GrowattV1ApiError: If the API request fails. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + if settings_data is None: + settings_data = self.detail() + + # Extract global discharge settings + discharge_power = settings_data.get("disChargePowerCommand", 0) + discharge_stop_soc = settings_data.get("wdisChargeSOCLowLimit", 10) + + # Handle null/empty values + if discharge_power == "null" or discharge_power is None or discharge_power == "": + discharge_power = 0 + if discharge_stop_soc == "null" or discharge_stop_soc is None or discharge_stop_soc == "": + discharge_stop_soc = 10 + + return { + "discharge_power": int(discharge_power), + "discharge_stop_soc": int(discharge_stop_soc), + "periods": self._parse_time_periods(settings_data, "Discharge") + }