From 741945a62d41c04defb937ed9474dd2b49621e78 Mon Sep 17 00:00:00 2001 From: Jason Cox Date: Sun, 17 Dec 2023 22:52:20 -0800 Subject: [PATCH] Add cloud setup to CLI --- .gitignore | 2 +- pypowerwall/__init__.py | 18 ++++++++- pypowerwall/__main__.py | 54 ++++++++++++++++++++++++- pypowerwall/cloud.py | 88 ++++++++++++++++++++++++++--------------- 4 files changed, 125 insertions(+), 37 deletions(-) diff --git a/.gitignore b/.gitignore index accff7f..91d9478 100644 --- a/.gitignore +++ b/.gitignore @@ -144,4 +144,4 @@ tools/set-mode.auth tools/set-mode.conf tools/tedapi/request.bin tools/tedapi/app* -pypowerwall/.pypowerwall.auth +.pypowerwall.auth diff --git a/pypowerwall/__init__.py b/pypowerwall/__init__.py index 597075c..41f1a53 100644 --- a/pypowerwall/__init__.py +++ b/pypowerwall/__init__.py @@ -58,8 +58,9 @@ import logging import sys from . import tesla_pb2 # Protobuf definition for vitals +from . import cloud # Tesla Cloud API -version_tuple = (0, 6, 4) +version_tuple = (0, 7, 0) version = __version__ = '%d.%d.%d' % version_tuple __author__ = 'jasonacox' @@ -86,7 +87,7 @@ class ConnectionError(Exception): pass class Powerwall(object): - def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="America/Los_Angeles", pwcacheexpire=5, timeout=10, poolmaxsize=10): + def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="America/Los_Angeles", pwcacheexpire=5, timeout=10, poolmaxsize=10, cloudmode=False): """ Represents a Tesla Energy Gateway Powerwall device. @@ -99,6 +100,7 @@ def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="A pwcacheexpire = Seconds to expire cached entries timeout = Seconds for the timeout on http requests poolmaxsize = Pool max size for http connection re-use (persistent connections disabled if zero) + cloudmode = If True, use Tesla cloud for connection (default is False) """ @@ -114,6 +116,18 @@ def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="A self.pwcachetime = {} # holds the cached data timestamps for api self.pwcache = {} # holds the cached data for api self.pwcacheexpire = pwcacheexpire # seconds to expire cache + self.cloudmode = cloudmode # cloud mode (default) or local mode + self.Tesla = False # cloud object for cloud connection + + # Check for cloud mode + if self.cloudmode or self.host == "": + log.debug('Tesla cloud mode enabled') + self.Tesla = cloud.TeslaCloud(self.email, pwcacheexpire, timeout) + # Check to see if we can connect to the cloud + if not self.Tesla.connect(): + err = "Unable to connect to Tesla Cloud - run pypowerwall setup" + log.debug(err) + raise ConnectionError(err) if self.poolmaxsize > 0: # Create session object for http connection re-use diff --git a/pypowerwall/__main__.py b/pypowerwall/__main__.py index 9448416..f520983 100644 --- a/pypowerwall/__main__.py +++ b/pypowerwall/__main__.py @@ -14,9 +14,13 @@ # Modules import pypowerwall import sys +import os +import json from . import scan +from . import cloud # Global Variables +AUTHFILE = ".pypowerwall.auth" timeout = 1.0 state = 0 color = True @@ -26,6 +30,8 @@ continue elif(i.lower() == "scan"): state = 0 + elif(i.lower() == "setup"): + state = 1 elif(i.lower() == "-nocolor"): color = False else: @@ -38,9 +44,52 @@ if(state == 0): scan.scan(color, timeout) -# State 1 = Future +# State 1 = Cloud Mode Setup if(state == 1): - print("Future Feature") + print("pyPowerwall [%s]\n" % (pypowerwall.version)) + print("Cloud Mode Setup\n") + + # Check for existing auth file + if os.path.isfile(AUTHFILE): + with open(AUTHFILE) as json_file: + try: + data = json.load(json_file) + tuser = list(data.keys())[0] + except Exception as err: + tuser = None + # Ask to overwrite + print(f"Found {AUTHFILE} configuration file for {tuser}") + answer = input("Overwrite and run setup? (y/n) ") + if answer.lower() == "y": + os.remove(AUTHFILE) + else: + print("Exiting") + exit(0) + + # Run Setup + c = cloud.TeslaCloud(None) + c.setup() + tuser = c.email + + # Test Connection + print("Testing connection to Tesla Cloud...") + c = cloud.TeslaCloud(tuser) + if c.connect(): + print("Connected to Tesla Cloud...") + sites = c.getsites() + print("Found %d Powerwall Sites:" % (len(sites))) + """ + "energy_site_id": 255476044283, + "resource_type": "battery", + "site_name": "Cox Energy Gateway", + """ + for s in sites: + print(" %s (%s) - Type: %s" % (s["site_name"], + s["energy_site_id"], s["resource_type"])) + print(f"\nSetup Complete. Auth file {AUTHFILE} ready to use.") + else: + print("ERROR: Failed to connect to Tesla Cloud") + exit(1) # State 2 = Show Usage if(state == 2): @@ -49,6 +98,7 @@ print(" python -m pypowerwall [command] [] [-nocolor] [-h]") print("") print(" command = scan Scan local network for Powerwall gateway.") + print(" command = setup Setup Tesla Login for Cloud Mode access.") print(" timeout Seconds to wait per host [Default=%0.1f]" % (timeout)) print(" -nocolor Disable color text output.") print(" -h Show usage.") diff --git a/pypowerwall/cloud.py b/pypowerwall/cloud.py index e9be77e..719194e 100644 --- a/pypowerwall/cloud.py +++ b/pypowerwall/cloud.py @@ -18,6 +18,7 @@ """ import sys +import os import time import logging import json @@ -622,6 +623,9 @@ def setup(self): TUSER = response break + # Update the Tesla User + self.email = TUSER + # Create retry instance for use after successful login retry = Retry(total=2, status_forcelist=(500, 502, 503, 504), backoff_factor=10) @@ -655,39 +659,59 @@ def setup(self): tesla.close() tesla = Tesla(self.email, retry=retry, cache_file=AUTHFILE) +if __name__ == "__main__": -# Test code -set_debug(False) -cloud = TeslaCloud("jason@jasonacox.com") - -if not cloud.connect(): - print("Failed to connect to Tesla Cloud") - cloud.setup() - if not cloud.connect(): - print("Failed to connect to Tesla Cloud") - exit(1) - -print("Connected to Tesla Cloud") - -#print("\nSite Data") -#sites = cloud.getsites() -#print(sites) - -#print("\Battery") -#r = cloud.get_battery() -#print(r) + # Test code + set_debug(False) + # Check for .pypowerwall.auth file + if os.path.isfile(AUTHFILE): + # Read the json file + with open(AUTHFILE) as json_file: + try: + data = json.load(json_file) + TUSER = list(data.keys())[0] + print(f"Using Tesla User: {TUSER}") + except Exception as err: + TUSER = None -#print("\Site Power") -#r = cloud.get_site_power() -#print(r) + while not TUSER: + response = input("Tesla User Email address: ").strip() + if "@" not in response: + print("Invalid email address\n") + else: + TUSER = response + break -#print("\Site Config") -#r = cloud.get_site_config() -#print(r) + cloud = TeslaCloud(TUSER) -# Test Poll -items = ['/api/status','/api/system_status/grid_status','/api/site_info/site_name','/api/devices/vitals','/api/system_status/soe','/api/meters/aggregates','/api/operation','/api/system_status'] #, '/api/logout','/api/login/Basic','/vitals','/api/meters/site','/api/meters/solar','/api/sitemaster','/api/powerwalls','/api/installer','/api/customer/registration','/api/system/update/status','/api/site_info','/api/system_status/grid_faults','/api/site_info/grid_codes','/api/solars','/api/solars/brands','/api/customer','/api/meters','/api/installer','/api/networks','/api/system/networks','/api/meters/readings','/api/synchrometer/ct_voltage_references'] -for i in items: - print(f"poll({i}):") - print(cloud.poll(i)) - print("\n") + if not cloud.connect(): + print("Failed to connect to Tesla Cloud") + cloud.setup() + if not cloud.connect(): + print("Failed to connect to Tesla Cloud") + exit(1) + + print("Connected to Tesla Cloud") + + #print("\nSite Data") + #sites = cloud.getsites() + #print(sites) + + #print("\Battery") + #r = cloud.get_battery() + #print(r) + + #print("\Site Power") + #r = cloud.get_site_power() + #print(r) + + #print("\Site Config") + #r = cloud.get_site_config() + #print(r) + + # Test Poll + items = ['/api/status','/api/system_status/grid_status','/api/site_info/site_name','/api/devices/vitals','/api/system_status/soe','/api/meters/aggregates','/api/operation','/api/system_status'] #, '/api/logout','/api/login/Basic','/vitals','/api/meters/site','/api/meters/solar','/api/sitemaster','/api/powerwalls','/api/installer','/api/customer/registration','/api/system/update/status','/api/site_info','/api/system_status/grid_faults','/api/site_info/grid_codes','/api/solars','/api/solars/brands','/api/customer','/api/meters','/api/installer','/api/networks','/api/system/networks','/api/meters/readings','/api/synchrometer/ct_voltage_references'] + for i in items: + print(f"poll({i}):") + print(cloud.poll(i)) + print("\n")