84c4922bbc
FastAPI app that pulls interval usage from Sungrow iSolarCloud (OAuth2), caches AER Consumer Data Right electricity plan data, and compares actual metered usage against every available plan to estimate real annual cost (TOU, tiered feed-in, and flat tariffs). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
"""Manual verification CLI for the real iSolarCloud client (both auth modes).
|
|
|
|
USE_MOCK_ISOLARCLOUD=false and ISOLARCLOUD_AUTH_MODE ("v1" or "oauth2") must
|
|
be set in .env along with the matching credentials -- see SETUP.md.
|
|
|
|
Usage (v1):
|
|
python scripts/verify_isolarcloud_cli.py login
|
|
python scripts/verify_isolarcloud_cli.py plants
|
|
python scripts/verify_isolarcloud_cli.py devices --ps-id 1234567
|
|
python scripts/verify_isolarcloud_cli.py history --hours 6
|
|
|
|
Usage (oauth2):
|
|
python scripts/verify_isolarcloud_cli.py authorize-url
|
|
# open the printed URL yourself, approve access, then copy the `code`
|
|
# param from the (broken, expected) redirect URL:
|
|
python scripts/verify_isolarcloud_cli.py exchange --code THE_CODE
|
|
python scripts/verify_isolarcloud_cli.py plants
|
|
python scripts/verify_isolarcloud_cli.py history --hours 6
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.config import settings
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"action", choices=["login", "authorize-url", "exchange", "plants", "devices", "history"]
|
|
)
|
|
parser.add_argument("--code", help="oauth2: the 'code' query param from the OAuth redirect")
|
|
parser.add_argument("--ps-id", help="plant id; auto-detected (first plant) if omitted")
|
|
parser.add_argument("--hours", type=int, default=3, help="for 'history': how far back to fetch")
|
|
args = parser.parse_args()
|
|
|
|
if settings.isolarcloud_auth_mode == "oauth2":
|
|
from app.isolarcloud.oauth_client import ISolarCloudOAuthClient
|
|
|
|
client = ISolarCloudOAuthClient()
|
|
if args.action == "authorize-url":
|
|
print(client.authorize_url())
|
|
return
|
|
if args.action == "exchange":
|
|
if not args.code:
|
|
parser.error("--code is required for 'exchange'")
|
|
auth_ps_list = client.exchange_code(args.code)
|
|
print("Tokens stored. Authorized plants:", auth_ps_list)
|
|
return
|
|
else:
|
|
from app.isolarcloud.client import DEVICE_TYPE_INVERTER, DEVICE_TYPE_METER, ISolarCloudClient
|
|
|
|
client = ISolarCloudClient()
|
|
if args.action == "login":
|
|
client.login()
|
|
print("Login succeeded, token acquired and cached.")
|
|
return
|
|
if args.action in ("authorize-url", "exchange"):
|
|
parser.error(f"'{args.action}' is only valid with ISOLARCLOUD_AUTH_MODE=oauth2")
|
|
|
|
if args.action == "plants":
|
|
print(json.dumps(client.get_plant_list(), indent=2))
|
|
return
|
|
|
|
ps_id = args.ps_id
|
|
if not ps_id:
|
|
plants = client.get_plant_list()
|
|
if not plants:
|
|
print("No plants visible to this account.")
|
|
return
|
|
ps_id = str(plants[0]["ps_id"])
|
|
print(f"(using first plant: {ps_id})")
|
|
|
|
if args.action == "devices":
|
|
if settings.isolarcloud_auth_mode == "oauth2":
|
|
print(json.dumps(client.get_device_list(ps_id), indent=2))
|
|
return
|
|
devices = client.get_device_list(ps_id)
|
|
print(json.dumps(devices, indent=2))
|
|
meter = next((d for d in devices if int(d.get("device_type", -1)) == DEVICE_TYPE_METER), None)
|
|
inverter = next((d for d in devices if int(d.get("device_type", -1)) == DEVICE_TYPE_INVERTER), None)
|
|
print()
|
|
print(f"Detected meter device: {meter.get('ps_key') if meter else 'NONE FOUND'}")
|
|
print(f"Detected inverter device: {inverter.get('ps_key') if inverter else 'NONE FOUND'}")
|
|
return
|
|
|
|
if args.action == "history":
|
|
until = datetime.now(timezone.utc)
|
|
since = until - timedelta(hours=args.hours)
|
|
points = client.fetch_interval_data(since, until)
|
|
for p in points[:20]:
|
|
print(p)
|
|
print(f"... {len(points)} points total")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|