API changed?

[skapow]

Thank you for providing the script.

I was trying to run it on a windows python installation, but couldn't manage to get useful data. I'm also new to python, so please excuse, in case I am asking stupid questions.

I'm using Python 3.13.7.

I got past the authentification, but the results of
plants = await plants_api.async_get_plants()
are quite cryptic:

{'result_msg': 'success', 'access_token': 'XXX', 'refresh_token': 'XXX', 'scope': 'all', 'result_data': {'access_token': 'XXX', 'refresh_token': 'XXX', 'auth_user': XXX, 'token_type': 'bearer', 'expires_in': XXX, 'auth_ps_list': ['XXX']}, 'result_code': '1', 'token_type': 'bearer', 'expires_in': 172451, 'req_serial_num': 'XXX'}
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x000002415BCE1010>
Unclosed connector
connections: ['deque([(<aiohttp.client_proto.ResponseHandler object at 0x000002415C895970>, 8248.3243461)])']
connector: <aiohttp.connector.TCPConnector object at 0x000002415BCE1160>

I replaced the tokens and IDs here.

Could it be that Sungrow changed their API, or the newer Python version creates issues?

Any help is appreciated!

[bugjam]

Hi,

You are on the right track, I guess I should have provided a more comprehensive example.

Here's one with more details:

import asyncio
from aiohttp import ClientSession

from pysolarcloud import Auth, Server
from pysolarcloud.plants import Plants

app_key = ...
secret_key = ...
app_id = ...
redirect_uri = ...


async def main():
    # Passing session is optional but enables you to explicitly close the session after use
    session = ClientSession(raise_for_status=True)

    try:
        auth = Auth(Server.Europe, app_key, secret_key, app_id, websession=session)
        url = auth.auth_url(redirect_uri)
        print(f"Go to the following URL to authorize the application:\n{url}")
        code = input("Enter the authorization code: ")

        await auth.async_authorize(code, redirect_uri)
        plants_api = Plants(auth)
        plant_list = await plants_api.async_get_plants()
        if plant_list:
            print(f"{len(plant_list)} plants found:")
            for plant in plant_list:
                print(f"Plant ID: {plant["ps_id"]}, Name: {plant["ps_name"]}")
        else:
            print("No plants found.")
            return
        
        print("\nFetching detailed information for each plant...\n")
        plant_ids = [str(plant["ps_id"]) for plant in plant_list]
        plant_details = await plants_api.async_get_plant_details(plant_ids)
        for plant in plant_details:
            print(f"Details for Plant ID {plant["ps_id"]}: {plant}")

        print("\nFetching real-time data for each plant...\n")
        real_time_data = await plants_api.async_get_realtime_data(plant_ids)
        for plant_id, data in real_time_data.items():
            data_values = {k: v for k, v in data.items() if v and v.get("value") is not None}
            print(f"Real-time data for Plant ID {plant_id}: {data_values}")
    finally:
        await session.close()

if __name__ == "__main__":
    asyncio.run(main())

[Rawsted]

Hi i fixed some things

1. line 28 and 37 i think "ps_id" needed to be 'ps_id' also "ps_name" fixed to 'ps_name'

USE Python 3.12 !

py -m pip install --upgrade aiohttp pysolarcloud

import asyncio
from aiohttp import ClientSession

from pysolarcloud import Auth, Server
from pysolarcloud.plants import Plants

app_key = 
secret_key = 
app_id = 
redirect_uri = 

async def main():
    # Passing session is optional but enables you to explicitly close the session after use
    session = ClientSession(raise_for_status=True)

    try:
        auth = Auth(Server.Europe, app_key, secret_key, app_id, websession=session)
        url = auth.auth_url(redirect_uri)
        print(f"Go to the following URL to authorize the application:\n{url}")
        code = input("Enter the authorization code: ")

        await auth.async_authorize(code, redirect_uri)
        plants_api = Plants(auth)
        plant_list = await plants_api.async_get_plants()
        if plant_list:
            print(f"{len(plant_list)} plants found:")
            for plant in plant_list:
                print(f"Plant ID: {plant['ps_id']}, Name: {plant['ps_name']}")
        else:
            print("No plants found.")
            return
        
        print("\nFetching detailed information for each plant...\n")
        plant_ids = [str(plant["ps_id"]) for plant in plant_list]
        plant_details = await plants_api.async_get_plant_details(plant_ids)
        for plant in plant_details:
            print(f"Details for Plant ID {plant['ps_id']}: {plant}")

        print("\nFetching real-time data for each plant...\n")
        real_time_data = await plants_api.async_get_realtime_data(plant_ids)
        for plant_id, data in real_time_data.items():
            data_values = {k: v for k, v in data.items() if v and v.get("value") is not None}
            print(f"Real-time data for Plant ID {plant_id}: {data_values}")
    finally:
        await session.close()

if __name__ == "__main__":
    asyncio.run(main())