SungrowAPIV1/sungrow_api.py

import os
import time
import uuid
import requests
import json
from dotenv import load_dotenv
from sungrow_state import load_ps_id



BASE_URL = os.getenv("SUNGROW_BASE_URL")
USERNAME = os.getenv("SUNGROW_USERNAME")
PASSWORD = os.getenv("SUNGROW_PASSWORD")
APPKEY = os.getenv("SUNGROW_APPKEY")
KEY = os.getenv("SUNGROW_KEY")


def login():
    url = f"{BASE_URL}/openapi/login"

    headers = {
        "Content-Type": "application/json;charset=UTF-8",
        "sys_code": "901",
        "x-access-key": KEY
    }

    body = {
        "user_account": USERNAME,
        "user_password": PASSWORD,
        "appkey": APPKEY,
        "lang": "_de_DE"
    }

    response = requests.post(url, json=body, headers=headers, timeout=10)

    print("Login Status:", response.status_code)
    print("Login Body:", response.text)

    response.raise_for_status()

    data = response.json()

    if data.get("result_code") != "1":
        raise Exception(f"Login failed: {data}")

    return data["result_data"]["token"]


def call_api(endpoint: str, token: str, params: dict = None, debug: bool = False):
    if params is None:
        params = {}

    # Auto-inject ps_id if requested
    if "ps_id" in params and params["ps_id"] is None:
        stored_ps_id = load_ps_id()
        if stored_ps_id is None:
            raise ValueError("ps_id was requested but no stored ps_id found.")
        params["ps_id"] = stored_ps_id

    url = f"{BASE_URL}{endpoint}"

    headers = {
        "Content-Type": "application/json;charset=UTF-8",
        "sys_code": "901",
        "x-access-key": KEY
    }

    base_body = {
        "appkey": APPKEY,
        "token": token,
        "lang": "_de_DE",
        "timestamp": str(int(time.time() * 1000)),
        "nonce": uuid.uuid4().hex
    }

    body = {**base_body, **params}

    # Debug: print full request
    if debug:
        import json
        print("\n=== API REQUEST ===")
        print("URL:", url)
        print("Headers:", json.dumps(headers, indent=4, ensure_ascii=False))
        print("Body:", json.dumps(body, indent=4, ensure_ascii=False))
        print("===================\n")

    response = requests.post(url, json=body, headers=headers, timeout=10)

    # Debug: print full response
    if debug:
        print("=== API RESPONSE ===")
        print("Status:", response.status_code)
        print("Body:", response.text)
        print("====================\n")

    response.raise_for_status()
    return response.json()


def get_power_station_list(token: str, cur_page: int = 1, size: int = 10, debug: bool = False):
    """
    Wrapper for /openapi/getPowerStationList
    """
    params = {
        "curPage": cur_page,
        "size": size
    }

    return call_api("/openapi/getPowerStationList", token, params, debug=debug)

def get_device_list(
    token: str,
    ps_id: int,
    cur_page: int = 1,
    size: int = 50,
    is_virtual_unit: str | None = None,
    device_type_list: list | None = None,
    rel_state: str | None = None,
    is_get_firmware_version: str | None = None,
    debug: bool = False
):
    """
    Wrapper for /openapi/getDeviceList

    Parameters:
        ps_id (int): Plant ID (required)
        cur_page (int): Page number (required)
        size (int): Page size (required)
        is_virtual_unit (str): "1" = virtual, "0" = physical
        device_type_list (list): e.g. [1, 3, 11]
        rel_state (str): "0" = unclaimed, "1" = claimed
        is_get_firmware_version (str): "0" = no, "1" = yes
    """

    params = {
        "ps_id": ps_id,
        "curPage": cur_page,
        "size": size
    }

    # Only include optional parameters if they are provided
    if is_virtual_unit is not None:
        params["is_virtual_unit"] = is_virtual_unit

    if device_type_list is not None:
        params["device_type_list"] = device_type_list

    if rel_state is not None:
        params["rel_state"] = rel_state

    if is_get_firmware_version is not None:
        params["is_get_firmware_version"] = is_get_firmware_version

    return call_api("/openapi/getDeviceList", token, params, debug=debug)

def get_device_realtime_data(
    token: str,
    point_id_list: list,
    device_type: int,
    ps_key_list: list | None = None,
    sn_list: list | None = None,
    debug: bool = False
):
    """
    Wrapper for /openapi/getDeviceRealTimeData

    Required:
        point_id_list (list[str])
        device_type (int)

    Optional (choose one):
        ps_key_list (list[str])
        sn_list (list[str])
    """

    if ps_key_list is None and sn_list is None:
        raise ValueError("You must provide either ps_key_list or sn_list.")

    params = {
        "point_id_list": point_id_list,
        "device_type": device_type
    }

    if ps_key_list is not None:
        params["ps_key_list"] = ps_key_list

    if sn_list is not None:
        params["sn_list"] = sn_list

    return call_api("/openapi/getDeviceRealTimeData", token, params, debug=debug)


def ensure_success(response: dict):
    if response.get("result_code") != "1":
        raise Exception(
            f"API error {response.get('result_code')}: {response.get('result_msg')}"
        )
        
        
def pretty_print(data):
    import json
    print(json.dumps(data, indent=4, ensure_ascii=False))