diff --git a/.gitignore b/.gitignore index 21010f5..64a54aa 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +.env ### Python ### # Byte-compiled / optimized / DLL files __pycache__/ @@ -28,4 +29,4 @@ share/python-wheels/ MANIFEST # Symlink -examples/growattServer +examples/growattServer \ No newline at end of file diff --git a/examples/inverter_example.py b/examples/inverter_example.py new file mode 100644 index 0000000..0a12677 --- /dev/null +++ b/examples/inverter_example.py @@ -0,0 +1,73 @@ +import json +import os + +import requests + +import growattServer + +""" +# Example script controlling a classic Growatt inverter 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 environment variable or use test token +api_token = os.environ.get("GROWATT_API_TOKEN") +if not api_token: + # test token from official API docs https://www.showdoc.com.cn/262556420217021/1494053950115877 + api_token = "6eb6f069523055a339d71e5b1f6c88cc" # noqa: S105 + +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"] + + # Devices + devices = api.device_list(plant_id) + + for device in devices["devices"]: + if device["type"] == growattServer.DeviceType.INVERTER.value: + inverter_sn = device["device_sn"] + print(f"Processing inverter: {inverter_sn}") + device_class = api.get_device(inverter_sn, device["type"]) + + # Get device details + inverter_data = device_class.detail() + 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 = device_class.energy() + 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 = device_class.energy_history() + 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) + + # Read power rate + active_p_rate = device_class.read_parameter("pv_active_p_rate") + print("Current power rate:", active_p_rate, "%") + + # Settings parameters - Uncomment to test + + # Set active power rate +# device_class.write_parameter('pv_active_p_rate', 100) +# print("set active power rate to 100%") + +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/min_example.py b/examples/min_example.py index b1fee17..5c07ef0 100644 --- a/examples/min_example.py +++ b/examples/min_example.py @@ -1,4 +1,5 @@ import json +import os import requests @@ -9,11 +10,11 @@ # 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 +# Get the API token from environment variable or use test token +api_token = os.environ.get("GROWATT_API_TOKEN") +if not api_token: + # test token from official API docs https://www.showdoc.com.cn/262556420217021/1494053950115877 + api_token = "6eb6f069523055a339d71e5b1f6c88cc" # noqa: S105 try: # Initialize the API with token instead of using login @@ -28,7 +29,7 @@ devices = api.device_list(plant_id) for device in devices["devices"]: - if device["type"] == 7: # (MIN/TLX) + if device["type"] == growattServer.DeviceType.MIN.value: inverter_sn = device["device_sn"] print(f"Processing inverter: {inverter_sn}") diff --git a/examples/min_example_dashboard.py b/examples/min_example_dashboard.py index 8f479ff..f7d1c1d 100644 --- a/examples/min_example_dashboard.py +++ b/examples/min_example_dashboard.py @@ -1,4 +1,5 @@ import json +import os import requests @@ -9,11 +10,11 @@ using the V1 API with token-based authentication. """ -# 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 +# Get the API token from environment variable or use test token +api_token = os.environ.get("GROWATT_API_TOKEN") +if not api_token: + # test token from official API docs https://www.showdoc.com.cn/262556420217021/1494053950115877 + api_token = "6eb6f069523055a339d71e5b1f6c88cc" # noqa: S105 try: # Initialize the API with token @@ -29,7 +30,7 @@ # Iterate over all devices energy_data = None for device in devices["devices"]: - if device["type"] == 7: # (MIN/TLX) + if device["type"] == growattServer.DeviceType.MIN.value: inverter_sn = device["device_sn"] # Get energy data diff --git a/growattServer/open_api_v1/__init__.py b/growattServer/open_api_v1/__init__.py index 5cc6468..efbd61a 100644 --- a/growattServer/open_api_v1/__init__.py +++ b/growattServer/open_api_v1/__init__.py @@ -8,13 +8,13 @@ from growattServer import GrowattApi from growattServer.exceptions import GrowattV1ApiError -from .devices import AbstractDevice, Min, Sph +from .devices import AbstractDevice, Inverter, Min, Sph class DeviceType(Enum): """Enumeration of Growatt device types.""" - INVERTER = 1 + INVERTER = Inverter.DEVICE_TYPE_ID STORAGE = 2 OTHER = 3 MAX = 4 @@ -364,6 +364,8 @@ def device_list(self, plant_id): def get_device(self, device_sn: str, device_type: int) -> AbstractDevice | None: """Get the device class by serial number and device_type id.""" match device_type: + case Inverter.DEVICE_TYPE_ID: + return Inverter(self, device_sn) case Sph.DEVICE_TYPE_ID: return Sph(self, device_sn) case Min.DEVICE_TYPE_ID: @@ -375,6 +377,120 @@ def get_device(self, device_sn: str, device_type: int) -> AbstractDevice | None: ) return None + def inverter_detail(self, device_sn): + """ + Get detailed data for a classic inverter. + + Args: + device_sn (str): The serial number of the classic inverter. + + Returns: + dict: A dictionary containing the classic inverter details. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + return Inverter(self, device_sn).detail() + + def inverter_energy(self, device_sn): + """ + Get energy data for a classic inverter. + + Args: + device_sn (str): The serial number of the classic inverter. + + Returns: + dict: A dictionary containing the classic inverter energy data. + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + return Inverter(self, device_sn).energy() + + def inverter_energy_history( + self, + device_sn, + start_date=None, + end_date=None, + timezone=None, + page=None, + limit=None, + ): + """ + Get classic inverter data history. + + Args: + device_sn (str): The ID of the classic inverter. + start_date (date, optional): Start date. Defaults to today. + end_date (date, optional): End date. Defaults to today. + timezone (str, optional): Timezone ID. + page (int, optional): Page number. + limit (int, optional): Results per page. + + Returns: + dict: A dictionary containing the classic inverter history data. + + Raises: + GrowattParameterError: If date interval is invalid (exceeds 7 days). + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + return Inverter(self, device_sn).energy_history( + start_date, end_date, timezone, page, limit + ) + + def inverter_read_parameter( + self, device_sn, parameter_id, start_address=None, end_address=None + ): + """ + Read setting from classic inverter. + + Args: + device_sn (str): The ID of the TLX inverter. + parameter_id (str): Parameter ID to read. Don't use start_address and end_address if this is set. + start_address (int, optional): Register start address (for set_any_reg). Don't use parameter_id if this is set. + end_address (int, optional): Register end address (for set_any_reg). Don't use parameter_id if this is set. + + Returns: + dict: A dictionary containing the setting value. + + Raises: + GrowattParameterError: If parameters are invalid. + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + return Inverter(self, device_sn).read_parameter( + parameter_id, start_address, end_address + ) + + def inverter_write_parameter(self, device_sn, parameter_id, parameter_values=None): + """ + Set parameters on a classic inverter. + + Args: + device_sn (str): Serial number of the inverter + parameter_id (str): Setting type to be configured + parameter_values: Parameter values to be sent to the system. + Can be a single string (for param1 only), + a list of strings (for sequential params), + or a dictionary mapping param positions to values + + Returns: + dict: JSON response from the server + + Raises: + GrowattV1ApiError: If the API returns an error response. + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + """ + return Inverter(self, device_sn).write_parameter(parameter_id, parameter_values) + def min_detail(self, device_sn): """ Get detailed data for a MIN inverter. diff --git a/growattServer/open_api_v1/devices/__init__.py b/growattServer/open_api_v1/devices/__init__.py index d2c0fe0..ddba83f 100644 --- a/growattServer/open_api_v1/devices/__init__.py +++ b/growattServer/open_api_v1/devices/__init__.py @@ -1,4 +1,5 @@ # noqa: D104 from .abstract_device import AbstractDevice # noqa: F401 +from .inverter import Inverter # noqa: F401 from .min import Min # noqa: F401 from .sph import Sph # noqa: F401 diff --git a/growattServer/open_api_v1/devices/inverter.py b/growattServer/open_api_v1/devices/inverter.py new file mode 100644 index 0000000..fd15f54 --- /dev/null +++ b/growattServer/open_api_v1/devices/inverter.py @@ -0,0 +1,250 @@ +"""Inverterdevice file.""" +from datetime import UTC, datetime, timedelta + +from growattServer.exceptions import GrowattParameterError + +from .abstract_device import AbstractDevice + + +class Inverter(AbstractDevice): + """Inverter device type.""" + + DEVICE_TYPE_ID = 1 + + def detail(self) -> dict: + """ + Get detailed data for an inverter. + + Args: + device_sn (str): The serial number of the inverter. + + Returns: + dict: A dictionary containing the inverter details. + + Raises: + GrowattV1ApiError: If the API returns an error response. Endpoint-specific error codes: + 10001 - System error + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + References: + https://www.showdoc.com.cn/262556420217021/6118559963559236 + + """ + response = self.api.session.get( + self.api.get_url("device/inverter/inv_data_info"), + params={ + "device_sn": self.device_sn + } + ) + + return self.api.process_response(response.json(), "getting inverter details") + + def energy(self) -> dict: + """ + Get energy data for a inverter. + + Args: + device_sn (str): The serial number of the inverter. + + Returns: + dict: A dictionary containing the inverter energy data. + + Raises: + GrowattV1ApiError: If the API returns an error response. Endpoint-specific error codes: + 10001 - system error + 10005 - device does not exist + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + References: + https://www.showdoc.com.cn/262556420217021/6118571427302257 + + """ + response = self.api.session.get( + url=self.api.get_url("device/inverter/last_new_data"), + data={ + "device_sn": self.device_sn, + }, + ) + + return self.api.process_response(response.json(), "getting inverter energy data") + + def energy_history(self, start_date=None, end_date=None, timezone=None, page=None, limit=None) -> dict: + """ + Get inverter data history. + + Args: + device_sn (str): The ID of the inverter. + start_date (date, optional): Start date. Defaults to today. + end_date (date, optional): End date. Defaults to today. + timezone (str, optional): Timezone ID. + page (int, optional): Page number. + limit (int, optional): Results per page. + + Returns: + dict: A dictionary containing the inverter history data. + + Raises: + GrowattParameterError: If date interval is invalid (exceeds 7 days). + GrowattV1ApiError: If the API returns an error response. Endpoint-specific error codes: + 10001 - system error + 10002 - device serial number error + 10003 - date format error + 10004 - date interval exceeds seven days + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + References: + https://www.showdoc.com.cn/262556420217021/6118823163304569 + + """ + if start_date is None and end_date is None: + start_date = datetime.now(tz=UTC).astimezone().date() + end_date = datetime.now(tz=UTC).astimezone().date() + elif start_date is None: + start_date = end_date + elif end_date is None: + end_date = start_date + + # check interval validity + if end_date - start_date > timedelta(days=7): + raise GrowattParameterError("date interval must not exceed 7 days") + + response = self.api.session.get( + url=self.api.get_url("device/inverter/data"), + data={ + "device_sn": self.device_sn, + "start_date": start_date.strftime("%Y-%m-%d"), + "end_date": end_date.strftime("%Y-%m-%d"), + "timezone_id": timezone, + "page": page, + "perpage": limit, + } + ) + + return self.api.process_response(response.json(), "getting inverter energy history") + + def read_parameter(self, parameter_id, start_address=None, end_address=None) -> dict: + """ + Read setting from inverter. + + Args: + parameter_id (str): Parameter ID to read. Don't use start_address and end_address if this is set. + start_address (int, optional): Register start address (for set_any_reg). Don't use parameter_id if this is set. + end_address (int, optional): Register end address (for set_any_reg). Don't use parameter_id if this is set. + + Returns: + dict: A dictionary containing the setting value. + + Raises: + GrowattParameterError: If parameters are invalid. + GrowattV1ApiError: If the API returns an error response. Endpoint-specific error codes: + 10001 - Reading failed + 10002 - Device does not exist + 10003 - Device offline + 10004 - Collector serial number is empty + 10005 - Collector offline + 10006 - Collector type does not support reading Get function + 10007 - The collector version does not support the reading function + 10008 - The collector connects to the server error, please restart and try again + 10009 - The read setting parameter type does not exist + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + References: + https://www.showdoc.com.cn/262556420217021/6119495760116670 + + """ + self.validate_read_parameter_input(parameter_id, start_address, end_address) + + if parameter_id is not None: + # named parameter + start_address = 0 + end_address = 0 + else: + # using register-number mode + parameter_id = "set_any_reg" + if start_address is None: + start_address = end_address + if end_address is None: + end_address = start_address + + response = self.api.session.post( + self.api.get_url("readInverterParam"), + data={ + "device_sn": self.device_sn, + "paramId": parameter_id, + "startAddr": start_address, + "endAddr": end_address, + } + ) + + return self.api.process_response(response.json(), f"reading parameter {parameter_id}") + + def write_parameter(self, parameter_id, parameter_values=None) -> dict: + """ + Set parameters on a inverter. + + Args: + device_sn (str): Serial number of the inverter + parameter_id (str): Setting type to be configured + parameter_values: Parameter values to be sent to the system. + Can be a single string (for param1 only), + a list of strings (for sequential params), + or a dictionary mapping param positions to values + + Returns: + dict: JSON response from the server + + Raises: + GrowattV1ApiError: If the API returns an error response. Endpoint-specific error codes: + 10001 - system error + 10002 - inverter server error + 10003 - inverter offline + 10004 - collector serial number is empty + 10005 - collector offline + 10006 - set The parameter type does not exist + 10007 - the parameter value is empty + 10008 - the parameter value is not in the range + 10009 - the date and time format is wrong + requests.exceptions.RequestException: If there is an issue with the HTTP request. + + References: + https://www.showdoc.com.cn/262556420217021/6118532122241417 + + """ + # Initialize all parameters as empty strings + max_inv_params = 2 + parameters = dict.fromkeys(range(1, max_inv_params + 1), "") + + # Process parameter values based on type + if parameter_values is not None: + if isinstance(parameter_values, (str, int, float, bool)): + # Single value goes to param1 + parameters[1] = str(parameter_values) + elif isinstance(parameter_values, list): + # List of values go to sequential params + for i, value in enumerate(parameter_values, 1): + if i <= max_inv_params: # Only use up to max_inv_params parameters + parameters[i] = str(value) + elif isinstance(parameter_values, dict): + # Dict maps param positions to values + for pos_raw, value in parameter_values.items(): + pos = int(pos_raw) if not isinstance(pos_raw, int) else pos_raw + if 1 <= pos <= max_inv_params: # Validate parameter positions + parameters[pos] = str(value) + + # IMPORTANT: Create a data dictionary with ALL parameters explicitly included + request_data = { + "device_sn": self.device_sn, + "paramId": parameter_id + } + + # Add all Inverter parameters to the request + for i in range(1, max_inv_params + 1): + request_data[f"command_{i}"] = str(parameters[i]) + + # Send the request + response = self.api.session.post( + self.api.get_url("inverterSet"), + data=request_data + ) + + return self.api.process_response(response.json(), f"writing parameter {parameter_id}")