diff --git a/.gitignore b/.gitignore index 21010f5..22d9e8a 100644 --- a/.gitignore +++ b/.gitignore @@ -26,6 +26,7 @@ share/python-wheels/ .installed.cfg *.egg MANIFEST +*.json # Symlink examples/growattServer diff --git a/.ruff.toml b/.ruff.toml new file mode 100644 index 0000000..cb62982 --- /dev/null +++ b/.ruff.toml @@ -0,0 +1,25 @@ +# The contents of this file is based on https://github.com/home-assistant/core/blob/dev/pyproject.toml + +target-version = "py313" + +[lint] +select = [ + "ALL", +] + +ignore = [ + "ANN401", # Dynamically typed expressions (typing.Any) are disallowed + "D203", # no-blank-line-before-class (incompatible with formatter) + "D212", # multi-line-summary-first-line (incompatible with formatter) + "COM812", # incompatible with formatter + "ISC001", # incompatible with formatter +] + +[lint.flake8-pytest-style] +fixture-parentheses = false + +[lint.pyupgrade] +keep-runtime-typing = true + +[lint.mccabe] +max-complexity = 25 \ No newline at end of file diff --git a/docs/openapiv1/min_tlx_settings.md b/docs/openapiv1/min_tlx_settings.md index 26730e2..6042063 100644 --- a/docs/openapiv1/min_tlx_settings.md +++ b/docs/openapiv1/min_tlx_settings.md @@ -32,4 +32,27 @@ For MIN/TLX systems, the public V1 API provides a more robust way to read and wr * function: `api.min_read_time_segments` * parameters: * `device_sn`: The device serial number - * `settings_data`: Optional settings data to avoid redundant API calls \ No newline at end of file + * `settings_data`: Optional settings data to avoid redundant API calls + + | Parameter name | Backflow prevention setting item | Set switch machine | Set time | Set upper limit of mains voltage | Set lower limit of mains voltage | Set off-grid enable | Set off-grid frequency | Set off-grid voltage | Set active power | Set none Power | Set PF value | Charging power | Charge stop SOC | Discharge power | Discharge stop SOC | Mains charging | Time period 1 | Time period 2 | Time period 3 | Time period 4 | Time period 5 | Time period 6 | Time Period 7 | Time Period 8 | Time Period 9 | Parameter Description | +|----------------|--------------------------------------------------|------------------------------------|---------------------|----------------------------------|----------------------------------|--------------------------------|----------------------------|---------------------------|------------------|--------------------|-------------------|------------------------|---------------------------|---------------------------|------------------------------|---------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------| +| Type | backflow_setting | tlx_on_off | pf_sys_year | pv_grid_voltage_high | pv_grid_voltage_low | tlx_off_grid_enable | tlx_ac_discharge_frequency | tlx_ac_discharge_voltage | pv_active_p_rate | pv_reactive_p_rate | pv_power_factor | charge_power | charge_stop_soc | discharge_power | discharge_stop_soc | ac_charge | time_segment1 | time_segment2 | time_segment3 | time_segment4 | time_segment5 | time_segment6 | time_segment7 | time_segment8 | time_segment9 | The parameter description is inside the brackets, and the parameter list or parameter range outside the brackets | +| param1 | 0 (disable), 1 (enable the meter), 2 (enable CT) | 0000 (shut down), 0001 (switch on) | 2020-09-10 10:00:00 | 270 | 180 | 1 (enable Yes), 0 (prohibited) | 0 (50HZ), 1 (60HZ) | 0 (230), 1 (208), 2 (240) | 0~100 | 0~100 | -0.8 ~ -1/0.8 ~ 1 | 0~100 (charging power) | 0~100 (charging stop SOC) | 0~100 (discharging power) | 0~100 (discharging stop SOC) | Mains enable: 1 (enable), 0 (disable) | Mode: 0 (load priority), 1 (battery priority), 2 (grid priority) | Mode: 0 (load priority), 1 (battery priority), 2 (grid priority) | Mode: 0 (load priority), 1 (Battery priority), 2 (grid priority) | mode: 0 (load priority), 1 (battery priority), 2 (grid priority) | mode: 0 (load priority), 1 (battery priority), 2 (grid priority) | Mode: 0 (load priority), 1 (battery priority), 2 (grid priority) | Mode: 0 (load priority), 1 (battery priority), 2 (grid priority) | Mode: 0 (load priority), 1 (Battery priority), 2 (grid priority) | Mode: 0 (load priority), 1 (battery priority), 2 (grid priority) | | +| param2 | | | | | | | | | | | | | | | | | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | | +| param3 | | | | | | | | | | | | | | | | | 00~59 (minutes) | 00~59 (minutes) | 00~59 (minutes) | 00~59 (minutes) | 00~59 (Minutes) | 00~59 (Minutes) | 00~59 (Minutes) | 00~59 (Minutes) | 00~59 (Minutes) | | +| param4 | | | | | | | | | | | | | | | | | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | 00~23 (Hour) | | +| param5 | | | | | | | | | | | | | | | | | 00~59 (minutes) | 00~59 (minutes) | 00~59 (minutes) | 00~59 (minutes) | 00~59 (Minutes) | 00~59 (Minutes) | 00~59 (Minutes) | 00~59 (Minutes) | 00~59 (Minutes) | | +| param6 | | | | | | | | | | | | | | | | | 1 (enable), 0 (disable) | 1 (enable), 0 (disable) | 1 (enable), 0 (disable) ) | 1 (enable), 0 (disable) | 1 (enable), 0 (disable) | 1 (enable), 0 (disable) | 1 (enable), 0 (disable) | 1 (enable) ), 0 (disable) | 1 (enable), 0 (disable) | | +| param7 | | | | | | | | | | | | | | +| param8 | | | | | | | | | | | | | | +| param9 | | | | | | | | | | | | | | +| param10 | | | | | | | | | | | | | | +| param11 | | | | | | | | | | | | | | +| param12 | | | | | | | | | | | | | | +| param13 | | | | | | | | | | | | | | +| param14 | | | | | | | | | | | | | | +| param15 | | | | | | | | | | | | | | +| param16 | | | | | | | | | | | | | | +| param17 | | | | | | | | | | | | | +| param18 | | | | | | | | | | | | | | | +| param19 | | | | | | | | | | | | | diff --git a/docs/openapiv1/mix_sph_settings.md b/docs/openapiv1/mix_sph_settings.md new file mode 100644 index 0000000..330a79d --- /dev/null +++ b/docs/openapiv1/mix_sph_settings.md @@ -0,0 +1,124 @@ +# MIX/SPH Inverter Settings + +This is part of the [OpenAPI V1 doc](../openapiv1.md). + +For MIX/SPH systems, the public V1 API provides a way to read and write inverter settings: + +* **Device Settings** + * function: `api.device_settings` + * parameters: + * `device_sn`: The device serial number + * `device_type`: Use `DeviceType.SPH_MIX` for MIX/SPH inverters + +* **Read Parameter** + * function: `api.read_parameter` + * parameters: + * `device_sn`: The device serial number + * `device_type`: Use `DeviceType.SPH_MIX` for MIX/SPH inverters + * `parameter_id`: Parameter ID to read (e.g., "pv_on_off", "mix_ac_discharge_time_period") + + +List: + mix_ac_discharge_time_period + mix_ac_charge_time_period + backflow_setting + pv_on_off + pf_sys_year + pv_grid_voltage_high + pv_grid_voltage_low + mix_off_grid_enable + mix_ac_discharge_frequency + mix_ac_discharge_voltage + v_reactive_p_rate + pv_power_factor + mix_load_flast_value_multi + mix_load_first_control + mix_single_export + + +* **Time Segments** + * function: `api.read_time_segments` + * parameters: + * `device_sn`: The device serial number + * `device_type`: Use `DeviceType.SPH_MIX` for MIX/SPH inverters + * `settings_data`: Optional settings data to avoid redundant API calls + + Returns: + ```python + [ + { + 'segment_id': int, # 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 + }, + # ... (up to 9 segments) + ] + ``` + +* **Write Time Segment** + * function: `api.write_time_segment` + * parameters: + * `device_sn`: The device serial number + * `device_type`: Use `DeviceType.SPH_MIX` for MIX/SPH inverters + * `settings_data`: Optional settings data to avoid redundant API calls + + Returns: + ```python + [] + ``` + +| Parameter name | Grid priority setting item | Battery priority setting item | Anti-backflow setting item | Set power on and off | Set time | Set the upper limit of mains voltage | Set the lower limit of mains voltage | Set off-grid enable | Set off-grid frequency | Setting Off-grid voltage | Set whether to store the following PF commands | Set active power | Set reactive power | Set PF value | Discharge stop SOC | LoadFirst three-phase independent output | Single-phase anti-reverse flow | Parameter description | +|----------------|------------------------------|---------------------------------|--------------------------------------------|-----------------------|-------------------------------------------------|--------------------------------------|--------------------------------------|---------------------------|----------------------------|---------------------------|------------------------------------------------|------------------|----------------------------|------------------------|--------------------|---------------------------------------------------------------------------------------------------------------------|--------------------------------|-----------------------| +| type | mix_ac_discharge_time_period | mix_ac_charge_time_period | backflow_setting | pv_on_off | pf_sys_year | pv_grid_voltage_high | pv_grid_voltage_low | mix_off_grid_enable | mix_ac_discharge_frequency | mix_ac_discharge_voltage | v_reactive_p_rate | pv_power_factor | mix_load_flast_value_multi | mix_load_first_control | mix_single_export | The parameter description is inside the brackets, and the parameter list is outside the brackets or parameter range | +| param1 | 0~100 (discharge power) | 0~100 (charging power) | 1 (on), 0 (off) | 0000 (off), 0001 (on) | 00 (hour): 00 (minute) ~ 00 (hour): 00 (minute) | 270 | 180 | 1 (enabled), 0 (disabled) | 0 (50HZ), 1 (60HZ) | 0 (230), 1 (208), 2 (240) | 1 (on), 0 (off) | 0~100 | 0~100 | -0.8 ~ -1/0.8 ~ 1 | 0~100 | 0 (three-phase sum enable), 1 (single phase enable) | 0( off),1(on) | | +| param2 | 0~100 (discharge stop SOC) | 0~100 (charge stop SOC) | 0~100 (anti-reverse flow power percentage) | | | | | | | | | | | | | | | | +| param3 | 00~23 (hour) | 1 (enable), 0 (disable) (mains) | | | | | | | | | | | | | | | | +| param4 | 00~59 (minutes) | 00~23 (hours) | | | | | | | | | | | | | | | | | +| param5 | 00~23 (hours) | 00~59 (minutes) | | | | | | | | | | | | | | | | | +| param6 | 00~59 (minutes) | 00~23 (hours) | | | | | | | | | | | | | | | | | +| param7 | 1 (enable), 0 (disable) | 00~59 (minutes) | | | | | | | | | | | | | | | | +| param8 | 00~23 (hour) | 1 (enable), 0 (disable) | | | | | | | | | | | | | | | | +| param9 | 00~59 (minutes) | 00~23 (hours) | | | | | | | | | | | | | | | | | +| param10 | 00~23 (hours) | 00~59 (minutes) | | | | | | | | | | | | | | | | | +| param11 | 00~59 (minutes) | 00~23 (hours) | | | | | | | | | | | | | | | | | +| param12 | 1 (enable), 0 (disable) | 00~59 (minutes) | | | | | | | | | | | | | | | | +| param13 | 00~23 (hour) | 1 (enabled), 0 (disabled) | | | | | | | | | | | | | | | | +| param14 | 00~59 (minutes) | 00~23 (hours) | | | | | | | | | | | | | | | | | +| param15 | 00~23 (hours) | 00~59 (minutes) | | | | | | | | | | | | | | | | | +| param16 | 00~59 (minutes) | 00~23 (hours) | | | | | | | | | | | | | | | | | +| param17 | 1 (enable), 0 (disable) | 00~59 (minutes) | | | | | | | | | | | | | | | | +| param18 | | 1 (enabled), 0 (disabled) | | | | | | | | | | | | | | | | | + + + + + + + +## Common Usage Examples + +### Reading Device Settings +```python +from growattServer import OpenApiV1, DeviceType + +api = OpenApiV1(token="your_api_token") +settings = api.device_settings("DEVICE_SN", DeviceType.SPH_MIX) +``` + +### Reading Parameters +```python +# Read a specific parameter +value = api.read_parameter("DEVICE_SN", DeviceType.SPH_MIX, "ac_charge") +``` + +### Reading Time Segments +```python +# Option 1: Single call +segments = api.read_time_segments("DEVICE_SN", DeviceType.SPH_MIX) + +# Option 2: Reuse settings data to avoid multiple API calls +settings = api.device_settings("DEVICE_SN", DeviceType.SPH_MIX) +segments = api.read_time_segments("DEVICE_SN", DeviceType.SPH_MIX, settings) \ No newline at end of file diff --git a/examples/min_example.py b/examples/min_example.py index 3594fd7..80f6ba1 100644 --- a/examples/min_example.py +++ b/examples/min_example.py @@ -1,7 +1,8 @@ -import growattServer +from . import growattServer import datetime import json import requests +import os """ # Example script controlling a MID/TLX Growatt (MID-30KTL3-XH + APX battery) system using the public growatt API @@ -27,60 +28,80 @@ devices = api.device_list(plant_id) for device in devices['devices']: - if device['type'] == 7: # (MIN/TLX) + print(device) + if device['device_type'] == growattServer.DeviceType.MIN_TLX.value: inverter_sn = device['device_sn'] - print(f"Processing inverter: {inverter_sn}") + device_type = device['device_type'] + print(f"Processing {device_type.name} inverter: {inverter_sn}") # Get device details - inverter_data = api.min_detail(inverter_sn) + inverter_data = api.min_detail( + device_sn=inverter_sn, + ) print("Saving inverter data to inverter_data.json") with open('inverter_data.json', 'w') as f: json.dump(inverter_data, f, indent=4, sort_keys=True) # Get energy data - energy_data = api.min_energy(device_sn=inverter_sn) + energy_data = api.min_energy( + device_sn=inverter_sn, + ) print("Saving energy data to energy_data.json") with open('energy_data.json', 'w') as f: json.dump(energy_data, f, indent=4, sort_keys=True) # Get energy history - energy_history_data = api.min_energy_history(inverter_sn) + energy_history_data = api.min_energy_history( + device_sn=inverter_sn, + ) print("Saving energy history data to energy_history.json") with open('energy_history.json', 'w') as f: - json.dump(energy_history_data['datas'], - f, indent=4, sort_keys=True) + json.dump(energy_history_data.get('datas', []), f, indent=4, sort_keys=True) # Get settings - settings_data = api.min_settings(device_sn=inverter_sn) + settings_data = api.min_settings( + device_sn=inverter_sn, + ) print("Saving settings data to settings_data.json") with open('settings_data.json', 'w') as f: json.dump(settings_data, f, indent=4, sort_keys=True) # Read time segments - tou = api.min_read_time_segments(inverter_sn, settings_data) - print(json.dumps(tou, indent=4)) + tou = api.read_time_segments( + device_sn=inverter_sn, + device_type=device_type, + settings_data=settings_data + ) + print("Time-of-Use Segments:") + with open('tou_data.json', 'w') as f: + json.dump(tou, f, indent=4, sort_keys=True) # Read discharge power - discharge_power = api.min_read_parameter( - inverter_sn, 'discharge_power') - print("Current discharge power:", discharge_power, "%") + discharge_power = api.common_read_parameter( + device_sn=inverter_sn, + device_type=device_type, + parameter_id='discharge_power' + ) + print(f"Current discharge power: {discharge_power}%") # Settings parameters - Uncomment to test # Turn on AC charging -# api.min_write_parameter(inverter_sn, 'ac_charge', 1) -# print("AC charging enabled successfully") - - # Enable Load First between 00:00 and 11:59 using time segment 1 -# api.min_write_time_segment( -# device_sn=inverter_sn, -# segment_id=1, -# batt_mode=growattServer.BATT_MODE_BATTERY_FIRST, -# start_time=datetime.time(0, 0), -# end_time=datetime.time(00, 59), -# enabled=True -# ) -# print("Time segment updated successfully") + # api.min_write_parameter(inverter_sn, 'ac_charge', 1) + # print("AC charging enabled successfully") + # # Enable Load First between 00:00 and 11:59 using time segment 1 + # params = { + # 'segment_id': 1, + # 'start_time': datetime.time(0, 0), + # 'end_time': datetime.time(00, 59), + # 'enabled': True + # } + # api.min_write_time_segment( + # device_sn=inverter_sn, + # params=params + # ) + + # print("Time segment updated successfully") except growattServer.GrowattV1ApiError as e: diff --git a/examples/min_example_dashboard.py b/examples/min_example_dashboard.py index fac0082..baa9e5b 100644 --- a/examples/min_example_dashboard.py +++ b/examples/min_example_dashboard.py @@ -1,10 +1,32 @@ -import growattServer +from . import growattServer import json import requests +import datetime +import os + + +def safe_float(val, default=0.0): + try: + # If already a float, return as is + if isinstance(val, float): + return val + # If it's an int, convert to float + if isinstance(val, int): + return float(val) + # If it's a string, try to parse + if isinstance(val, str): + # Remove any commas, spaces, etc. + val = val.replace(',', '').strip() + return float(val) + # If it's a type that can be cast to float (e.g., numpy.float64) + return float(val) + except (TypeError, ValueError, KeyError, AttributeError): + return default + """ -Example script fetching key power and today+total energy metrics from a Growatt MID-30KTL3-XH (TLX) + APX battery hybrid system -using the V1 API with token-based authentication. +# Example script controlling a MIX/SPH Growatt (SPH3~6k TL BL UP + battery) system using the public growatt API +# You can obtain an API token from the Growatt API documentation or developer portal. """ # Get the API token from user input or environment variable @@ -17,30 +39,33 @@ # Initialize the API with token api = growattServer.OpenApiV1(token=api_token) - # Get plant list using V1 API + # Plant info plants = api.plant_list() + print(f"Plants: Found {plants['count']} plants") plant_id = plants['plants'][0]['plant_id'] + today = datetime.date.today() + devices = api.get_devices(plant_id) - # Get devices in plant - devices = api.device_list(plant_id) - - # Iterate over all devices energy_data = None - for device in devices['devices']: - if device['type'] == 7: # (MIN/TLX) - inverter_sn = device['device_sn'] - - # Get energy data - energy_data = api.min_energy(device_sn=inverter_sn) - with open('energy_data.json', 'w') as f: - json.dump(energy_data, f, indent=4, sort_keys=True) + for device in devices: + # Works automatically for MIN, MIX, or any future device type! + energy_data = device.energy() + print(f"Energy: {energy_data}") + + if energy_data is None: + raise Exception("No MIN_TLX device found to get energy data from.") # energy data does not contain epvToday for some reason, so we need to calculate it - epv_today = energy_data["epv1Today"] + energy_data["epv2Today"] + # Dynamically calculate epvToday by summing all epvXToday fields + epv_today = sum( + safe_float(energy_data.get(f"epv{i}Today"), 0.0) + for i in range(1, 4) # Assuming a maximum of 4 devices + if f"epv{i}Today" in energy_data + ) - solar_production = f'{float(epv_today):.1f}/{float(energy_data["epvTotal"]):.1f}' - solar_production_pv1 = f'{float(energy_data["epv1Today"]):.1f}/{float(energy_data["epv1Total"]):.1f}' - solar_production_pv2 = f'{float(energy_data["epv2Today"]):.1f}/{float(energy_data["epv2Total"]):.1f}' + solar_production = f'{safe_float(epv_today):.1f}/{safe_float(energy_data.get("epvTotal")):.1f}' + solar_production_pv1 = f'{safe_float(energy_data.get("epv1Today")):.1f}/{safe_float(energy_data.get("epv1Total")):.1f}' + solar_production_pv2 = f'{safe_float(energy_data.get("epv2Today")):.1f}/{safe_float(energy_data.get("epv2Total")):.1f}' energy_output = f'{float(energy_data["eacToday"]):.1f}/{float(energy_data["eacTotal"]):.1f}' system_production = f'{float(energy_data["esystemToday"]):.1f}/{float(energy_data["esystemTotal"]):.1f}' battery_charged = f'{float(energy_data["echargeToday"]):.1f}/{float(energy_data["echargeTotal"]):.1f}' @@ -68,22 +93,22 @@ print(f'Export to grid {exported_to_grid:>22}') print("\nPower overview (Watts)") - print(f'AC Power {float(energy_data["pac"]):>22.1f}') - print(f'Self power {float(energy_data["pself"]):>22.1f}') + print(f'AC Power {float(energy_data["pac"]):>22.1f}') + print(f'Self power {float(energy_data["pself"]):>22.1f}') print( - f'Export power {float(energy_data["pacToGridTotal"]):>22.1f}') + f'Export power {float(energy_data["pacToGridTotal"]):>22.1f}') print( - f'Import power {float(energy_data["pacToUserTotal"]):>22.1f}') + f'Import power {float(energy_data["pacToUserTotal"]):>22.1f}') print( - f'Local load power {float(energy_data["pacToLocalLoad"]):>22.1f}') - print(f'PV power {float(energy_data["ppv"]):>22.1f}') - print(f'PV #1 power {float(energy_data["ppv1"]):>22.1f}') - print(f'PV #2 power {float(energy_data["ppv2"]):>22.1f}') + f'Local load power {float(energy_data["pacToLocalLoad"]):>22.1f}') + print(f'PV power {float(energy_data["ppv"]):>22.1f}') + print(f'PV #1 power {float(energy_data["ppv1"]):>22.1f}') + print(f'PV #2 power {float(energy_data["ppv2"]):>22.1f}') print( - f'Battery charge power {float(energy_data["bdc1ChargePower"]):>22.1f}') + f'Battery charge power {float(energy_data["bdc1ChargePower"]):>22.1f}') print( - f'Battery discharge power {float(energy_data["bdc1DischargePower"]):>22.1f}') - print(f'Battery SOC {int(energy_data["bdc1Soc"]):>21}%') + f'Battery discharge power {float(energy_data["bdc1DischargePower"]):>22.1f}') + print(f'Battery SOC {int(energy_data["bdc1Soc"]):>21}%') except growattServer.GrowattV1ApiError as e: print(f"API Error: {e} (Code: {e.error_code}, Message: {e.error_msg})") @@ -92,4 +117,6 @@ except requests.exceptions.RequestException as e: print(f"Network Error: {e}") except Exception as e: + import traceback print(f"Unexpected error: {e}") + traceback.print_exc() diff --git a/examples/mix_example.py b/examples/mix_example.py index b28b3e6..4dcf8ec 100755 --- a/examples/mix_example.py +++ b/examples/mix_example.py @@ -1,7 +1,8 @@ -import growattServer +from . import growattServer import datetime import getpass import pprint +import json """ This is a very trivial script that logs into a user's account and prints out useful data for a "Mix" system (Hybrid). @@ -76,7 +77,16 @@ def indent_print(to_output, indent): #These two API calls return lots of duplicated information, but each also holds unique information as well mix_info = api.mix_info(device_sn, plant_id) pp.pprint(mix_info) + + print("Saving inverter data to old_inverter_data.json") + with open('old_inverter_data.json', 'w') as f: + json.dump(mix_info, f, indent=4, sort_keys=True) + mix_totals = api.mix_totals(device_sn, plant_id) + print("Saving energy data to old_energy_data.json") + with open('old_energy_data.json', 'w') as f: + json.dump(mix_totals, f, indent=4, sort_keys=True) + #pp.pprint(mix_totals) indent_print("*TOTAL VALUES*", 4) indent_print("==Today Totals==", 4) @@ -94,6 +104,10 @@ def indent_print(to_output, indent): print("") mix_detail = api.mix_detail(device_sn, plant_id) + + print("Saving energy data to old_detail_data.json") + with open('old_detail_data.json', 'w') as f: + json.dump(mix_detail, f, indent=4, sort_keys=True) #pp.pprint(mix_detail) #Some of the 'totals' values that are returned by this function do not align to what we would expect, however the graph data always seems to be accurate. diff --git a/examples/mix_v1_example_dashboard.py b/examples/mix_v1_example_dashboard.py new file mode 100644 index 0000000..c7181d6 --- /dev/null +++ b/examples/mix_v1_example_dashboard.py @@ -0,0 +1,112 @@ + +from . import growattServer +import json +import requests +import datetime +import os + + +def safe_float(val, default=0.0): + try: + # If already a float, return as is + if isinstance(val, float): + return val + # If it's an int, convert to float + if isinstance(val, int): + return float(val) + # If it's a string, try to parse + if isinstance(val, str): + # Remove any commas, spaces, etc. + val = val.replace(',', '').strip() + return float(val) + # If it's a type that can be cast to float (e.g., numpy.float64) + return float(val) + except (TypeError, ValueError, KeyError, AttributeError): + return default + + +""" +# Example script controlling a MIX/SPH Growatt (SPH3~6k TL BL UP + battery) system using the public growatt API +# You can obtain an API token from the Growatt API documentation or developer portal. +""" + +# Get the API token from user input or environment variable +api_token = os.environ.get("GROWATT_API_TOKEN") or input("Enter your Growatt API token: ") + +# test token from official API docs https://www.showdoc.com.cn/262556420217021/1494053950115877 +# api_token = "6eb6f069523055a339d71e5b1f6c88cc" # gitleaks:allow + +try: + # Initialize the API with token + api = growattServer.OpenApiV1(token=api_token) + + # Plant info + plants = api.plant_list() + print(f"Plants: Found {plants['count']} plants") + plant_id = plants['plants'][0]['plant_id'] + today = datetime.date.today() + devices = api.get_devices(plant_id) + + energy_data = None + for device in devices: + # Works automatically for MIN, MIX, or any future device type! + device_type = device.device_type + device_sn = device.device_sn + print(f"Device: {device_type} SN: {device_sn}") + energy_data = device.energy() + print(f"Energy: {energy_data}") + + if energy_data is None: + raise Exception("No SPH_MIX device found to get energy data from.") + + solar_production = f'{safe_float(energy_data.get('epvtoday')):.1f}/{safe_float(energy_data.get("epvTotal")):.1f}' + solar_production_pv1 = f'{safe_float(energy_data.get("epv1Today")):.1f}/{safe_float(energy_data.get("epv1Total")):.1f}' + solar_production_pv2 = f'{safe_float(energy_data.get("epv2Today")):.1f}/{safe_float(energy_data.get("epv2Total")):.1f}' + energy_output = f'{safe_float(energy_data.get("eacToday")):.1f}/{safe_float(energy_data.get("eacTotal")):.1f}' + system_production = f'{safe_float(energy_data.get("esystemtoday")):.1f}/{safe_float(energy_data.get("esystemtotal")):.1f}' + battery_charged = f'{safe_float(energy_data.get("echarge1Today")):.1f}/{safe_float(energy_data.get("echarge1Total")):.1f}' + battery_grid_charge = f'{safe_float(energy_data.get("acChargeEnergyToday")):.1f}/{safe_float(energy_data.get("acChargeEnergyTotal")):.1f}' + battery_discharged = f'{safe_float(energy_data.get("edischarge1Today")):.1f}/{safe_float(energy_data.get("edischarge1Total")):.1f}' + exported_to_grid = f'{safe_float(energy_data.get("etoGridToday")):.1f}/{safe_float(energy_data.get("etogridTotal")):.1f}' + imported_from_grid = f'{safe_float(energy_data.get("etoUserToday")):.1f}/{safe_float(energy_data.get("etoUserTotal")):.1f}' + load_consumption = f'{safe_float(energy_data.get("elocalLoadToday")):.1f}/{safe_float(energy_data.get("elocalLoadTotal")):.1f}' + self_consumption = f'{safe_float(energy_data.get("eselfToday")):.1f}/{safe_float(energy_data.get("eselfTotal")):.1f}' + + # Output the dashboard + print("\nGeneration overview Today/Total(kWh)") + print(f'Solar production {solar_production:>22}') + print(f' Solar production, PV1 {solar_production_pv1:>22}') + print(f' Solar production, PV2 {solar_production_pv2:>22}') + print(f'Energy Output {energy_output:>22}') + print(f'System production {system_production:>22}') + print(f'Self consumption {self_consumption:>22}') + print(f'Load consumption {load_consumption:>22}') + print(f'Battery Charged {battery_charged:>22}') + print(f' Charged from grid {battery_grid_charge:>22}') + print(f'Battery Discharged {battery_discharged:>22}') + print(f'Import from grid {imported_from_grid:>22}') + print(f'Export to grid {exported_to_grid:>22}') + + print("\nPower overview (Watts)") + print(f'AC Power {safe_float(energy_data.get("pac")):>22.1f}') + print(f'Self power {safe_float(energy_data.get("pself")):>22.1f}') + print(f'Export power {safe_float(energy_data.get("pacToGridTotal")):>22.1f}') + print(f'Import power {safe_float(energy_data.get("pacToUserTotal")):>22.1f}') + print(f'Local load power {safe_float(energy_data.get("pacToLocalLoad")):>22.1f}') + print(f'PV power {safe_float(energy_data.get("ppv")):>22.1f}') + print(f'PV #1 power {safe_float(energy_data.get("ppv1")):>22.1f}') + print(f'PV #2 power {safe_float(energy_data.get("ppv2")):>22.1f}') + print(f'Battery charge power {safe_float(energy_data.get("bdc1ChargePower")):>22.1f}') + print(f'Battery discharge power {safe_float(energy_data.get("bdc1DischargePower")):>22.1f}') + print(f'Battery SOC {int(safe_float(energy_data.get("bmsSOC"))):>21}%') + +except growattServer.GrowattV1ApiError as e: + print(f"API Error: {e} (Code: {e.error_code}, Message: {e.error_msg})") +except growattServer.GrowattParameterError as e: + print(f"Parameter Error: {e}") +except requests.exceptions.RequestException as e: + print(f"Network Error: {e}") +except Exception as e: + import traceback + print(f"Unexpected error: {e}") + traceback.print_exc() diff --git a/examples/sph_mix_write_example.py b/examples/sph_mix_write_example.py new file mode 100644 index 0000000..4985803 --- /dev/null +++ b/examples/sph_mix_write_example.py @@ -0,0 +1,191 @@ +#!/usr/bin/env python3 +""" +Example script for writing time segments to a Growatt SPH/MIX inverter. + +This example demonstrates how to set AC charge time periods on SPH/MIX devices +using the write_time_segment method with proper parameter objects. +""" + +import os +from datetime import time +import growattServer + + +def main(): + """Main example function.""" + # Get the API token from user input or environment variable + api_token = os.environ.get("GROWATT_API_TOKEN") or input("Enter your Growatt API token: ") + + # Test token from official API docs (for testing only) + # api_token = "6eb6f069523055a339d71e5b1f6c88cc" # gitleaks:allow + + try: + # Initialize the API with token + api = growattServer.OpenApiV1(token=api_token) + + # Get plant information + plants = api.plant_list() + print(f"Found {plants['count']} plants") + + if plants['count'] == 0: + print("No plants found!") + return + + plant_id = plants['plants'][0]['plant_id'] + print(f"Using plant ID: {plant_id}") + + # Get devices + devices = api.device_list(plant_id) + print(f"Found {devices['count']} devices") + + # Find SPH_MIX device + sph_mix_device = None + for device in devices['devices']: + if device['device_type'] == growattServer.DeviceType.SPH_MIX: + sph_mix_device = device + break + + if not sph_mix_device: + print("No SPH/MIX device found!") + return + + device_sn = sph_mix_device['device_sn'] + print(f"Found SPH/MIX device: {device_sn}") + + # Example 1: Set AC charge time period with high power charging + print("\n=== Setting AC Charge Time Period ===") + + charge_params = api.MixAcChargeTimeParams( + charge_power=80, # 80% charging power + charge_stop_soc=95, # Stop charging at 95% SOC + mains_enabled=True, # Enable mains charging + start_hour=23, # Start at 23:00 (11 PM) + start_minute=0, + end_hour=6, # End at 06:00 (6 AM) + end_minute=0, + enabled=True # Enable this time period + ) + + result = api.write_time_segment( + device_sn=device_sn, + device_type=growattServer.DeviceType.SPH_MIX, + command="mix_ac_charge_time_period", + params=charge_params + ) + + print(f"AC charge time period set successfully: {result}") + + # Example 2: Alternative approach using write_parameter method + print("\n=== Alternative: Using write_parameter method ===") + + # Set a different charge time period (off-peak hours) + charge_params2 = api.MixAcChargeTimeParams( + charge_power=100, # 100% charging power + charge_stop_soc=100, # Charge to full + mains_enabled=True, # Enable mains charging + start_hour=1, # Start at 01:00 (1 AM) + start_minute=30, + end_hour=5, # End at 05:00 (5 AM) + end_minute=30, + enabled=True + ) + + result2 = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.SPH_MIX, + command="mix_ac_charge_time_period", + params=charge_params2 + ) + + print(f"Alternative charge time period set: {result2}") + + # Example 3: Disable charging during a specific period + print("\n=== Disabling Charge Time Period ===") + + disable_charge_params = api.MixAcChargeTimeParams( + charge_power=0, # No charging power + charge_stop_soc=0, # No stop SOC + mains_enabled=False, # Disable mains charging + start_hour=12, # Noon + start_minute=0, + end_hour=18, # 6 PM + end_minute=0, + enabled=False # Disable this time period + ) + + result3 = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.SPH_MIX, + command="mix_ac_charge_time_period", + params=disable_charge_params + ) + + print(f"Charge time period disabled: {result3}") + + # Example 4: Read current settings to verify changes + print("\n=== Reading Current Settings ===") + + settings = api.device_settings(device_sn, growattServer.DeviceType.SPH_MIX) + + # Display relevant charge time settings + print("Current charge time settings:") + for i in range(1, 4): # Typically 3 time periods + start_key = f"forcedChargeTimeStart{i}" + stop_key = f"forcedChargeTimeStop{i}" + enabled_key = f"forcedChargeStopSwitch{i}" + + if start_key in settings: + print(f" Period {i}:") + print(f" Start: {settings.get(start_key, 'N/A')}") + print(f" Stop: {settings.get(stop_key, 'N/A')}") + print(f" Enabled: {settings.get(enabled_key, 'N/A')}") + + except growattServer.GrowattV1ApiError as e: + print(f"API Error: {e} (Code: {e.error_code}, Message: {e.error_msg})") + except growattServer.GrowattParameterError as e: + print(f"Parameter Error: {e}") + except Exception as e: + print(f"Unexpected error: {e}") + import traceback + traceback.print_exc() + + +def demonstrate_other_parameters(): + """Demonstrate other parameter types for SPH_MIX devices.""" + print("\n=== Other Parameter Examples ===") + + # Note: These are just examples of parameter creation + # You would use them with api.write_parameter() as shown above + + # Backflow setting + backflow_params = growattServer.OpenApiV1.BackflowSettingParams( + backflow_enabled=True, + anti_reverse_power_percentage=50 # 50% anti-reverse flow power + ) + print(f"Backflow params: {backflow_params}") + + # PV on/off control + pv_params = growattServer.OpenApiV1.PvOnOffParams( + pv_enabled=True # Turn PV on + ) + print(f"PV control params: {pv_params}") + + # Grid voltage limits + voltage_params = growattServer.OpenApiV1.GridVoltageParams( + voltage_high=270, # Upper limit + voltage_low=180 # Lower limit + ) + print(f"Voltage params: {voltage_params}") + + # Off-grid settings + offgrid_params = growattServer.OpenApiV1.OffGridParams( + off_grid_enabled=True, + frequency=0, # 50Hz + voltage=0 # 230V + ) + print(f"Off-grid params: {offgrid_params}") + + +if __name__ == "__main__": + main() + demonstrate_other_parameters() \ No newline at end of file diff --git a/examples/tlx_example.py b/examples/tlx_example.py index 6488cee..4613432 100644 --- a/examples/tlx_example.py +++ b/examples/tlx_example.py @@ -93,18 +93,18 @@ # Examples of updating settings, uncomment to use # Set charging power to 95% -#res = api.update_tlx_inverter_setting(inverter_sn, 'charge_power', 95) -#print(res) +# res = api.update_tlx_inverter_setting(inverter_sn, 'charge_power', 95) +# print(res) # Turn on AC charging -#res = api.update_tlx_inverter_setting(inverter_sn, 'ac_charge', 1) -#print(res) +# res = api.update_tlx_inverter_setting(inverter_sn, 'ac_charge', 1) +# print(res) # Enable Load First between 00:01 and 11:59 using time segment 1 -#res = api.update_tlx_inverter_time_segment(serial_number = inverter_sn, +# res = api.update_tlx_inverter_time_segment(serial_number = inverter_sn, # segment_id = 1, # batt_mode = growattServer.BATT_MODE_LOAD_FIRST, # start_time = datetime.time(00, 1), # end_time = datetime.time(11, 59), # enabled=True) -#print(res) \ No newline at end of file +# print(res) \ No newline at end of file diff --git a/examples/v1_read_example.py b/examples/v1_read_example.py new file mode 100755 index 0000000..d54acd5 --- /dev/null +++ b/examples/v1_read_example.py @@ -0,0 +1,55 @@ +from . import growattServer +import datetime +import json +import requests +import os + +""" +# Example script controlling a MIX/SPH Growatt (SPH3~6k TL BL UP + battery) system using the public growatt API +# You can obtain an API token from the Growatt API documentation or developer portal. +""" + +# Get the API token from user input or environment variable +api_token = os.environ.get("GROWATT_API_TOKEN") or input("Enter your Growatt API token: ") + +# test token from official API docs https://www.showdoc.com.cn/262556420217021/1494053950115877 +# api_token = "6eb6f069523055a339d71e5b1f6c88cc" # gitleaks:allow + +try: + # Initialize the API with token instead of using login + api = growattServer.OpenApiV1(token=api_token) + + # Plant info + plants = api.plant_list() + print(f"Plants: Found {plants['count']} plants") + plant_id = plants['plants'][0]['plant_id'] + today = datetime.date.today() + devices = api.get_devices(plant_id) + + for device in devices: + # Works automatically for MIN, MIX, or any future device type! + device_type = device.device_type + device_sn = device.device_sn + print(f"Device: {device_type} SN: {device_sn}") + # details = device.details() + # energy = device.energy() + # settings = device.settings() + # history = device.energy_history(start_date=today) + read_parameter = device.read_parameter("pv_on_off") + read_time_segments = device.read_time_segments() + + # print(f"Details: {details}") + # print(f"Energy: {energy}") + # print(f"Settings: {settings}") + # print(f"History: {history}") + print(f"Read Parameter PV On/Off: {read_parameter}") + print(f"Read Time Segments: {read_time_segments}") + +except growattServer.GrowattV1ApiError as e: + print(f"API Error: {e} (Code: {e.error_code}, Message: {e.error_msg})") +except growattServer.GrowattParameterError as e: + print(f"Parameter Error: {e}") +except requests.exceptions.RequestException as e: + print(f"Network Error: {e}") +except Exception as e: + print(f"Unexpected error: {e}") diff --git a/examples/v1_write_example.py b/examples/v1_write_example.py new file mode 100755 index 0000000..ac869bb --- /dev/null +++ b/examples/v1_write_example.py @@ -0,0 +1,299 @@ +#!/usr/bin/env python3 +""" +Example script demonstrating generic parameter writing for Growatt devices. + +This script shows how to use the write_parameter() method to set various +parameters on both SPH_MIX and MIN_TLX device types using the V1 API. +""" + +from . import growattServer +import datetime +from datetime import time +import json +import requests +import os + + +def demonstrate_sph_mix_parameters(api, device_sn): + """Demonstrate parameter writing for SPH_MIX devices.""" + print(f"\n=== SPH_MIX Device Parameters for {device_sn} ===") + + try: + # 1. AC Charge Time Period + print("Setting AC charge time period...") + charge_params = api.MixAcChargeTimeParams( + charge_power=80, # 80% charging power + charge_stop_soc=95, # Stop at 95% SOC + mains_enabled=True, # Enable mains charging + start_hour=14, # Start at 14:00 + start_minute=0, + end_hour=16, # End at 16:00 + end_minute=0, + enabled=True + ) + result = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.SPH_MIX, + command="mix_ac_charge_time_period", + params=charge_params + ) + print(f"AC charge time period result: {result}") + + # # 2. AC Discharge Time Period + # print("Setting AC discharge time period...") + # discharge_params = api.MixAcDischargeTimeParams( + # discharge_power=60, # 60% discharge power + # discharge_stop_soc=20, # Stop at 20% SOC + # start_hour=16, # Start at 16:00 + # start_minute=0, + # end_hour=22, # End at 22:00 + # end_minute=0, + # enabled=True + # ) + # result = api.write_parameter( + # device_sn=device_sn, + # device_type=growattServer.DeviceType.SPH_MIX, + # command="mix_ac_discharge_time_period", + # params=discharge_params + # ) + # print(f"AC discharge time period result: {result}") + + # # 3. Backflow Setting + # print("Setting backflow prevention...") + # backflow_params = api.BackflowSettingParams( + # backflow_enabled=True, + # anti_reverse_power_percentage=50 # 50% anti-reverse flow + # ) + # result = api.write_parameter( + # device_sn=device_sn, + # device_type=growattServer.DeviceType.SPH_MIX, + # command="backflow_setting", + # params=backflow_params + # ) + # print(f"Backflow setting result: {result}") + + # # 4. PV On/Off Control + # print("Turning PV on...") + # pv_params = api.PvOnOffParams(pv_enabled=True) + # result = api.write_parameter( + # device_sn=device_sn, + # device_type=growattServer.DeviceType.SPH_MIX, + # command="pv_on_off", + # params=pv_params + # ) + # print(f"PV on/off result: {result}") + + # # 5. Grid Voltage Limits + # print("Setting grid voltage limits...") + # voltage_params = api.GridVoltageParams( + # voltage_high=270, # Upper limit + # voltage_low=180 # Lower limit + # ) + # # Note: These would be separate commands for high and low voltage + # result_high = api.write_parameter( + # device_sn=device_sn, + # device_type=growattServer.DeviceType.SPH_MIX, + # command="pv_grid_voltage_high", + # params=voltage_params + # ) + # print(f"Grid voltage high result: {result_high}") + + # # 6. Off-grid Settings + # print("Setting off-grid parameters...") + # offgrid_params = api.OffGridParams( + # off_grid_enabled=True, + # frequency=0, # 50Hz + # voltage=0 # 230V + # ) + # result = api.write_parameter( + # device_sn=device_sn, + # device_type=growattServer.DeviceType.SPH_MIX, + # command="mix_off_grid_enable", + # params=offgrid_params + # ) + # print(f"Off-grid enable result: {result}") + + except Exception as e: + print(f"Error with SPH_MIX parameters: {e}") + + +def demonstrate_min_tlx_parameters(api, device_sn): + """Demonstrate parameter writing for MIN_TLX devices.""" + print(f"\n=== MIN_TLX Device Parameters for {device_sn} ===") + + try: + # 1. Time Segments (TOU settings) + print("Setting time segment 1...") + time_params = api.TimeSegmentParams( + segment_id=1, + batt_mode=1, # Battery First + start_time=time(8, 0), # 08:00 + end_time=time(16, 0), # 16:00 + enabled=True + ) + result = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.MIN_TLX, + command="time_segment1", + params=time_params + ) + print(f"Time segment 1 result: {result}") + + # 2. Backflow Setting (different params for MIN_TLX) + print("Setting backflow prevention...") + backflow_params = api.BackflowSettingParams( + backflow_enabled=True, + backflow_mode=1 # Enable meter mode for MIN_TLX + ) + result = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.MIN_TLX, + command="backflow_setting", + params=backflow_params + ) + print(f"Backflow setting result: {result}") + + # 3. Charge/Discharge Parameters + print("Setting charge/discharge parameters...") + charge_discharge_params = api.ChargeDischargeParams( + charge_power=90, # 90% charge power + charge_stop_soc=100, # Charge to 100% + discharge_power=80, # 80% discharge power + discharge_stop_soc=10, # Stop at 10% + ac_charge_enabled=True # Enable AC charging + ) + + # Set charge power + result = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.MIN_TLX, + command="charge_power", + params=charge_discharge_params + ) + print(f"Charge power result: {result}") + + # Set discharge power + result = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.MIN_TLX, + command="discharge_power", + params=charge_discharge_params + ) + print(f"Discharge power result: {result}") + + # 4. PV On/Off Control + print("Turning PV on...") + pv_params = api.PvOnOffParams(pv_enabled=True) + result = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.MIN_TLX, + command="tlx_on_off", + params=pv_params + ) + print(f"PV on/off result: {result}") + + # 5. Power Factor Settings + print("Setting power factor...") + power_params = api.PowerParams( + active_power=100, # 100% active power + reactive_power=0, # 0% reactive power + power_factor=1.0 # Unity power factor + ) + result = api.write_parameter( + device_sn=device_sn, + device_type=growattServer.DeviceType.MIN_TLX, + command="pv_power_factor", + params=power_params + ) + print(f"Power factor result: {result}") + + except Exception as e: + print(f"Error with MIN_TLX parameters: {e}") + + +def main(): + """Main demonstration function.""" + # Get the API token from user input or environment variable + api_token = os.environ.get("GROWATT_API_TOKEN") or input("Enter your Growatt API token: ") + + # Test token from official API docs (for testing only) + # api_token = "6eb6f069523055a339d71e5b1f6c88cc" # gitleaks:allow + + try: + # Initialize the API with token + api = growattServer.OpenApiV1(token=api_token) + + # Plant info + plants = api.plant_list() + print(f"Plants: Found {plants['count']} plants") + + if plants['count'] == 0: + print("No plants found!") + return + + plant_id = plants['plants'][0]['plant_id'] + print(f"Using plant ID: {plant_id}") + + # Get devices + devices = api.get_devices(plant_id) + print(f"Found {len(devices)} devices") + + # Demonstrate parameters for each device type + for device in devices: + device_type = device.device_type + device_sn = device.device_sn + + print(f"\n{'='*60}") + print(f"Processing Device: {device_type.name} - SN: {device_sn}") + print(f"{'='*60}") + + if device_type == growattServer.DeviceType.SPH_MIX: + demonstrate_sph_mix_parameters(api, device_sn) + elif device_type == growattServer.DeviceType.MIN_TLX: + demonstrate_min_tlx_parameters(api, device_sn) + else: + print(f"Device type {device_type.name} not supported in this example") + + # Read current settings to verify changes + print(f"\n--- Reading current settings for {device_sn} ---") + try: + settings = device.settings() + print(f"Settings keys: {list(settings.keys())[:10]}...") # Show first 10 keys + + # Show some relevant settings based on device type + if device_type == growattServer.DeviceType.SPH_MIX: + for i in range(1, 4): + start_key = f"forcedChargeTimeStart{i}" + stop_key = f"forcedChargeTimeStop{i}" + if start_key in settings: + print(f" Charge period {i}: {settings.get(start_key)} - {settings.get(stop_key)}") + + elif device_type == growattServer.DeviceType.MIN_TLX: + for i in range(1, 4): + start_key = f"forcedTimeStart{i}" + stop_key = f"forcedTimeStop{i}" + mode_key = f"time{i}Mode" + if start_key in settings: + print(f" Time segment {i}: {settings.get(start_key)} - {settings.get(stop_key)} (Mode: {settings.get(mode_key)})") + + except Exception as e: + print(f"Error reading settings: {e}") + + print(f"\n{'='*60}") + print("Parameter writing demonstration completed!") + print(f"{'='*60}") + + except growattServer.GrowattV1ApiError as e: + print(f"API Error: {e} (Code: {e.error_code}, Message: {e.error_msg})") + except growattServer.GrowattParameterError as e: + print(f"Parameter Error: {e}") + except requests.exceptions.RequestException as e: + print(f"Network Error: {e}") + except Exception as e: + print(f"Unexpected error: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + main() diff --git a/growattServer/__init__.py b/growattServer/__init__.py old mode 100755 new mode 100644 index 927f010..eb8ddfb --- a/growattServer/__init__.py +++ b/growattServer/__init__.py @@ -1,7 +1,7 @@ # Import everything from base_api to ensure backward compatibility from .base_api import * # Import the V1 API class -from .open_api_v1 import OpenApiV1 +from .open_api_v1 import OpenApiV1, DeviceType # Import exceptions from .exceptions import GrowattError, GrowattParameterError, GrowattV1ApiError diff --git a/growattServer/base_api.py b/growattServer/base_api.py index 0f74fff..bdfbc9c 100644 --- a/growattServer/base_api.py +++ b/growattServer/base_api.py @@ -168,6 +168,18 @@ def plant_list(self, user_id): ) return response.json().get('back', []) + + def plant_energy_overview(self, plant_id: str) -> dict: + """ + Fetches energy overview for a plant. + Replace this stub with actual API logic. + """ + # Example return structure + return { + "today_energy": 0.0, + "total_energy": 0.0, + "current_power": 0.0, + } def plant_detail(self, plant_id, timespan, date=None): """ diff --git a/growattServer/open_api_v1.py b/growattServer/open_api_v1.py index 02ae3cc..a5ffe79 100644 --- a/growattServer/open_api_v1.py +++ b/growattServer/open_api_v1.py @@ -1,32 +1,386 @@ +import json # noqa: D100 +import os +import platform +import re +from unittest import case import warnings -from datetime import date, timedelta +from datetime import UTC, date, datetime, time, timedelta +from enum import Enum +from typing import ClassVar, NamedTuple, Optional + +# Load environment variables from .env file (optional, for development) +try: + from dotenv import load_dotenv + + load_dotenv() +except ImportError: + # dotenv not available, skip loading .env file + pass + from . import GrowattApi -import platform from .exceptions import GrowattParameterError, GrowattV1ApiError +class DeviceType(Enum): + """Enumeration of Growatt device types.""" + + INVERTER = 1 + STORAGE = 2 + OTHER = 3 + MAX = 4 + SPH_MIX = 5 + SPA = 6 + MIN_TLX = 7 + PCS = 8 + HPS = 9 + PBD = 10 + + @classmethod + def get_url_prefix(cls, device_type) -> str: # noqa: ANN001 + """Get the URL prefix for a given device type.""" + if device_type == cls.SPH_MIX: + return "mix" + elif device_type == cls.SPH_MIX.value: # noqa: RET505 + return "mix" + elif device_type == cls.MIN_TLX: # noqa: RET505 + return "tlx" + elif device_type == cls.MIN_TLX.value: # noqa: RET505 + return "tlx" + else: + msg = f"Unsupported device type: {device_type}" + raise GrowattParameterError(msg) + + @classmethod + def get_url_read_param(cls, device_type: "DeviceType") -> str: + """Get the URL param for a given device type.""" + if device_type == cls.SPH_MIX: + return "readMixParam" + elif device_type == cls.MIN_TLX: # noqa: RET505 + return "readMinParam" + else: + msg = f"Unsupported device type: {device_type}" + raise GrowattParameterError(msg) + + +class DeviceFieldTemplates: + """Template strings for device field names.""" + + MIN_TLX_TEMPLATES: ClassVar = { + "start_time": "forcedTimeStart{segment_id}", + "stop_time": "forcedTimeStop{segment_id}", + "mode": "time{segment_id}Mode", + "enabled": "forcedStopSwitch{segment_id}", + } + + SPH_MIX_TEMPLATES_CHARGE: ClassVar = { + "start_time": "forcedChargeTimeStart{segment_id}", + "stop_time": "forcedChargeTimeStop{segment_id}", + "mode": "Battery First", + "enabled": "forcedChargeStopSwitch{segment_id}", + } + + SPH_MIX_TEMPLATES_DIS_CHARGE: ClassVar = { + "start_time": "forcedDischargeTimeStart{segment_id}", + "stop_time": "forcedDischargeTimeStop{segment_id}", + "mode": "Battery First", + "enabled": "forcedDischargeStopSwitch{segment_id}", + } + + +class ApiDataType(Enum): + """Enumeration of Growatt device types.""" + + LAST_DATA = "last_data" + HISTORY_DATA = "history_data" + BASIC_INFO = "basic_info" + DEVICE_SETTINGS = "settings" + READ_PARAM = "read_param" + SET_PARAM = "set_param" + + class OpenApiV1(GrowattApi): """ Extended Growatt API client with V1 API support. - This class extends the base GrowattApi class with methods for MIN inverters using - the public V1 API described here: https://www.showdoc.com.cn/262556420217021/0 + + This class extends the base GrowattApi class with methods for MIN/TLX and MIX/SPH + inverters using the public V1 API described here: + https://www.showdoc.com.cn/262556420217021/0. """ - def _create_user_agent(self): + DEVICE_ENDPOINTS: ClassVar = { + DeviceType.SPH_MIX: { + # https://www.showdoc.com.cn/262556420217021/6129764434976910 + ApiDataType.LAST_DATA: "device/mix/mix_last_data", + # https://www.showdoc.com.cn/262556420217021/6129763571291058 + ApiDataType.BASIC_INFO: "device/mix/mix_data_info", + # https://www.showdoc.com.cn/262556420217021/6129765461123058 + ApiDataType.HISTORY_DATA: "device/mix/mix_data", + # https://www.showdoc.com.cn/262556420217021/6129763571291058 + ApiDataType.DEVICE_SETTINGS: "device/mix/mix_data_info", + # https://www.showdoc.com.cn/262556420217021/6129766954561259 + ApiDataType.READ_PARAM: "readMixParam", + # https://www.showdoc.com.cn/262556420217021/6129761750718760 + ApiDataType.SET_PARAM: "mixSet", + }, + DeviceType.MIN_TLX: { + # https://www.showdoc.com.cn/262556420217021/6129822090975531 + ApiDataType.LAST_DATA: "device/tlx/tlx_last_data", + # https://www.showdoc.com.cn/262556420217021/6129816412127075 + ApiDataType.BASIC_INFO: "device/tlx/tlx_data_info", + # https://www.showdoc.com.cn/262556420217021/8559849784929961 + ApiDataType.HISTORY_DATA: "device/tlx/tlx_data", + # https://www.showdoc.com.cn/262556420217021/8696815667375182 + ApiDataType.DEVICE_SETTINGS: "device/tlx/tlx_set_info", + # https://www.showdoc.com.cn/262556420217021/6129828239577315 + ApiDataType.READ_PARAM: "readMinParam", + # https://www.showdoc.com.cn/262556420217021/6129826876191828 + ApiDataType.SET_PARAM: "tlxSet", + }, + } + + class TimeSegmentParams(NamedTuple): + """ + Parameters for a time segment in a MIN inverter. + + segment_id (int): Segment number (1-9). + batt_mode (int): Battery mode (0=Load First, 1=Battery First, 2=Grid First). + start_time (object): Start time (should be datetime.time). + end_time (object): End time (should be datetime.time). + enabled (bool): Whether the segment is enabled. + """ + + segment_id: int + batt_mode: int + start_time: time # Should be datetime.time + end_time: time # Should be datetime.time + enabled: bool = True + + class MixAcDischargeTimeParams(NamedTuple): + """Parameters for SPH_MIX AC discharge time period.""" + + discharge_power: int # 0-100 + discharge_stop_soc: int # 0-100 + start_hour: int # 0-23 + start_minute: int # 0-59 + end_hour: int # 0-23 + end_minute: int # 0-59 + enabled: bool = True + segment_id: int = 1 # Period number (1-6) + + class MixAcChargeTimeParams(NamedTuple): + """Parameters for SPH_MIX AC charge time period.""" + + charge_power: int # 0-100 + charge_stop_soc: int # 0-100 + mains_enabled: bool # True=enable, False=disable + start_hour: int # 0-23 + start_minute: int # 0-59 + end_hour: int # 0-23 + end_minute: int # 0-59 + enabled: bool = True + segment_id: int = 1 # Period number (1-6) + + class BackflowSettingParams(NamedTuple): + """Parameters for backflow prevention setting.""" + + backflow_enabled: bool # True=on, False=off + anti_reverse_power_percentage: int = 0 # 0-100 (for SPH_MIX) + backflow_mode: int = 0 # 0=disable, 1=enable meter, 2=enable CT (for MIN_TLX) + + class PvOnOffParams(NamedTuple): + """Parameters for PV on/off setting.""" + + pv_enabled: bool # True=on (0001), False=off (0000) + + class GridVoltageParams(NamedTuple): + """Parameters for grid voltage limits.""" + + voltage_high: int = 270 # Upper limit + voltage_low: int = 180 # Lower limit + + class OffGridParams(NamedTuple): + """Parameters for off-grid settings.""" + + off_grid_enabled: bool # True=enabled, False=disabled + frequency: int = 0 # 0=50HZ, 1=60HZ + voltage: int = 0 # 0=230V, 1=208V, 2=240V + + class PowerParams(NamedTuple): + """Parameters for power settings.""" + + active_power: int = 0 # 0-100 + reactive_power: int = 0 # 0-100 + power_factor: float = 1.0 # -0.8 to -1 or 0.8 to 1 + + class ChargeDischargeParams(NamedTuple): + """Parameters for charge/discharge settings.""" + + charge_power: int = 0 # 0-100 + charge_stop_soc: int = 0 # 0-100 + discharge_power: int = 0 # 0-100 + discharge_stop_soc: int = 0 # 0-100 + ac_charge_enabled: bool = False # Mains charging enabled + + class WriteParamsInterface: + """Interface for converting parameter objects to API format.""" + + @staticmethod + def time_segment_to_params(params: "OpenApiV1.TimeSegmentParams") -> dict: + """Convert TimeSegmentParams to API parameters for MIN_TLX.""" + return { + "param1": str(params.batt_mode), + "param2": str(params.start_time.hour), + "param3": str(params.start_time.minute), + "param4": str(params.end_time.hour), + "param5": str(params.end_time.minute), + "param6": "1" if params.enabled else "0", + } + + @staticmethod + def mix_discharge_to_params( + params: "OpenApiV1.MixAcDischargeTimeParams", period: int = 1 + ) -> dict: + """Convert MixAcDischargeTimeParams to API parameters for SPH_MIX.""" + base_param = (period - 1) * 5 # Each period uses 5 parameters + return { + f"param1": str(params.discharge_power), + f"param2": str(params.discharge_stop_soc), + f"param{base_param + 3}": str(params.start_hour), + f"param{base_param + 4}": str(params.start_minute), + f"param{base_param + 5}": str(params.end_hour), + f"param{base_param + 6}": str(params.end_minute), + f"param{base_param + 7}": "1" if params.enabled else "0", + } + + @staticmethod + def mix_charge_to_params( + params: "OpenApiV1.MixAcChargeTimeParams", period: int = 1 + ) -> dict: + """Convert MixAcChargeTimeParams to API parameters for SPH_MIX.""" + base_param = (period - 1) * 5 # Each period uses 5 parameters + return { + f"param1": str(params.charge_power), + f"param2": str(params.charge_stop_soc), + f"param3": "1" if params.mains_enabled else "0", + f"param{base_param + 4}": str(params.start_hour), + f"param{base_param + 5}": str(params.start_minute), + f"param{base_param + 6}": str(params.end_hour), + f"param{base_param + 7}": str(params.end_minute), + f"param{base_param + 8}": "1" if params.enabled else "0", + } + + @staticmethod + def backflow_to_params( + params: "OpenApiV1.BackflowSettingParams", device_type: DeviceType + ) -> dict: + """Convert BackflowSettingParams to API parameters.""" + if device_type == DeviceType.SPH_MIX: + return { + "param1": "1" if params.backflow_enabled else "0", + "param2": str(params.anti_reverse_power_percentage), + } + elif device_type == DeviceType.MIN_TLX: + return { + "param1": str(params.backflow_mode), + } + else: + msg = f"Unsupported device type: {device_type}" + raise GrowattParameterError(msg) + + @staticmethod + def pv_onoff_to_params(params: "OpenApiV1.PvOnOffParams") -> dict: + """Convert PvOnOffParams to API parameters.""" + return { + "param1": "0001" if params.pv_enabled else "0000", + } + + @staticmethod + def grid_voltage_to_params(params: "OpenApiV1.GridVoltageParams") -> dict: + """Convert GridVoltageParams to API parameters.""" + return { + "param1": str(params.voltage_high), # For high voltage + # OR + "param1": str( + params.voltage_low + ), # For low voltage (different command) + } + + @staticmethod + def off_grid_to_params(params: "OpenApiV1.OffGridParams") -> dict: + """Convert OffGridParams to API parameters.""" + return { + "param1": "1" if params.off_grid_enabled else "0", + # OR for frequency/voltage (different commands) + "param1": str(params.frequency), + # OR + "param1": str(params.voltage), + } + + @staticmethod + def power_to_params(params: "OpenApiV1.PowerParams") -> dict: + """Convert PowerParams to API parameters.""" + return { + "param1": str(params.active_power), # For active power + # OR + "param1": str(params.reactive_power), # For reactive power + # OR + "param1": str(params.power_factor), # For power factor + } + + @staticmethod + def charge_discharge_to_params( + params: "OpenApiV1.ChargeDischargeParams", + ) -> dict: + """Convert ChargeDischargeParams to API parameters.""" + return { + "param1": str(params.charge_power), # For charge_power command + # OR + "param1": str(params.charge_stop_soc), # For charge_stop_soc command + # OR + "param1": str(params.discharge_power), # For discharge_power command + # OR + "param1": str( + params.discharge_stop_soc + ), # For discharge_stop_soc command + # OR + "param1": "1" + if params.ac_charge_enabled + else "0", # For ac_charge command + } + + class DeviceEnergyHistoryParams(NamedTuple): + """Parameters for querying device energy history.""" + + start_date: date | None = None + end_date: date | None = None + timezone: str | None = None + page: int | None = None + limit: int | None = None + + class PlantEnergyHistoryParams(NamedTuple): + """Parameters for querying plant energy history.""" + + start_date: date | None = None + end_date: date | None = None + time_unit: str = "day" + page: int | None = None + perpage: int | None = None + + def _create_user_agent(self) -> str: python_version = platform.python_version() system = platform.system() release = platform.release() machine = platform.machine() - user_agent = f"Python/{python_version} ({system} {release}; {machine})" - return user_agent + return f"Python/{python_version} ({system} {release}; {machine})" - def __init__(self, token): + def __init__(self, token: str) -> None: """ Initialize the Growatt API client with V1 API support. Args: token (str): API token for authentication (required for V1 API access). + """ # Initialize the base class super().__init__(agent_identifier=self._create_user_agent()) @@ -37,7 +391,29 @@ def __init__(self, token): # 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"): + @staticmethod + def slugify(s: str) -> str: + """ + Convert a string to a slug by removing non-word characters. + + Replace whitespace with underscores. + + Args: + s (str): The string to slugify. + + Returns: + str: The slugified string. + + """ + # Remove all non-word characters (everything except numbers and letters) + s = re.sub(r"[^\w\s]", "", s) + + # Replace all runs of whitespace with a single underscrore + return re.sub(r"\s+", "_", s) + + def _process_response( + self, response: dict, operation_name: str = "API operation" + ) -> dict: """ Process API response and handle errors. @@ -50,49 +426,80 @@ def _process_response(self, response, operation_name="API operation"): Raises: GrowattV1ApiError: If the API returns an error response + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ - if response.get('error_code', 1) != 0: + # Only write debug output if DEBUG environment variable is set to true + if os.getenv("DEBUG", "false").lower() == "true": + with open("response.json", "w") as f: + json.dump(response, f, indent=4, sort_keys=True) + + if response.get("error_code", 1) != 0: + msg = f"Error during {operation_name}" raise GrowattV1ApiError( - f"Error during {operation_name}", - error_code=response.get('error_code'), - error_msg=response.get('error_msg', 'Unknown error') + msg, + error_code=response.get("error_code"), + error_msg=response.get("error_msg", "Unknown error"), ) - return response.get('data') - def _get_url(self, page): - """ - Simple helper function to get the page URL for v1 API. - """ + return response.get("data") or {} + + def _get_url(self, page: str) -> str: + """Return the page URL for v1 API.""" return self.api_url + page - def plant_list(self): + def _get_device_url(self, device_type: DeviceType, operation: ApiDataType) -> str: + """ + Get the API URL for a specific device type and operation. + + Args: + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX) + operation (str): The operation to perform ('energy', 'settings', etc.) + + Returns: + str: The complete API URL for the operation + + Raises: + GrowattParameterError: If the device type or operation is not supported + + """ + if device_type not in self.DEVICE_ENDPOINTS: + msg = f"Unsupported device type: {device_type}" + raise GrowattParameterError(msg) + if operation not in self.DEVICE_ENDPOINTS[device_type]: + msg = f"Unsupported operation '{operation}' for device type {device_type}" + raise GrowattParameterError(msg) + return self.api_url + self.DEVICE_ENDPOINTS[device_type][operation] + + def plant_list(self) -> dict: """ Get a list of all plants with detailed information. Returns: - dict: A dictionary containing plants information with 'count' and 'plants' keys. + dict: A dictionary containing plants information. + Includes 'count' and 'plants' keys. Raises: GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ # Prepare request data request_data = { - 'page': '', - 'perpage': '', - 'search_type': '', - 'search_keyword': '' + "page": "", + "perpage": "", + "search_type": "", + "search_keyword": "", } # Make the request - response = self.session.get( - url=self._get_url('plant/list'), - data=request_data - ) + response = self.session.get(url=self._get_url("plant/list"), data=request_data) return self._process_response(response.json(), "getting plant list") - def plant_details(self, plant_id): + def plant_details(self, plant_id: int) -> dict: """ Get basic information about a power station. @@ -104,42 +511,45 @@ def plant_details(self, plant_id): Raises: GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. - """ + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ response = self.session.get( - self._get_url('plant/details'), - params={'plant_id': plant_id} + self._get_url("plant/details"), params={"plant_id": plant_id} ) return self._process_response(response.json(), "getting plant details") - def plant_energy_overview(self, plant_id): + def plant_energy_overview(self, plant_id: str) -> dict: """ Get an overview of a plant's energy data. Args: - plant_id (int): Power Station ID + plant_id (str): Power Station ID Returns: dict: A dictionary containing the plant energy overview. Raises: GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. - """ + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ response = self.session.get( - self._get_url('plant/data'), - params={'plant_id': plant_id} + self._get_url("plant/data"), params={"plant_id": plant_id} ) return self._process_response(response.json(), "getting plant energy overview") - def plant_power_overview(self, plant_id: int, day: str | date = None) -> dict: + def plant_power_overview( + self, plant_id: int, day: str | date | None = None + ) -> dict: """ Obtain power data of a certain power station. - Get the frequency once every 5 minutes + + Get the frequency once every 5 minutes. Args: plant_id (int): Power Station ID @@ -156,36 +566,38 @@ def plant_power_overview(self, plant_id: int, day: str | date = None) -> dict: # 'time': str, # Time of the power reading # 'power': float | None # Power value in Watts (can be None) } + Raises: GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. API-Doc: https://www.showdoc.com.cn/262556420217021/1494062656174173 + """ if day is None: - day = date.today() + day = datetime.now(tz=UTC).date() response = self.session.get( - self._get_url('plant/power'), + self._get_url("plant/power"), params={ - 'plant_id': plant_id, - 'date': day, - } + "plant_id": plant_id, + "date": str(day), + }, ) 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): + def plant_energy_history( + self, plant_id: int, params: Optional["PlantEnergyHistoryParams"] = None + ) -> dict: """ Retrieve plant energy data for multiple days/months/years. Args: plant_id (int): Power Station ID - start_date (date, optional): Start Date - defaults to today - end_date (date, optional): End Date - defaults to today - time_unit (str, optional): Time unit ('day', 'month', 'year') - defaults to 'day' - page (int, optional): Page number - defaults to 1 - perpage (int, optional): Number of items per page - defaults to 20, max 100 + params (PlantEnergyHistoryParams): + Grouped parameters for energy history query. Returns: dict: A dictionary containing the plant energy history. @@ -197,98 +609,167 @@ def plant_energy_history(self, plant_id, start_date=None, end_date=None, time_un Raises: GrowattParameterError: If date parameters are invalid. - GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. + """ + if params is None: + params = self.PlantEnergyHistoryParams() + start_date = params.start_date + end_date = params.end_date + time_unit = params.time_unit + page = params.page + perpage = params.perpage + end_date = params.end_date + time_unit = params.time_unit + page = params.page + perpage = params.perpage if start_date is None and end_date is None: - start_date = date.today() - end_date = date.today() + today = datetime.now(tz=UTC).date() + start_date = today + end_date = today elif start_date is None: - start_date = end_date + start_date = ( + end_date if end_date is not None else datetime.now(tz=UTC).date() + ) elif end_date is None: - end_date = start_date + end_date = ( + start_date if start_date is not None else datetime.now(tz=UTC).date() + ) # Validate date ranges based on time_unit - if time_unit == "day" and (end_date - start_date).days > 7: + if ( + time_unit == "day" + and start_date is not None + and end_date is not None + and (end_date - start_date).days > 7 # noqa: PLR2004 + ): warnings.warn( - "Date interval must not exceed 7 days in 'day' mode.", RuntimeWarning) - elif time_unit == "month" and (end_date.year - start_date.year > 1): + "Date interval must not exceed 7 days in 'day' mode.", + RuntimeWarning, + stacklevel=2, + ) + elif ( + time_unit == "month" + and start_date is not None + and end_date is not None + and (end_date.year - start_date.year > 1) + ): warnings.warn( - "Start date must be within same or previous year in 'month' mode.", RuntimeWarning) - elif time_unit == "year" and (end_date.year - start_date.year > 20): + "Start date must be within same or previous year in 'month' mode.", + RuntimeWarning, + stacklevel=2, + ) + elif ( + time_unit == "year" + and start_date is not None + and end_date is not None + and (end_date.year - start_date.year > 20) # noqa: PLR2004 + ): warnings.warn( - "Date interval must not exceed 20 years in 'year' mode.", RuntimeWarning) + "Date interval must not exceed 20 years in 'year' mode.", + RuntimeWarning, + stacklevel=2, + ) + + # Ensure start_date and end_date are not None + if start_date is None: + start_date = datetime.now(tz=UTC).date() + if end_date is None: + end_date = datetime.now(tz=UTC).date() + + # Ensure start_date and end_date are not None + if start_date is None: + start_date = datetime.now(tz=UTC).date() + if end_date is None: + end_date = datetime.now(tz=UTC).date() 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"), - 'end_date': end_date.strftime("%Y-%m-%d"), - 'time_unit': time_unit, - 'page': page, - 'perpage': perpage - } + "plant_id": plant_id, + "start_date": start_date.strftime("%Y-%m-%d"), + "end_date": ( + (end_date or datetime.now(tz=UTC).date()).strftime("%Y-%m-%d") + ), + "time_unit": time_unit, + "page": page, + "perpage": perpage, + }, ) return self._process_response(response.json(), "getting plant energy history") - def device_list(self, plant_id): + def device_list(self, plant_id: int) -> dict: """ - Get devices associated with plant. + Get a list of devices in a plant. - Note: - returned "device_type" mappings: - 1: inverter (including MAX) - 2: storage - 3: other - 4: max (single MAX) - 5: sph - 6: spa - 7: min (including TLX) - 8: pcs - 9: hps - 10: pbd + The device list includes type information that maps to the DeviceType enum + (5 for SPH_MIX, 7 for MIN_TLX). Args: - plant_id (int): Power Station ID + plant_id (int): The plant ID to get devices for. Returns: - DeviceList - e.g. - { - "data": { - "count": 3, - "devices": [ - { - "device_sn": "ZT00100001", - "last_update_time": "2018-12-13 11:03:52", - "model": "A0B0D0T0PFU1M3S4", - "lost": True, - "status": 0, - "manufacturer": "Growatt", - "device_id": 116, - "datalogger_sn": "CRAZT00001", - "type": 1 - }, - ] - }, - "error_code": 0, - "error_msg": "" - } + dict: A dictionary containing device information with 'count' and + 'devices' fields. + Each device includes a 'type' field that maps to DeviceType enum + values. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ response = self.session.get( - url=self._get_url("device/list"), - params={ - "plant_id": plant_id, - "page": "", - "perpage": "", - }, + url=self._get_url("device/list"), params={"plant_id": plant_id} + ) + + data = self._process_response(response.json(), "getting device list") + + # Add device_type mapping for each device based on type + if "devices" in data: + for device in data["devices"]: + try: + device_type = device.get("type") + if device_type: + device["device_type"] = DeviceType(device_type) + except ValueError: + device["device_type"] = None + return data + + def device_details(self, device_sn: str, device_type: DeviceType) -> dict: + """ + Get detailed data for a device. + + Args: + device_sn (str): The serial number of the device. + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + + Returns: + dict: A dictionary containing the device details. + + Raises: + GrowattParameterError: If the device type is not supported. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + + """ + if not isinstance(device_type, DeviceType): + msg = f"Invalid device type: {device_type}" + raise GrowattParameterError(msg) + + response = self.session.get( + self._get_device_url(device_type, ApiDataType.BASIC_INFO), + params={"device_sn": device_sn}, + ) + + return self._process_response( + response.json(), f"getting {device_type.name} details" ) - return self._process_response(response.json(), "getting device list") - def min_detail(self, device_sn): + def min_detail(self, device_sn: str) -> dict: """ Get detailed data for a MIN inverter. @@ -300,22 +781,55 @@ def min_detail(self, device_sn): Raises: GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ + return self.device_details(device_sn, DeviceType.MIN_TLX) - response = self.session.get( - self._get_url('device/tlx/tlx_data_info'), - params={ - 'device_sn': device_sn - } + def device_energy(self, device_sn: str, device_type: DeviceType) -> dict: + """ + Get energy data for a device. + + Args: + device_sn (str): The serial number of the device. + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + + Returns: + dict: A dictionary containing the device energy data. + + Raises: + GrowattParameterError: If the device type is not supported. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + + """ + if not isinstance(device_type, DeviceType): + msg = f"Invalid device type: {device_type}" + raise GrowattParameterError(msg) + + url_prefix = DeviceType.get_url_prefix(device_type) + + response = self.session.post( + url=self._get_device_url(device_type, ApiDataType.LAST_DATA), + data={f"{url_prefix}_sn": device_sn}, ) - return self._process_response(response.json(), "getting MIN inverter details") + # responseHydrated = self._process_response(response.json(), f"getting {device_type.name} energy data") + + # responseHydrated['epvToday'] = responseHydrated.get('epv1Today', 0) + responseHydrated.get("epv2Today", 0) - def min_energy(self, device_sn): + return self._process_response( + response.json(), f"getting {device_type.name} energy data" + ) + + def min_energy(self, device_sn: str) -> dict: """ Get energy data for a MIN inverter. + return self.device_energy(device_sn, DeviceType.MIN_TLX). + Args: device_sn (str): The serial number of the MIN inverter. @@ -324,98 +838,217 @@ def min_energy(self, device_sn): Raises: GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ + return self.device_energy(device_sn, DeviceType.MIN_TLX) - response = self.session.post( - url=self._get_url("device/tlx/tlx_last_data"), - data={ - "tlx_sn": device_sn, - }, + def device_settings(self, device_sn: str, device_type: DeviceType) -> dict: + """ + Get settings for a device. + + Args: + device_sn (str): The serial number of the device. + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + + Returns: + dict: A dictionary containing the device settings. + + Raises: + GrowattParameterError: If the device type is not supported. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + + """ + if not isinstance(device_type, DeviceType): + msg = f"Invalid device type: {device_type}" + raise GrowattParameterError(msg) + + response = self.session.get( + self._get_device_url(device_type, ApiDataType.DEVICE_SETTINGS), + params={"device_sn": device_sn}, ) - return self._process_response(response.json(), "getting MIN inverter energy data") + return self._process_response( + response.json(), f"getting {device_type.name} settings" + ) - def min_energy_history(self, device_sn, start_date=None, end_date=None, timezone=None, page=None, limit=None): + def min_settings(self, device_sn: str) -> dict: """ - Get MIN inverter data history. + Get settings for a MIN inverter. 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. + device_sn (str): The serial number of the MIN inverter. Returns: - dict: A dictionary containing the MIN inverter history data. + 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. + + """ + return self.device_settings(device_sn, DeviceType.MIN_TLX) + + def device_energy_history( + self, + device_sn: str, + device_type: DeviceType, + params: Optional["DeviceEnergyHistoryParams"] = None, + ) -> dict: + """ + Get device energy history data. + + Args: + device_sn (str): The ID of the device. + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + params (DeviceEnergyHistoryParams, optional): + Grouped parameters for energy history query. + + Returns: + dict: A dictionary containing the device history data. Raises: - GrowattParameterError: If date interval is invalid (exceeds 7 days). + GrowattParameterError: If device type is invalid or date interval + exceeds 7 days. GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ + if not isinstance(device_type, DeviceType): + msg = f"Invalid device type: {device_type}" + raise GrowattParameterError(msg) + + if params is None: + params = self.DeviceEnergyHistoryParams() + + start_date = params.start_date + end_date = params.end_date + timezone = params.timezone + page = params.page + limit = params.limit if start_date is None and end_date is None: - start_date = date.today() - end_date = date.today() + start_date = datetime.now(tz=UTC).date() + end_date = datetime.now(tz=UTC).date() elif start_date is None: - start_date = end_date + start_date = ( + end_date if end_date is not None else datetime.now(tz=UTC).date() + ) elif end_date is None: - end_date = start_date + end_date = ( + start_date if start_date is not None else datetime.now(tz=UTC).date() + ) # check interval validity - if end_date - start_date > timedelta(days=7): - raise GrowattParameterError("date interval must not exceed 7 days") + if ( + start_date is not None + and end_date is not None + and (end_date - start_date > timedelta(days=7)) + ): + msg = "date interval must not exceed 7 days" + raise GrowattParameterError(msg) + + url_prefix = DeviceType.get_url_prefix(device_type) + # Ensure end_date is not None before formatting + if end_date is None: + end_date = datetime.now(tz=UTC).date() response = self.session.post( - url=self._get_url('device/tlx/tlx_data'), + url=self._get_device_url(device_type, ApiDataType.HISTORY_DATA), data={ - "tlx_sn": device_sn, + f"{url_prefix}_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 self._process_response( + response.json(), f"getting {device_type.name} energy history" + ) - def min_settings(self, device_sn): + def min_energy_history( + self, device_sn: str, params: Optional["DeviceEnergyHistoryParams"] = None + ) -> dict: """ - Get settings for a MIN inverter. + Get energy history data for a MIN inverter. Args: - device_sn (str): The serial number of the MIN inverter. + device_sn (str): The ID of the MIN inverter. + params (DeviceEnergyHistoryParams, optional): + Grouped parameters for energy history query. Returns: - dict: A dictionary containing the MIN inverter settings. + dict: A dictionary containing the MIN inverter history data. Raises: + GrowattParameterError: If date interval exceeds 7 days. GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ + return self.device_energy_history(device_sn, DeviceType.MIN_TLX, params) - response = self.session.get( - self._get_url('device/tlx/tlx_set_info'), - params={ - 'device_sn': device_sn - } - ) + def min_read_parameter(self, device_sn: str, params: dict) -> dict: + """ + Read setting from MIN inverter. + + Args: + device_sn (str): The ID of the TLX inverter. + params (dict): Parameters for reading the setting. + Should include either 'parameter_id' or 'start_address' and 'end_address'. + + Returns: + dict: A dictionary containing the setting value. - return self._process_response(response.json(), "getting MIN inverter settings") + 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. - def min_read_parameter(self, device_sn, parameter_id, start_address=None, end_address=None): + """ + parameter_id = ( + str(params.get("parameter_id")) + if params.get("parameter_id") is not None + else "" + ) + return self.common_read_parameter( + device_sn, + DeviceType.MIN_TLX, + parameter_id, + params.get("start_address"), + params.get("end_address"), + ) + + def common_read_parameter( + self, + device_sn: str, + device_type: DeviceType, + parameter_id: str, + start_address: int | None = None, + end_address: int | None = 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. + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + 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. @@ -423,16 +1056,20 @@ def min_read_parameter(self, device_sn, parameter_id, start_address=None, end_ad 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. + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ + if not isinstance(device_type, DeviceType): + msg = f"Invalid device type: {device_type}" + raise GrowattParameterError(msg) if parameter_id is None and start_address is None: - raise GrowattParameterError( - "specify either parameter_id or start_address/end_address") - elif parameter_id is not None and start_address is not None: - raise GrowattParameterError( - "specify either parameter_id or start_address/end_address - not both." - ) + msg = "specify either parameter_id or start_address/end_address" + raise GrowattParameterError(msg) + elif parameter_id is not None and start_address is not None: # noqa: RET506 + msg = "specify either parameter_id or start_address/end_address - not both." + raise GrowattParameterError(msg) elif parameter_id is not None: # named parameter start_address = 0 @@ -446,18 +1083,22 @@ def min_read_parameter(self, device_sn, parameter_id, start_address=None, end_ad end_address = start_address response = self.session.post( - self._get_url('readMinParam'), + self._get_device_url(device_type, ApiDataType.READ_PARAM), 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 self._process_response( + response.json(), f"reading parameter {parameter_id}" + ) - def min_write_parameter(self, device_sn, parameter_id, parameter_values=None): + def min_write_parameter( + self, device_sn: str, parameter_id: str, parameter_values: object = None + ) -> dict: """ Set parameters on a MIN inverter. @@ -474,11 +1115,12 @@ def min_write_parameter(self, device_sn, parameter_id, parameter_values=None): Raises: GrowattV1ApiError: If the API returns an error response. - requests.exceptions.RequestException: If there is an issue with the HTTP request. - """ + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + """ # Initialize all parameters as empty strings - parameters = {i: "" for i in range(1, 20)} + parameters = dict.fromkeys(range(1, 20), "") # Process parameter values based on type if parameter_values is not None: @@ -488,44 +1130,40 @@ def min_write_parameter(self, device_sn, parameter_id, parameter_values=None): elif isinstance(parameter_values, list): # List of values go to sequential params for i, value in enumerate(parameter_values, 1): - if i <= 19: # Only use up to 19 parameters + if i <= 19: # Only use up to 19 parameters # noqa: PLR2004 parameters[i] = str(value) elif isinstance(parameter_values, dict): # Dict maps param positions to values for pos, value in parameter_values.items(): - pos = int(pos) if not isinstance(pos, int) else pos - if 1 <= pos <= 19: # Validate parameter positions - parameters[pos] = str(value) + param_pos = int(pos) if not isinstance(pos, int) else pos + if ( + 1 <= param_pos <= 19 + ): # Validate parameter positions # noqa: E501, PLR2004 + parameters[param_pos] = str(value) # IMPORTANT: Create a data dictionary with ALL parameters explicitly included - request_data = { - "tlx_sn": device_sn, - "type": parameter_id - } + request_data = {"tlx_sn": device_sn, "type": parameter_id} # Add all 19 parameters to the request for i in range(1, 20): request_data[f"param{i}"] = str(parameters[i]) # Send the request - response = self.session.post( - self._get_url('tlxSet'), - data=request_data - ) + response = self.session.post(self._get_url("tlxSet"), data=request_data) - return self._process_response(response.json(), f"writing parameter {parameter_id}") + return self._process_response( + response.json(), f"writing parameter {parameter_id}" + ) - def min_write_time_segment(self, device_sn, segment_id, batt_mode, start_time, end_time, enabled=True): + def min_write_time_segment( + self, device_sn: str, params: "TimeSegmentParams" + ) -> 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. + params (TimeSegmentParams): Grouped parameters for the time segment. Returns: dict: The server response. @@ -533,42 +1171,280 @@ def min_write_time_segment(self, device_sn, segment_id, batt_mode, start_time, e 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. - """ + requests.exceptions.RequestException: + If there is an issue with the HTTP request. - if not 1 <= segment_id <= 9: - raise GrowattParameterError("segment_id must be between 1 and 9") + """ + if not 1 <= params.segment_id <= 9: # noqa: PLR2004 + msg = "segment_id must be between 1 and 9" + raise GrowattParameterError(msg) - if not 0 <= batt_mode <= 2: - raise GrowattParameterError("batt_mode must be between 0 and 2") + if not 0 <= params.batt_mode <= 2: # noqa: PLR2004 + msg = "batt_mode must be between 0 and 2" + 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}" - } + all_params = {"tlx_sn": device_sn, "type": f"time_segment{params.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" + all_params["param1"] = str(params.batt_mode) + all_params["param2"] = str(params.start_time.hour) + all_params["param3"] = str(params.start_time.minute) + all_params["param4"] = str(params.end_time.hour) + all_params["param5"] = str(params.end_time.minute) + all_params["param6"] = "1" if params.enabled else "0" # Add empty strings for all unused parameters for i in range(7, 20): 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 {params.segment_id}" + ) + + """ + # MIN_TLX time segment + params = api.TimeSegmentParams( + segment_id=1, + batt_mode=1, # Battery First + start_time=time(8, 0), + end_time=time(16, 0), + enabled=True + ) + api.write_time_segment(device_sn, DeviceType.MIN_TLX, "time_segment1", params) + + # SPH_MIX discharge time + params = api.MixAcDischargeTimeParams( + discharge_power=50, + discharge_stop_soc=10, + start_hour=8, + start_minute=0, + end_hour=16, + end_minute=0, + enabled=True + ) + api.write_parameter(device_sn, DeviceType.SPH_MIX, "mix_ac_discharge_time_period", params) + + # Backflow setting + params = api.BackflowSettingParams( + backflow_enabled=True, + anti_reverse_power_percentage=50 + ) + api.write_parameter(device_sn, DeviceType.SPH_MIX, "backflow_setting", params) + """ + + def write_time_segment( + self, + device_sn: str, + device_type: DeviceType, + command: str, + params: "TimeSegmentParams | MixAcDischargeTimeParams | MixAcChargeTimeParams", + ) -> dict: + """ + Set a time segment for a MIN/TLX or SPH/MIX inverter. + + Args: + device_sn (str): The serial number of the inverter. + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + command (str): The command type to execute. + params: Parameter object appropriate for the command type. + + 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. + + """ + url_prefix = DeviceType.get_url_prefix(device_type) + + # Convert parameter object to API format based on type + if isinstance(params, self.TimeSegmentParams): + if not 1 <= params.segment_id <= 9: # noqa: PLR2004 + msg = "segment_id must be between 1 and 9" + raise GrowattParameterError(msg) + if not 0 <= params.batt_mode <= 2: # noqa: PLR2004 + msg = "batt_mode must be between 0 and 2" + raise GrowattParameterError(msg) + + api_params = self.WriteParamsInterface.time_segment_to_params(params) + command = f"time_segment{params.segment_id}" + + elif isinstance(params, self.MixAcDischargeTimeParams): + api_params = self.WriteParamsInterface.mix_discharge_to_params( + params, params.segment_id + ) + + elif isinstance(params, self.MixAcChargeTimeParams): + api_params = self.WriteParamsInterface.mix_charge_to_params( + params, params.segment_id + ) + + else: + msg = f"Unsupported parameter type: {type(params)}" + raise GrowattParameterError(msg) + + # Initialize ALL 19 parameters as empty strings + all_params = {f"{url_prefix}_sn": device_sn, "type": command} + + # Add the converted parameters + all_params.update(api_params) + + # Add empty strings for all unused parameters (1-19) + for i in range(1, 20): + param_key = f"param{i}" + if param_key not in all_params: + all_params[param_key] = "" + + if os.getenv("DEBUG", "false").lower() == "true": + with open("params.json", "w") as f: + json.dump(all_params, f, indent=4, sort_keys=True) + + # Send the request + response = self.session.post( + url=self._get_device_url(device_type, ApiDataType.SET_PARAM), + data=all_params, + ) + + return self._process_response( + response.json(), f"writing time segment command {command}" + ) + + def write_parameter( + self, device_sn: str, device_type: DeviceType, command: str, params: object + ) -> dict: + """ + Universal parameter writing method for both MIN_TLX and SPH_MIX devices. + + Args: + device_sn (str): The serial number of the device. + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + command (str): The command type to execute. + params: Parameter object appropriate for the command type. + + 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. + + Examples: + # Time segment for MIN_TLX + params = api.TimeSegmentParams( + segment_id=1, + batt_mode=1, + start_time=time(8, 0), + end_time=time(16, 0), + enabled=True + ) + api.write_parameter(device_sn, DeviceType.MIN_TLX, "time_segment1", params) + + # Discharge time for SPH_MIX + params = api.MixAcDischargeTimeParams( + discharge_power=50, + discharge_stop_soc=10, + start_hour=8, + start_minute=0, + end_hour=16, + end_minute=0, + enabled=True + ) + api.write_parameter(device_sn, DeviceType.SPH_MIX, "mix_ac_discharge_time_period", params) + + # Backflow setting + params = api.BackflowSettingParams( + backflow_enabled=True, + anti_reverse_power_percentage=50 + ) + api.write_parameter(device_sn, DeviceType.SPH_MIX, "backflow_setting", params) + + """ + url_prefix = DeviceType.get_url_prefix(device_type) + + # Convert parameter object to API format based on type + api_params = {} + + if isinstance(params, self.TimeSegmentParams): + if not 1 <= params.segment_id <= 9: # noqa: PLR2004 + msg = "segment_id must be between 1 and 9" + raise GrowattParameterError(msg) + if not 0 <= params.batt_mode <= 2: # noqa: PLR2004 + msg = "batt_mode must be between 0 and 2" + raise GrowattParameterError(msg) + api_params = self.WriteParamsInterface.time_segment_to_params(params) + + elif isinstance(params, self.MixAcDischargeTimeParams): + api_params = self.WriteParamsInterface.mix_discharge_to_params( + params, params.segment_id + ) + + elif isinstance(params, self.MixAcChargeTimeParams): + api_params = self.WriteParamsInterface.mix_charge_to_params( + params, params.segment_id + ) + + elif isinstance(params, self.BackflowSettingParams): + api_params = self.WriteParamsInterface.backflow_to_params( + params, device_type + ) + + elif isinstance(params, self.PvOnOffParams): + api_params = self.WriteParamsInterface.pv_onoff_to_params(params) + + elif isinstance(params, self.GridVoltageParams): + api_params = self.WriteParamsInterface.grid_voltage_to_params(params) + + elif isinstance(params, self.OffGridParams): + api_params = self.WriteParamsInterface.off_grid_to_params(params) + + elif isinstance(params, self.PowerParams): + api_params = self.WriteParamsInterface.power_to_params(params) + + elif isinstance(params, self.ChargeDischargeParams): + api_params = self.WriteParamsInterface.charge_discharge_to_params(params) + + else: + msg = f"Unsupported parameter type: {type(params)}" + raise GrowattParameterError(msg) + + # Initialize ALL 19 parameters as empty strings + all_params = {f"{url_prefix}_sn": device_sn, "type": command} + + # Add the converted parameters + all_params.update(api_params) + + # Add empty strings for all unused parameters (1-19) + for i in range(1, 20): + param_key = f"param{i}" + if param_key not in all_params: + all_params[param_key] = "" + + if os.getenv("DEBUG", "false").lower() == "true": + with open("params.json", "w") as f: + json.dump(all_params, f, indent=4, sort_keys=True) + # Send the request response = self.session.post( - self._get_url('tlxSet'), - data=all_params + url=self._get_device_url(device_type, ApiDataType.SET_PARAM), + data=all_params, ) - return self._process_response(response.json(), f"writing time segment {segment_id}") + return self._process_response( + response.json(), f"writing parameter command {command}" + ) - def min_read_time_segments(self, device_sn, settings_data=None): + def min_read_time_segments( + self, device_sn: str, settings_data: dict | None = None + ) -> list[dict]: """ Read Time-of-Use (TOU) settings from a Growatt MIN/TLX inverter. @@ -581,8 +1457,9 @@ def min_read_time_segments(self, device_sn, settings_data=None): 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. + 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: @@ -599,37 +1476,180 @@ def min_read_time_segments(self, device_sn, settings_data=None): # 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) + 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. - """ + 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" - } + 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") + 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 if isinstance(batt_mode, int) else -1, "Unknown" + ), + "start_time": start_time, + "end_time": end_time, + "enabled": enabled, + } + + segments.append(segment) + + return segments + + def read_time_segments( + self, device_sn: str, device_type: DeviceType, settings_data: dict | None = None + ) -> list[dict]: + """ + Read Time-of-Use (TOU) settings from a Growatt MIN/TLX or MIX/SPH inverter. + + Retrieves all 9 time segments from a Growatt MIN/TLX or MIX/SPH inverter and + parses them into a structured format. + + Note that this function uses device_settings() internally to get the + settings data. To avoid endpoint rate limit, you can pass the + settings_data parameter with the data returned from device_settings(). + + Args: + device_sn (str): The device serial number of the inverter + device_type (DeviceType): The type of device (MIN_TLX or SPH_MIX). + settings_data (dict, optional): Settings data from device_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.read_time_segments( + "DEVICE_SERIAL_NUMBER", + DeviceType.MIN_TLX + ) + # Option 2: Reuse existing settings data + settings_response = api.device_settings( + "DEVICE_SERIAL_NUMBER", + DeviceType.MIN_TLX + ) + tou_settings = api.read_time_segments( + "DEVICE_SERIAL_NUMBER", + DeviceType.MIN_TLX, + settings_response + ) + + Raises: + GrowattParameterError: If device type is invalid. + GrowattV1ApiError: If the API request fails + requests.exceptions.RequestException: + If there is an issue with the HTTP request. + + """ + # Select appropriate templates + if device_type == DeviceType.MIN_TLX: + templates = DeviceFieldTemplates.MIN_TLX_TEMPLATES + elif device_type == DeviceType.SPH_MIX: + templates = DeviceFieldTemplates.SPH_MIX_TEMPLATES_CHARGE + else: + msg = f"Unsupported device type: {device_type}" + raise GrowattParameterError(msg) + + # Process the settings data + if settings_data is None: + # Fetch settings if not provided + settings_data = self.device_settings(device_sn, device_type=device_type) + + # 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 + # Format field names using templates + start_field = templates["start_time"].format(segment_id=i) + stop_field = templates["stop_time"].format(segment_id=i) + mode_field = templates["mode"].format(segment_id=i) + enabled_field = templates["enabled"].format(segment_id=i) + + # Get values using formatted field names + start_time_raw = settings_data.get(start_field, "0:0") + end_time_raw = settings_data.get(stop_field, "0:0") + mode_raw = settings_data.get(mode_field) + enabled_raw = settings_data.get(enabled_field, 0) # Handle 'null' string values - if start_time_raw == 'null' or not start_time_raw: + 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: + if end_time_raw == "null" or not end_time_raw: end_time_raw = "0:0" # Format times with leading zeros (HH:MM) @@ -650,8 +1670,7 @@ def min_read_time_segments(self, device_sn, settings_data=None): 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: + if mode_raw == "null" or mode_raw is None: batt_mode = None else: try: @@ -660,8 +1679,7 @@ def min_read_time_segments(self, device_sn, settings_data=None): 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: + if enabled_raw == "null" or enabled_raw is None: enabled = False else: try: @@ -670,14 +1688,104 @@ def min_read_time_segments(self, device_sn, settings_data=None): 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 + "segment_id": i, + "batt_mode": batt_mode, + "mode_name": mode_names.get( + batt_mode if batt_mode is not None else -1, "Unknown" + ), + "start_time": start_time, + "end_time": end_time, + "enabled": enabled, } segments.append(segment) return segments + + def get_devices(self, plant_id: int) -> list["GrowattDevice"]: + """Get devices as GrowattDevice objects for easier use.""" + data = self.device_list(plant_id) + return [GrowattDevice(self, device) for device in data["devices"]] + + +class GrowattDevice: + """Represents a Growatt device with automatic type handling.""" + + def __init__(self, api: OpenApiV1, device_data: dict) -> None: + """ + Initialize a GrowattDevice instance. + + Args: + api: The API client instance. + device_data: Dictionary containing device information. + + """ + self._api = api + self.device_sn = device_data["device_sn"] + self.device_type = device_data["device_type"] + self.model = device_data.get("model") + self.status = device_data.get("status") + # Store other device metadata... + + def details(self) -> dict: + """Get detailed device data.""" + return self._api.device_details(self.device_sn, self.device_type) + + def energy(self) -> dict: + """Get current energy data.""" + return self._api.device_energy(self.device_sn, self.device_type) + + def settings(self) -> dict: + """Get device settings.""" + return self._api.device_settings(self.device_sn, self.device_type) + + def energy_history( + self, + start_date: date | None = None, + end_date: date | None = None, + timezone: str | None = None, + page: int | None = None, + limit: int | None = None, + ) -> dict: + """Get energy history data.""" + params = self._api.DeviceEnergyHistoryParams( + start_date=start_date, + end_date=end_date, + timezone=timezone, + page=page, + limit=limit, + ) + return self._api.device_energy_history(self.device_sn, self.device_type, params) + + def read_time_segments(self, settings_data: dict | None = None) -> list[dict]: + """Read TOU time segments.""" + return self._api.read_time_segments( + self.device_sn, self.device_type, settings_data + ) + + def read_parameter( + self, + parameter_id: str, + start_address: int | None = None, + end_address: int | None = None, + ) -> dict: + """Read a parameter from the device.""" + return self._api.common_read_parameter( + self.device_sn, self.device_type, parameter_id, start_address, end_address + ) + + def common_write_time_segment( + self, + command: str, + params: "TimeSegmentParams | MixAcDischargeTimeParams | MixAcChargeTimeParams", + ) -> dict: + """Write a time segment parameter to the device.""" + return self._api.write_time_segment( + self.device_sn, self.device_type, command, params + ) + + def common_write_parameter(self, command: str, params: object) -> dict: + """Write a parameter to the device.""" + return self._api.write_parameter( + self.device_sn, self.device_type, command, params + )