query.py

#!/usr/bin/env python3

import argparse
from dataclasses import dataclass
from datetime import date, datetime

import pandas as pd
import requests


@dataclass
class Item:
    group: str
    name: str
    unit: str


def parse_date(date_str):
    dt = datetime.strptime(date_str, "%Y-%m-%d")
    return dt.date()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("-u", "--user", required=True)
    parser.add_argument("-p", "--password", required=True)
    parser.add_argument("--date", type=parse_date, default=date.today())
    parser.add_argument("--plant")
    parser.add_argument("--inverter")
    args = parser.parse_args()

    base = "https://pv.inteless.com"
    with requests.Session() as session:
        session.headers["Accept"] = "application/json"
        with session.post(f"{base}/oauth/token", json={
            "username": args.user, "password": args.password, "grant_type": "password", "client_id": "csp-web"
        }) as resp:
            token = resp.json()["data"]["access_token"]
        session.headers["Authorization"] = "Bearer " + token

        if args.plant is None:
            with session.get(f"{base}/api/v1/plants", params={"page": 1, "limit": 10}) as resp:
                plant_id = str(resp.json()["data"]["infos"][0]["id"])
        else:
            plant_id = args.plant

        if args.inverter is None:
            with session.get(
                f"{base}/api/v1/plant/{plant_id}/inverters",
                params={"page": 1, "limit": 10, "status": -1, "type": -2}
            ) as resp:
                inverter_id = resp.json()["data"]["infos"][0]["sn"]
        else:
            inverter_id = args.inverter

        items = {}
        with session.get(
            f"{base}/api/v1/inverter/params",
            params={"lan": "en", "devType": 2, "sn": inverter_id}
        ) as resp:
            for group in resp.json()["data"]["infos"]:
                for field in group["groupContent"]:
                    items[field["id"]] = Item(group["groupName"], field["label"], field["unit"])

        date_str = args.date.strftime("%Y-%m-%d")
        series = []
        with session.get(
            f"{base}/api/v1/inverter/{inverter_id}/day",
            params={
                "sn": inverter_id,
                "date": date_str,
                "edate": date_str,
                "lan": "en",
                "params": ",".join(str(x) for x in items.keys())
            }
        ) as resp:
            for row in resp.json()["data"]["infos"]:
                item = items[row["id"]]
                timestamps = []
                values = []
                for record in row["records"]:
                    timestamps.append(datetime.strptime(record["time"], "%Y-%m-%d %H:%M:%S"))
                    values.append(float(record["value"]))
                if not all(v == 0.0 for v in values):
                    series.append(pd.Series(values, timestamps, name=f"{item.group} {item.name}"))
        df = pd.DataFrame(series)
        print(df)


if __name__ == "__main__":
    main()