diff --git a/README.md b/README.md index 191becf..7b2df8b 100644 --- a/README.md +++ b/README.md @@ -17,15 +17,11 @@ poll API endpoints on the Gateway. pyPowerwall will cache the authentication headers and API call responses to help reduce the number of calls made to the Gateway (useful if you are polling the Powerwall frequently for trending data). * Works with Tesla Energy Gateways - Powerwall and Powerwall+ -* Simple access through easy to use functions using customer credentials +* Access provided via Local Gateway API, Tesla FleetAPI (official), and Tesla Owners API (unofficial). * Will cache authentication to reduce load on Powerwall Gateway -* Will cache responses to limit number of calls to Powerwall Gateway (optional/user definable) +* Will cache responses to limit number of calls to Powerwall Gateway or Cloud (optional/user definable) * Will re-use http connections to Powerwall Gateway for reduced load and faster response times -* Easy access to decoded binary device vitals (/api/devices/vitals in JSON format) -* Provides solar string data for Powerwall+ systems - -NOTE: This module requires that you (or your installer) have set up *Customer Login* credentials -on your Powerwall Gateway. +* Provides solar string data for Powerwall+ systems. ## Setup @@ -35,13 +31,41 @@ You can clone this repo or install the package with pip. Once installed, pyPowe # Install pyPowerwall python3 -m pip install pypowerwall -# Scan Network for Powerwalls +# Option 1 - LOCAL MODE - Scan Network for Powerwalls python3 -m pypowerwall scan -# (optional) Setup to use Tesla Owners cloud API +# Option 2 - FLEETAPI MODE - Setup to use the official Tesla FleetAPI - See notes below. +python3 -m pypowerwal fleetapi + +# Option 3 - CLOUD MODE - Setup to use Tesla Owners cloud API python3 -m pypowerwall setup ``` +### Local Setup - Option 1 + +The Tesla Powerwall, Powerwall 2 and Powerwall+ have a local LAN based API that you can use to monitor your Powerwall. It requires that you (or your installer) have the IP address (see scan above) and set up *Customer Login* credentials on your Powerwall Gateway. That is all that is needed to connect. Unfortunately, Powerwall 3 does not have a local API but you can access it via the cloud (options 2 and 3). + +### FleetAPI Setup - Option 2 + +FleetAPI is the official Tesla API for accessing your Tesla products. This setup has some additional setup requirements that you will be prompted to do: + +Step 1 - Tesla Partner Account - Sign in to Tesla Developer Portal and make an App Access Request: See [Tesla App Access Request](https://developer.tesla.com/request) - During this process, you will need to set up and remember the following account settings: + + * CLIENT_ID - This will be provided to you by Tesla when your request is approved. + * CLIENT_SECRET - Same as above. + * DOMAIN - The domain name of a website your own and control. + * REDIRECT_URI - This is the URL that Tesla will direct you to after you authenticate. This landing URL (on your website) will extract the GET variable `code`, which is a one-time use authorization code needed during the pyPowerwall setup. You can use [index.html](./tools/fleetapi/index.html) on your site and update REDIRECT_URI with that url. Alternatively, you can just copy the URL from the 404 page during the authorization process (the code is in the URL). + +Step 2 - Run the [create_pem_key.py](./tools/fleetapi/create_pem_key.py) script and place the **public** key on your website at the URL: https://{DOMAIN}/.well-known/appspecific/com.tesla.3p.public-key.pem + +Step 3 - Run `python3 -m pypowerwal fleetapi` - The credentials and tokens will be stored in the `.pypowerwall.fleetapi` file. + +### Cloud Mode - Option 3 + +The unofficial Tesla Owners API allows FleetAPI access (option 2) without having to set up a website and PEM key. Follow the directions given to you by running `python3 -m pypowerwall setup`. The credentials and site_id will be stored in `.pypowerwall.auth` and `.pypowerwall.site`. + +### FreeBSD Install + FreeBSD users can install from ports or pkg [FreshPorts](https://www.freshports.org/net-mgmt/py-pypowerwall): Via pkg: @@ -67,20 +91,23 @@ and call function to poll data. Here is an example: # Optional: Turn on Debug Mode # pypowerwall.set_debug(True) - # Local Mode - Credentials for your Powerwall - Customer Login - password='password' - email='email@example.com' + # Option 1 - LOCAL MODE - Credentials for your Powerwall - Customer Login + password="password" + email="email@example.com" host = "10.0.1.123" # Address of your Powerwall Gateway timezone = "America/Los_Angeles" # Your local timezone - # (Optional) Cloud Mode - Requires Setup - password = "" + # Option 2 - FLEETAPI MODE - Requires Setup + host = password = email = "" + timezone = "America/Los_Angeles" + + # Option 3 - CLOUD MODE - Requires Setup + host = password = "" email='email@example.com' - host = "" - timezone = "America/Los_Angeles" # Your local timezone + timezone = "America/Los_Angeles" - # Connect to Powerwall - pw = pypowerwall.Powerwall(host,password,email,timezone) + # Connect to Powerwall - auto_select mode (local, fleetapi, cloud) + pw = pypowerwall.Powerwall(host,password,email,timezone,auto_select=True) # Some System Info print("Site Name: %s - Firmware: %s - DIN: %s" % (pw.site_name(), pw.version(), pw.din())) @@ -117,12 +144,13 @@ and call function to poll data. Here is an example: ``` ### pyPowerwall Module Class and Functions + ``` set_debug(True, color=True) Classes Powerwall(host, password, email, timezone, pwcacheexpire, timeout, poolmaxsize, - cloudmode, siteid, authpath, authmode, cachefile) + cloudmode, siteid, authpath, authmode, cachefile, fleetapi, auto_select) Parameters host # Hostname or IP of the Tesla gateway @@ -134,10 +162,12 @@ and call function to poll data. Here is an example: poolmaxsize = 10 # Pool max size for http connection re-use (persistent connections disabled if zero) cloudmode = False # If True, use Tesla cloud for data (default is False) - siteid # If cloudmode is True, use this siteid (default is None) - authpath # Path to cloud auth and site cache files (default is "") + siteid = None # If cloudmode is True, use this siteid (default is None) + authpath = "" # Path to cloud auth and site files (default current directory) authmode = "cookie" # "cookie" (default) or "token" - use cookie or bearer token for auth cachefile = ".powerwall" # Path to cache file (default current directory) + fleetapi = False # If True, use Tesla FleetAPI for data (default is False) + auto_select = False # If True, select the best available mode to connect (default is False) Functions poll(api, json, force) # Return data from Powerwall api (dict if json=True, bypass cache force=True) @@ -160,21 +190,17 @@ and call function to poll data. Here is an example: temps() # Return Powerwall Temperatures alerts() # Return array of Alerts from devices system_status(json) # Returns the system status - battery_blocks(json) # Returns battery specific information merged from - # system_status() and vitals() - grid_status(type) # Return the power grid status, type ="string" (default), - # "json", or "numeric": + battery_blocks(json) # Returns battery specific information merged from system_status() and vitals() + grid_status(type) # Return the power grid status, type ="string" (default), "json", or "numeric" # - "string": "UP", "DOWN", "SYNCING" # - "numeric": -1 (Syncing), 0 (DOWN), 1 (UP) - is_connected() # Returns True if able to connect to Powerwall + is_connected() # Returns True if able to connect and login to Powerwall get_reserve(scale) # Get Battery Reserve Percentage get_mode() # Get Current Battery Operation Mode - set_reserve(level) # Set Battery Reserve Percentage (only cloud mode) - set_mode(mode) # Set Current Battery Operation Mode (only cloud mode) + set_reserve(level) # Set Battery Reserve Percentage + set_mode(mode) # Set Current Battery Operation Mode get_time_remaining() # Get the backup time remaining on the battery - - set_operation(level, mode, json) # Set Battery Reserve Percentage and/or Operation Mode - + set_operation(level, mode, json) # Set Battery Reserve Percentage and/or Operation Mode ``` ## Tools diff --git a/proxy/RELEASE.md b/proxy/RELEASE.md index d7f32fd..01e6068 100644 --- a/proxy/RELEASE.md +++ b/proxy/RELEASE.md @@ -1,5 +1,13 @@ ## pyPowerwall Proxy Release Notes +### Proxy t57 (15 May 2024) + +* Add pypowerwall v0.9.0 capabilities, specifically supporting Tesla FleetAPI for cloud connections (main data and control). + +### Proxy t56 (14 May 2024) + +* Fix error with site_name on Solar Only systems. + ### Proxy t55 (4 May 2024) * Fix `/pod` API to add `time_remaining_hours` and `backup_reserve_percent` for cloud mode. diff --git a/proxy/requirements.txt b/proxy/requirements.txt index 11e9898..82b086f 100644 --- a/proxy/requirements.txt +++ b/proxy/requirements.txt @@ -1,2 +1,2 @@ -pypowerwall==0.8.5 +pypowerwall==0.9.0 bs4==0.0.2 diff --git a/proxy/server.py b/proxy/server.py index 2a9911c..b6d5ad1 100755 --- a/proxy/server.py +++ b/proxy/server.py @@ -51,10 +51,11 @@ import pypowerwall from pypowerwall import parse_version +from pypowerwall.fleetapi.fleetapi import CONFIGFILE from transform import get_static, inject_js from urllib.parse import urlparse, parse_qs -BUILD = "t56" +BUILD = "t57" ALLOWLIST = [ '/api/status', '/api/site_info/site_name', '/api/meters/site', '/api/meters/solar', '/api/sitemaster', '/api/powerwalls', @@ -97,6 +98,7 @@ # Global Stats proxystats = { 'pypowerwall': "%s Proxy %s" % (pypowerwall.version, BUILD), + 'mode': "Unknown", 'gets': 0, 'posts': 0, 'errors': 0, @@ -109,6 +111,7 @@ 'mem': 0, 'site_name': "", 'cloudmode': False, + 'fleetapi': False, 'siteid': None, 'counter': 0 } @@ -166,7 +169,8 @@ def get_value(a, key): pw = pypowerwall.Powerwall(host, password, email, timezone, cache_expire, timeout, pool_maxsize, siteid=siteid, authpath=authpath, authmode=authmode, - cachefile=cachefile) + cachefile=cachefile, auto_select=True, + retry_modes=True) except Exception as e: log.error(e) log.error("Fatal Error: Unable to connect. Please fix config and restart.") @@ -175,9 +179,15 @@ def get_value(a, key): time.sleep(5) # Infinite loop to keep container running except (KeyboardInterrupt, SystemExit): sys.exit(0) + site_name = pw.site_name() or "Unknown" -if pw.cloudmode: - log.info("pyPowerwall Proxy Server - Cloud Mode") +if pw.cloudmode or pw.fleetapi: + if pw.fleetapi: + proxystats['mode'] = "FleetAPI" + log.info("pyPowerwall Proxy Server - FleetAPI Mode") + else: + proxystats['mode'] = "Cloud" + log.info("pyPowerwall Proxy Server - Cloud Mode") log.info("Connected to Site ID %s (%s)" % (pw.client.siteid, site_name.strip())) if siteid is not None and siteid != str(pw.client.siteid): log.info("Switch to Site ID %s" % siteid) @@ -189,6 +199,7 @@ def get_value(a, key): except (KeyboardInterrupt, SystemExit): sys.exit(0) else: + proxystats['mode'] = "Local" log.info("pyPowerwall Proxy Server - Local Mode") log.info("Connected to Energy Gateway %s (%s)" % (host, site_name.strip())) @@ -196,17 +207,17 @@ def get_value(a, key): if control_secret: log.info("Control Commands Activating - WARNING: Use with caution!") try: - if pw.cloudmode: + if pw.cloudmode or pw.fleetapi: pw_control = pw else: pw_control = pypowerwall.Powerwall("", password, email, siteid=siteid, authpath=authpath, authmode=authmode, - cachefile=cachefile) + cachefile=cachefile, auto_select=True) except Exception as e: log.error("Control Mode Failed: Unable to connect to cloud - Run Setup") control_secret = "" if pw_control: - log.info("Control Mode Enabled: Cloud Mode Connected") + log.info(f"Control Mode Enabled: Cloud Mode ({pw_control.mode}) Connected") else: log.error("Control Mode Failed: Unable to connect to cloud - Run Setup") control_secret = None @@ -335,7 +346,8 @@ def do_GET(self): proxystats['mem'] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss proxystats['site_name'] = pw.site_name() proxystats['cloudmode'] = pw.cloudmode - if pw.cloudmode and pw.client is not None: + proxystats['fleetapi'] = pw.fleetapi + if (pw.cloudmode or pw.fleetapi) and pw.client is not None: proxystats['siteid'] = pw.client.siteid proxystats['counter'] = pw.client.counter proxystats['authmode'] = pw.authmode @@ -505,7 +517,8 @@ def do_GET(self): proxystats['mem'] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss proxystats['site_name'] = pw.site_name() proxystats['cloudmode'] = pw.cloudmode - if pw.cloudmode and pw.client is not None: + proxystats['fleetapi'] = pw.fleetapi + if (pw.cloudmode or pw.fleetapi) and pw.client is not None: proxystats['siteid'] = pw.client.siteid proxystats['counter'] = pw.client.counter proxystats['authmode'] = pw.authmode @@ -574,7 +587,7 @@ def do_GET(self): if fcontent: log.debug("Served from local web root: {} type {}".format(self.path, ftype)) # If not found, serve from Powerwall web server - elif pw.cloudmode: + elif pw.cloudmode or pw.fleetapi: log.debug("Cloud Mode - File not found: {}".format(self.path)) fcontent = bytes("Not Found", 'utf-8') ftype = "text/plain" diff --git a/pypowerwall/__init__.py b/pypowerwall/__init__.py index 41d99db..56ccd52 100644 --- a/pypowerwall/__init__.py +++ b/pypowerwall/__init__.py @@ -33,6 +33,9 @@ authpath = "" # Path to cloud auth and site files (default current directory) authmode = "cookie" # "cookie" (default) or "token" - use cookie or bearer token for auth cachefile = ".powerwall" # Path to cache file (default current directory) + fleetapi = False # If True, use Tesla FleetAPI for data (default is False) + auth_path = "" # Path to configfile (default current directory) + auto_select = False # If True, select the best available mode to connect (default is False) Functions poll(api, json, force) # Return data from Powerwall api (dict if json=True, bypass cache force=True) @@ -65,7 +68,6 @@ set_reserve(level) # Set Battery Reserve Percentage set_mode(mode) # Set Current Battery Operation Mode get_time_remaining() # Get the backup time remaining on the battery - set_operation(level, mode, json) # Set Battery Reserve Percentage and/or Operation Mode Requirements @@ -78,19 +80,23 @@ import sys from json import JSONDecodeError from typing import Union, Optional +import time # noinspection PyPackageRequirements import urllib3 -from pypowerwall.aux import HOST_REGEX, IPV4_6_REGEX, EMAIL_REGEX +from pypowerwall.regex import HOST_REGEX, IPV4_6_REGEX, EMAIL_REGEX from pypowerwall.exceptions import PyPowerwallInvalidConfigurationParameter, InvalidBatteryReserveLevelException from pypowerwall.cloud.pypowerwall_cloud import PyPowerwallCloud from pypowerwall.local.pypowerwall_local import PyPowerwallLocal +from pypowerwall.fleetapi.pypowerwall_fleetapi import PyPowerwallFleetAPI from pypowerwall.pypowerwall_base import parse_version, PyPowerwallBase +from pypowerwall.fleetapi.fleetapi import CONFIGFILE +from pypowerwall.cloud.pypowerwall_cloud import AUTHFILE urllib3.disable_warnings() # Disable SSL warnings -version_tuple = (0, 8, 5) +version_tuple = (0, 9, 0) version = __version__ = '%d.%d.%d' % version_tuple __author__ = 'jasonacox' @@ -98,7 +104,6 @@ log.debug('%s version %s', __name__, __version__) log.debug('Python %s on %s', sys.version, sys.platform) - def set_debug(toggle=True, color=True): """Enable verbose logging""" if toggle: @@ -115,7 +120,8 @@ def set_debug(toggle=True, color=True): class Powerwall(object): def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="America/Los_Angeles", pwcacheexpire=5, timeout=5, poolmaxsize=10, - cloudmode=False, siteid=None, authpath="", authmode="cookie", cachefile=".powerwall"): + cloudmode=False, siteid=None, authpath="", authmode="cookie", cachefile=".powerwall", + fleetapi=False, auto_select=False, retry_modes=False): """ Represents a Tesla Energy Gateway Powerwall device. @@ -133,6 +139,9 @@ def __init__(self, host="", password="", email="nobody@nowhere.com", authpath = Path to cloud auth and site cache files (default current directory) authmode = "cookie" (default) or "token" - use cookie or bearer token for authorization cachefile = Path to cache file (default current directory) + fleetapi = If True, use Tesla FleetAPI for data (default is False) + auto_select = If True, select the best available mode to connect (default is False) + retry_modes = If True, retry connection to Powerwall """ # Attributes @@ -145,8 +154,6 @@ def __init__(self, host="", password="", email="nobody@nowhere.com", self.poolmaxsize = poolmaxsize # pool max size for http connection re-use self.auth = {} # caches auth cookies self.token = None # caches bearer token - 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 or local mode (default) self.siteid = siteid # siteid for cloud mode @@ -155,23 +162,95 @@ def __init__(self, host="", password="", email="nobody@nowhere.com", self.pwcooldown = 0 # rate limit cooldown time - pause api calls self.vitals_api = True # vitals api is available for local mode self.client: PyPowerwallBase + self.fleetapi = fleetapi + self.retry_modes = retry_modes # Make certain assumptions here if not self.host: self.cloudmode = True + elif not self.cloudmode and not self.fleetapi: + self.mode = "local" + else: + self.mode = "unknown" + + # Auto select mode if requested + if auto_select: + if self.host and not self.cloudmode and not self.fleetapi: + log.debug("Auto selecting local mode") + self.cloudmode = self.fleetapi = False + self.mode = "local" + elif os.path.exists(os.path.join(self.authpath, CONFIGFILE)): + log.debug("Auto selecting FleetAPI Mode") + self.cloudmode = self.fleetapi = True + self.mode = "fleetapi" + elif os.path.exists(os.path.join(self.authpath, AUTHFILE)): + if not self.email or self.email == "nobody@nowhere.com": + with open(authpath + AUTHFILE, 'r') as file: + auth = json.load(file) + self.email = list(auth.keys())[0] + self.cloudmode = True + self.fleetapi = False + self.mode = "cloud" + log.debug("Auto selecting Cloud Mode (email: %s)" % self.email) + else: + log.error("Auto Select Failed: Unable to use local, cloud or fleetapi mode.") # Validate provided parameters self._validate_init_configuration() - # Check for cloud mode - if self.cloudmode: - self.client = PyPowerwallCloud(self.email, self.pwcacheexpire, self.timeout, self.siteid, self.authpath) - # Check to see if we can connect to the cloud - else: - self.client = PyPowerwallLocal(self.host, self.password, self.email, self.timezone, self.timeout, - self.pwcacheexpire, self.poolmaxsize, self.authmode, self.cachefile) + # Connect to Powerwall + self.connect(self.retry_modes) + + def connect(self, retry=False): + """ + Connect to Tesla Energy Gateway Powerwall - self.client.authenticate() + Args: + retry = If True, retry connection to Powerwall + """ + if self.mode == "unknown": + log.error("Unable to determine mode to connect.") + return + count = 0 + while count < 3: + count += 1 + if retry and count == 3: + log.error("Failed to connect to Powerwall with all modes. Waiting 30s to retry.") + time.sleep(30) + count = 0 + if self.mode == "local": + try: + self.client = PyPowerwallLocal(self.host, self.password, self.email, self.timezone, self.timeout, + self.pwcacheexpire, self.poolmaxsize, self.authmode, self.cachefile) + self.client.authenticate() + self.cloudmode = self.fleetapi = False + break + except Exception as exc: + log.error(f"Failed to connect using Local mode: {exc} - trying fleetapi mode.") + self.mode = "fleetapi" + continue + if self.mode == "fleetapi": + try: + self.client = PyPowerwallFleetAPI(self.email, self.pwcacheexpire, self.timeout, self.siteid, + self.authpath) + self.client.authenticate() + self.cloudmode = self.fleetapi = True + break + except Exception as exc: + log.error(f"Failed to connect using FleetAPI mode: {exc} - trying cloud mode.") + self.mode = "cloud" + continue + if self.mode == "cloud": + try: + self.client = PyPowerwallCloud(self.email, self.pwcacheexpire, self.timeout, self.siteid, self.authpath) + self.client.authenticate() + self.cloudmode = True + self.fleetapi = False + break + except Exception as exc: + log.error(f"Failed to connect using Cloud mode: {exc} - trying fleetapi mode.") + self.mode = "local" + continue def is_connected(self): """ @@ -740,7 +819,7 @@ def _validate_init_configuration(self): f"form of IP address or a valid form of a hostname or FQDN.") # If cloud mode requested, check appropriate parameters - if self.cloudmode: + if self.cloudmode and not self.fleetapi: # Ensure email is set and syntactically correct if not self.email or not isinstance(self.email, str) or not EMAIL_REGEX.match(self.email): raise PyPowerwallInvalidConfigurationParameter(f"A valid email address is required to run in " @@ -752,9 +831,15 @@ def _validate_init_configuration(self): log.debug("No authpath provided, using current directory.") dirname = '.' self._check_if_dir_is_writable(dirname, "authpath") - # If local mode, check appropriate parameters, too + elif self.fleetapi: + # Ensure we can write to the configfile + self.configfile = os.path.join(self.authpath, CONFIGFILE) + if os.access(self.configfile, os.W_OK): + log.debug(f"Config file '{self.configfile}' is writable.") + else: + raise PyPowerwallInvalidConfigurationParameter(f"Config file '{self.configfile}' is not writable.") else: - # Ensure we can create a cachefile + # If local mode, check appropriate parameters, and ensure we can write to the provided cachefile dirname = os.path.dirname(self.cachefile) if not dirname: log.debug("No cachefile provided, using current directory.") diff --git a/pypowerwall/__main__.py b/pypowerwall/__main__.py index 409983a..058643f 100644 --- a/pypowerwall/__main__.py +++ b/pypowerwall/__main__.py @@ -17,10 +17,11 @@ import json # Modules -from pypowerwall import version +from pypowerwall import version, set_debug +from pypowerwall.cloud.pypowerwall_cloud import AUTHFILE +from pypowerwall.fleetapi.fleetapi import CONFIGFILE # Global Variables -AUTHFILE = ".pypowerwall.auth" authpath = os.getenv("PW_AUTH_PATH", "") timeout = 1.0 @@ -38,6 +39,8 @@ setup_args = subparsers.add_parser("setup", help='Setup Tesla Login for Cloud Mode access') setup_args.add_argument("-email", type=str, default=email, help="Email address for Tesla Login.") +setup_args = subparsers.add_parser("fleetapi", help='Setup Tesla FleetAPI for Cloud Mode access') + scan_args = subparsers.add_parser("scan", help='Scan local network for Powerwall gateway') scan_args.add_argument("-timeout", type=float, default=timeout, help=f"Seconds to wait per host [Default={timeout:.1f}]") @@ -61,6 +64,9 @@ version_args = subparsers.add_parser("version", help='Print version information') +# Add a global debug flag +p.add_argument("-debug", action="store_true", default=False, help="Enable debug output") + if len(sys.argv) == 1: p.print_help(sys.stderr) sys.exit(1) @@ -69,6 +75,10 @@ args = p.parse_args() command = args.command +# Set Debug Mode +if args.debug: + set_debug(True) + # Cloud Mode Setup if command == 'setup': from pypowerwall import PyPowerwallCloud @@ -78,10 +88,22 @@ # Run Setup c = PyPowerwallCloud(None, authpath=authpath) if c.setup(email): - print("Setup Complete. Auth file %s ready to use." % AUTHFILE) + print(f"Setup Complete. Auth file {c.authfile} ready to use.") else: print("ERROR: Failed to setup Tesla Cloud Mode") exit(1) +# FleetAPI Mode Setup +elif command == 'fleetapi': + from pypowerwall import PyPowerwallFleetAPI + + print("pyPowerwall [%s] - FleetAPI Mode Setup\n" % version) + # Run Setup + c = PyPowerwallFleetAPI(None, authpath=authpath) + if c.setup(): + print(f"Setup Complete. Config file {c.configfile} ready to use.") + else: + print("Setup Aborted.") + exit(1) # Run Scan elif command == 'scan': from pypowerwall import scan @@ -94,17 +116,17 @@ scan.scan(color, timeout, hosts, ip) # Set Powerwall Mode elif command == 'set': + # If no arguments, print usage + if not args.mode and not args.reserve and not args.current: + print("usage: pypowerwall set [-h] [-mode MODE] [-reserve RESERVE] [-current]") + exit(1) import pypowerwall print("pyPowerwall [%s] - Set Powerwall Mode and Power Levels\n" % version) - # Load email from auth file - auth_file = authpath + AUTHFILE - if not os.path.exists(auth_file): - print("ERROR: Auth file %s not found. Run 'setup' to create." % auth_file) + # Determine which cloud mode to use + pw = pypowerwall.Powerwall(auto_select=True, host="", authpath=authpath) + if not pw.client: + print("ERROR: FleetAPI and Cloud access are not configured. Run 'fleetapi' or 'setup' to create.") exit(1) - with open(auth_file, 'r') as file: - auth = json.load(file) - email = list(auth.keys())[0] - pw = pypowerwall.Powerwall(email=email, host="", authpath=authpath) if args.mode: mode = args.mode.lower() if mode not in ['self_consumption', 'backup', 'autonomous']: @@ -124,16 +146,10 @@ elif command == 'get': import pypowerwall # Load email from auth file - auth_file = authpath + AUTHFILE - if not os.path.exists(auth_file): - print("ERROR: Auth file %s not found. Run 'setup' to create." % auth_file) - exit(1) - with open(auth_file, 'r') as file: - auth = json.load(file) - email = list(auth.keys())[0] - pw = pypowerwall.Powerwall(email=email, host="", authpath=authpath) + pw = pypowerwall.Powerwall(auto_select=True, host="", authpath=authpath) output = { 'site': pw.site_name(), + 'site_id': pw.client.fleet.site_id, 'din': pw.din(), 'mode': pw.get_mode(), 'reserve': pw.get_reserve(), diff --git a/pypowerwall/fleetapi/__init__.py b/pypowerwall/fleetapi/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pypowerwall/fleetapi/__main__.py b/pypowerwall/fleetapi/__main__.py new file mode 100644 index 0000000..6df5eca --- /dev/null +++ b/pypowerwall/fleetapi/__main__.py @@ -0,0 +1,175 @@ +# pyPowerWall Module - FleetAPI Command Line Interface +# -*- coding: utf-8 -*- +""" + Command Line Interface for Tesla FleetAPI to read and control Powerwall + status and settings. This module is a command line interface to the + FleetAPI class in the pypowerwall module. + + Author: Jason A. Cox + For more information see https://github.com/jasonacox/pypowerwall + + Command Line Interface for FleetAPI: + python3 -m pypowerwall.fleetapi + +""" + +# Import Libraries +import sys +import json +from .fleetapi import FleetAPI, CONFIGFILE +import argparse + +# Display help if no arguments +if len(sys.argv) == 1: + print("Tesla FleetAPI - Command Line Interface\n") + print(f"Usage: {sys.argv[0]} command [arguments] [-h] [--debug] [--config CONFIG] [--site SITE] [--json]\n") + print("Commands:") + print(" setup Setup FleetAPI for your site") + print(" sites List available sites") + print(" status Report current power status for your site") + print(" info Display information about your site") + print(" getmode Get current operational mode setting") + print(" getreserve Get current battery reserve level setting") + print(" setmode Set operatinoal mode (self_consumption or autonomous)") + print(" setreserve Set battery reserve level (prcentage or 'current')\n") + print("options:") + print(" --debug Enable debug mode") + print(" --config CONFIG Specify alternate config file (default: .fleetapi.config)") + print(" --site SITE Specify site_id") + print(" --json Output in JSON format") + exit(0) + +# Parse command line arguments +parser = argparse.ArgumentParser(description='Tesla FleetAPI - Command Line Interface') +parser.add_argument("command", choices=["setup", "sites", "status", "info", "getmode", "getreserve", + "setmode", "setreserve"], help="Select command to execute") +parser.add_argument("argument", nargs="?", default=None, help="Argument for setmode or setreserve command") +parser.add_argument("--debug", action="store_true", help="Enable debug mode") +parser.add_argument("--config", help="Specify alternate config file") +parser.add_argument("--site", help="Specify site_id") +parser.add_argument("--json", action="store_true", help="Output in JSON format") + +# Adding descriptions for each command +parser.add_help = False # Disabling default help message +args = parser.parse_args() + +settings_file = CONFIGFILE +if args.config: + # Use alternate config file if specified + settings_file = args.config + +# Create FleetAPI object +settings_debug = False +settings_site = None +if args.debug: + settings_debug = True +if args.site: + settings_site = args.site + +# Create FleetAPI object +fleet = FleetAPI(configfile=settings_file, debug=settings_debug, site_id=settings_site) + +# Load Configuration +if not fleet.load_config(): + print(f" Configuration file not found: {settings_file}") + if args.command != "setup": + print(" Run setup to access Tesla FleetAPI.") + exit(1) + else: + fleet.setup() + if not fleet.load_config(): + print(" Setup failed, exiting...") + exit(1) + exit(0) + +# Command: Run Setup +if args.command == "setup": + fleet.setup() + exit(0) + +# Command: List Sites +if args.command == "sites": + sites = fleet.getsites() + if args.json: + print(json.dumps(sites, indent=4)) + else: + for site in sites: + print(f" {site['energy_site_id']} - {site['site_name']}") + exit(0) + +# Command: Status +if args.command == "status": + status = fleet.get_live_status() + if args.json: + print(json.dumps(status, indent=4)) + else: + for key in status: + print(f" {key}: {status[key]}") + exit(0) + +# Command: Site Info +if args.command == "info": + info = fleet.get_site_info() + if args.json: + print(json.dumps(info, indent=4)) + else: + for key in info: + print(f" {key}: {info[key]}") + exit(0) + +# Command: Get Operating Mode +if args.command == "getmode": + mode = fleet.get_operating_mode() + if args.json: + print(json.dumps({"mode": mode}, indent=4)) + else: + print(f"{mode}") + exit(0) + +# Command: Get Battery Reserve +if args.command == "getreserve": + reserve = fleet.get_battery_reserve() + if args.json: + print(json.dumps({"reserve": reserve}, indent=4)) + else: + print(f"{reserve}") + exit(0) + +# Command: Set Operating Mode +if args.command == "setmode": + if args.argument: + # autonomous or self_consumption + if args.argument in ["self", "self_consumption"]: + print(fleet.set_operating_mode("self_consumption")) + elif args.argument in ["auto", "time", "autonomous"]: + print(fleet.set_operating_mode("autonomous")) + else: + print("Invalid mode, must be 'self' or 'auto'") + exit(1) + else: + print("No mode specified, exiting...") + exit(0) + +# Command: Set Battery Reserve +if args.command == "setreserve": + if args.argument: + if args.argument.isdigit(): + val = int(args.argument) + if val < 0 or val > 100: + print(f"Invalid reserve level {val}, must be 0-100") + exit(1) + elif args.argument == "current": + val = fleet.battery_level() + else: + print("Invalid reserve level, must be 0-100 or 'current' to set to current level.") + exit(1) + print(fleet.set_battery_reserve(int(val))) + else: + print("No reserve level specified, exiting...") + exit(0) + +print("No command specified, exiting...") +exit(1) + + + diff --git a/pypowerwall/fleetapi/decorators.py b/pypowerwall/fleetapi/decorators.py new file mode 100644 index 0000000..96b58d0 --- /dev/null +++ b/pypowerwall/fleetapi/decorators.py @@ -0,0 +1,19 @@ +import functools +import logging + +log = logging.getLogger('pypowerwall.fleetapi.pypowerwall_fleetapi') +WARNED_ONCE = {} + + +def not_implemented_mock_data(func): + @functools.wraps(func) + def wrapper(*args, **kwargs): + if not WARNED_ONCE.get(func.__name__): + log.warning(f"This API [{func.__name__}] is using mock data in fleetapi mode. This message will be " + "printed only once at the warning level.") + WARNED_ONCE[func.__name__] = 1 + else: + log.debug(f"This API [{func.__name__}] is using mock data in fleetapi mode.") + return func(*args, **kwargs) + + return wrapper diff --git a/pypowerwall/fleetapi/exceptions.py b/pypowerwall/fleetapi/exceptions.py new file mode 100644 index 0000000..5514e13 --- /dev/null +++ b/pypowerwall/fleetapi/exceptions.py @@ -0,0 +1,14 @@ +class PyPowerwallFleetAPINoTeslaAuthFile(Exception): + pass + + +class PyPowerwallFleetAPITeslaNotConnected(Exception): + pass + + +class PyPowerwallFleetAPINotImplemented(Exception): + pass + + +class PyPowerwallFleetAPIInvalidPayload(Exception): + pass diff --git a/pypowerwall/fleetapi/fleetapi.py b/pypowerwall/fleetapi/fleetapi.py new file mode 100644 index 0000000..260a592 --- /dev/null +++ b/pypowerwall/fleetapi/fleetapi.py @@ -0,0 +1,771 @@ +# pyPowerWall - Tesla FleetAPI Class +# -*- coding: utf-8 -*- +""" + Tesla FleetAPI Class + + This module allows you to access the Tesla FleetAPI to manage + your Powerwall. It has a CLI that can be run in setup mode to + walk you through the steps to get access to the Tesla FleetAPI. + + Class: + FleetAPI - Tesla FleetAPI Class + + Functions: + poll(api, action, data) - poll FleetAPI + getsites() - get sites + site_name() - get site name + ... Get + get_live_status() - get the current power information for the site + get_site_info() - get site info + get_battery_reserve() - get battery reserve level + get_operating_mode() - get operating mode + solar_power() - get solar power + grid_power() - get grid power + battery_power() - get battery power + load_power() - get load power + battery_level() - get battery level + energy_left() - get energy left + total_pack_energy() - get total pack energy + grid_status() - get grid status + island_status() - get island status + firmware_version() - get firmware version + ... Set + set_battery_reserve(reserve) - set battery reserve level (percent) + set_operating_mode(mode) - set operating mode (self_consumption or autonomous) + + Author: Jason A. Cox + Date: 18 Feb 2024 + For more information see https://github.com/jasonacox/pypowerwall + + Requirements + + * Register your application https://developer.tesla.com/ + * Before running this script, you must first run create_pem_key.py + to create a PEM key and register it with Tesla. Put the public + key in {site}/.well-known/appspecific/com.tesla.3p.public-key.pem + * Python: pip install requests + + Tesla FleetAPI Reference: https://developer.tesla.com/docs/fleet-api +""" + +# FleetAPI Class + +import os +import json +import requests +import logging +import sys +import urllib.parse +import time + +# Defaults +CONFIGFILE = ".pypowerwall.fleetapi" +SCOPE = "openid offline_access energy_device_data energy_cmds" + +fleet_api_urls = { + "North America, Asia-Pacific": "https://fleet-api.prd.na.vn.cloud.tesla.com", + "Europe, Middle East, Africa": "https://fleet-api.prd.eu.vn.cloud.tesla.com", + "China": "https://fleet-api.prd.cn.vn.cloud.tesla.cn" +} + +# Set up logging +log = logging.getLogger(__name__) + +class FleetAPI: + def __init__(self, configfile=CONFIGFILE, debug=False, site_id=None, + pwcacheexpire: int = 5, timeout: int = 5): + self.CLIENT_ID = "" + self.CLIENT_SECRET = "" + self.DOMAIN = "" + self.REDIRECT_URI = "" + self.AUDIENCE = "" + self.partner_token = "" + self.partner_account = {} + self.access_token = "" + self.refresh_token = "" + self.site_id = "" + self.debug = debug + self.configfile = configfile + self.pwcachetime = {} # holds the cached data timestamps for api + self.pwcacheexpire = pwcacheexpire # seconds to expire cache + self.pwcache = {} # holds the cached data for api + self.refreshing = False + self.timeout = timeout + + if debug: + log.setLevel(logging.DEBUG) + if configfile: + self.configfile = configfile + self.load_config() + if site_id: + self.site_id = site_id + if not self.site_id: + log.error("No site_id set or returned by FleetAPI.") + raise Exception("No site_id found - Run Setup.") + + # Function to return a random string of characters and numbers + def random_string(self, length): + import random + import string + return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(length)) + + # Return key value from data or None + def keyval(self, data, key): + return data.get(key) if data and key else None + + # Load Configuration + def load_config(self): + if os.path.isfile(self.configfile): + with open(self.configfile, 'r') as f: + config = json.loads(f.read()) + # Set the global variables + self.CLIENT_ID = config.get('CLIENT_ID') + self.CLIENT_SECRET = config.get('CLIENT_SECRET') + self.DOMAIN = config.get('DOMAIN') + self.REDIRECT_URI = config.get('REDIRECT_URI') + self.AUDIENCE = config.get('AUDIENCE') + self.partner_token = config.get('partner_token') + self.partner_account = config.get('partner_account') + self.access_token = config.get('access_token') + self.refresh_token = config.get('refresh_token') + self.site_id = config.get('site_id') + # Check for valid site_id + if not self.site_id: + sites = self.getsites() + self.site_id = sites[0]['energy_site_id'] + log.debug(f"Configuration loaded: {self.configfile}") + return config + else: + log.debug(f"Configuration file not found: {self.configfile}") + return False + + # Save Configuration + def save_config(self): + # Copy the global variables to the config dictionary + config = { + "CLIENT_ID": self.CLIENT_ID, + "CLIENT_SECRET": self.CLIENT_SECRET, + "DOMAIN": self.DOMAIN, + "REDIRECT_URI": self.REDIRECT_URI, + "AUDIENCE": self.AUDIENCE, + "partner_token": self.partner_token, + "partner_account": self.partner_account, + "access_token": self.access_token, + "refresh_token": self.refresh_token, + "site_id": self.site_id + } + # Save the config dictionary to the file + with open(self.configfile, 'w') as f: + f.write(json.dumps(config, indent=4)) + + # Refresh Token + def new_token(self): + # Lock to prevent multiple refreshes + if self.refreshing: + return + self.refreshing = True + print("Token expired, refreshing token...") + data = { + 'grant_type': 'refresh_token', + 'client_id': self.CLIENT_ID, + 'refresh_token': self.refresh_token + } + headers = { + 'Content-Type': 'application/x-www-form-urlencoded' + } + response = requests.post('https://auth.tesla.com/oauth2/v3/token', + data=data, headers=headers) + # Extract access_token and refresh_token from this response + access = response.json().get('access_token') + refresh = response.json().get('refresh_token') + # If access or refresh token is None return + if not access or not refresh or response.status_code > 201: + print(f"Unable to refresh token. Response code: {response.status_code}") + self.refreshing = False + return + self.access_token = access + self.refresh_token = refresh + log.info("Token refreshed - saving.") + log.debug(f" Response Code: {response.status_code}") + log.debug(f" Access Token: {self.access_token}") + log.debug(f" Refresh Token: {self.refresh_token}") + # Update config + self.save_config() + self.refreshing = False + + # Poll FleetAPI + def poll(self, api="api/1/products", action="GET", data=None, recursive=False, force=False): + url = f"{self.AUDIENCE}/{api}" + headers = { + "Content-Type": "application/json", + "Authorization": "Bearer " + self.access_token + } + if action == "POST": + # Post to FleetAPI with json data payload + log.debug(f"POST: {url} {json.dumps(data)}") + # Check for timeout exception + try: + response = requests.post(url, headers=headers, + data=json.dumps(data), timeout=self.timeout) + except requests.exceptions.Timeout: + log.error(f"Timeout error posting to {url}") + return None + else: + # Check if we have a cached response + if not force and api in self.pwcachetime: + if time.time() - self.pwcachetime[api] < self.pwcacheexpire: + log.debug(f"Using cached data for {api}") + return self.pwcache[api] + log.debug(f"GET: {url}") + try: + response = requests.get(url, headers=headers, timeout=self.timeout) + except requests.exceptions.Timeout: + log.error(f"Timeout error polling {url}") + return None + if response.status_code == 401 and not recursive: + # Token expired, refresh token and try again + self.new_token() + data = self.poll(api, action, data, True) + elif response.status_code == 401: + print("Token expired, refresh token failed, exiting...") + data = None + elif response.status_code != 200: + print(f"Error: {response.status_code}") + print(response.text) + data = None + else: + data = response.json() + if action == "GET": + # Cache the data + self.pwcachetime[api] = time.time() + self.pwcache[api] = data + return data + + def get_live_status(self, force=False): + # Get the current power information for the site. + """ + { + 'response': { + 'solar_power': 0, + 'percentage_charged': 46.6731017783813, + 'backup_capable': True, + 'battery_power': 780, + 'load_power': 780, + 'grid_status': 'Active', + 'grid_services_active': False, + 'grid_power': 0, + 'grid_services_power': 0, + 'generator_power': 0, + 'island_status': 'on_grid', + 'storm_mode_active': False, + 'timestamp': '2024-05-12T00:18:19-07:00', + 'wall_connectors': [] + } + } + """ + payload = self.poll(f"api/1/energy_sites/{self.site_id}/live_status", force=force) + log.debug(f"get_live_status: {payload}") + return self.keyval(payload, "response") + + def get_site_info(self, force=False): + # Get site info + """ + { + 'id': '1234000-00-E--TG12345678904G', + 'site_name': 'TeslaEnergyGateway', + 'backup_reserve_percent': 20, + 'default_real_mode': 'self_consumption', + 'installation_date': '2021-09-25T15: 53: 47-07: 00', + 'user_settings': { + 'go_off_grid_test_banner_enabled': False, + 'storm_mode_enabled': False, + 'powerwall_onboarding_settings_set': True, + 'powerwall_tesla_electric_interested_in': False, + 'vpp_tour_enabled': True, + 'sync_grid_alert_enabled': True, + 'breaker_alert_enabled': False + }, + 'components': { + 'solar': True, + 'solar_type': 'pv_panel', + 'battery': True, + 'grid': True, + 'backup': True, + 'gateway': 'teg', + 'load_meter': True, + 'tou_capable': True, + 'storm_mode_capable': True, + 'flex_energy_request_capable': False, + 'car_charging_data_supported': False, + 'off_grid_vehicle_charging_reserve_supported': True, + 'vehicle_charging_performance_view_enabled': False, + 'vehicle_charging_solar_offset_view_enabled': False, + 'battery_solar_offset_view_enabled': True, + 'solar_value_enabled': True, + 'energy_value_header': 'EnergyValue', + 'energy_value_subheader': 'EstimatedValue', + 'energy_service_self_scheduling_enabled': True, + 'show_grid_import_battery_source_cards': True, + 'set_islanding_mode_enabled': True, + 'wifi_commissioning_enabled': True, + 'backup_time_remaining_enabled': True, + 'battery_type': 'solar_powerwall', + 'configurable': True, + 'grid_services_enabled': False, + 'gateways': [ + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '1232100-00-E--TG12345678904G', + 'serial_number': 'TG12345678904G', + 'part_number': '1232100-00-E', + 'part_type': 10, + 'part_name': 'TeslaBackupGateway2', + 'is_active': True, + 'site_id': 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxx', + 'firmware_version': '24.4.00fe780c9', + 'updated_datetime': '2024-05-11T09: 20: 26.225Z' + } + ], + 'batteries': [ + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '2012170-25-E--TG12345678904G', + 'serial_number': 'TG12345678904G', + 'part_number': '2012170-25-E', + 'part_type': 2, + 'part_name': 'Powerwall2', + 'nameplate_max_charge_power': 5400, + 'nameplate_max_discharge_power': 5400, + 'nameplate_energy': 13500 + }, + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '3012170-05-B--TG12345678904G', + 'serial_number': 'TG12345678904G', + 'part_number': '3012170-05-B', + 'part_type': 2, + 'part_name': 'Powerwall2', + 'nameplate_max_charge_power': 5400, + 'nameplate_max_discharge_power': 5400, + 'nameplate_energy': 13500 + } + ], + 'inverters': [ + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '1530000-00-F--CN12345678901J', + 'part_number': '1538100-00-F', + 'part_type': 7, + 'part_name': 'Non-TeslaInverter', + 'is_active': True, + 'site_id': 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxx', + } + ], + 'edit_setting_permission_to_export': True, + 'edit_setting_grid_charging': True, + 'edit_setting_energy_exports': True, + 'system_alerts_enabled': True + }, + 'version': '24.4.00fe780c9', + 'battery_count': 2, + 'tariff_content': {} + } + """ + payload = self.poll(f"api/1/energy_sites/{self.site_id}/site_info", force=force) + log.debug(f"get_site_info: {payload}") + return self.keyval(payload, "response") + + def get_site_status(self, force=False): + # Get site status + """ + { + 'response': { + 'resource_type': 'battery', + 'site_name': 'Tesla Energy Gateway', + 'gateway_id': '1234000-00-E--TG12345678904G', + 'percentage_charged': 46.6731017783813, + 'battery_type': 'ac_powerwall', + 'backup_capable': True, + 'battery_power': 820, + 'go_off_grid_test_banner_enabled': None, + 'storm_mode_enabled': False, + 'powerwall_onboarding_settings_set': True, + 'powerwall_tesla_electric_interested_in': None, + 'vpp_tour_enabled': None, + 'sync_grid_alert_enabled': True, + 'breaker_alert_enabled': True + } + } + """ + payload = self.poll(f"api/1/energy_sites/{self.site_id}/site_status", force=force) + log.debug(f"get_site_status: {payload}") + return self.keyval(payload, "response") + + def get_backup_time_remaining(self, force=False): + # Get backup time remaining + """ + {'response': {'time_remaining_hours': 9.863332186566478}} + """ + payload = self.poll(f"api/1/energy_sites/{self.site_id}/backup_time_remaining", force=force) + log.debug(f"get_backup_time_remaining: {payload}") + return self.keyval(payload, "response") + + def get_products(self, force=False): + # Get list of Tesla products assigned to user + """ + { + "response": [ + { + "id": 100021, + "user_id": 429511308124, + "vehicle_id": 99999, + "vin": "5YJ3000000NEXUS01", + "color": null, + "access_type": "OWNER", + "display_name": "Owned", + "option_codes": "TEST0,COUS", + "cached_data": null, + "granular_access": { + "hide_private": false + }, + "tokens": [ + "4f993c5b9e2b937b", + "7a3153b1bbb48a96" + ], + "state": null, + "in_service": false, + "id_s": "100021", + "calendar_enabled": false, + "api_version": null, + "backseat_token": null, + "backseat_token_updated_at": null, + "command_signing": "off" + }, + { + "energy_site_id": 429124, + "resource_type": "battery", + "site_name": "My Home", + "id": "STE12345678-12345", + "gateway_id": "1112345-00-E--TG0123456789", + "energy_left": 35425, + "total_pack_energy": 39362, + "percentage_charged": 90, + "battery_power": 1000 + } + ], + "count": 2 + } + """ + payload = self.poll(f"api/1/products", force=force) + log.debug(f"get_products: {payload}") + return self.keyval(payload, "response") + + def set_battery_reserve(self, reserve: int): + if reserve < 0 or reserve > 100: + log.debug(f"Invalid reserve level: {reserve}") + return False + data = {"backup_reserve_percent": reserve} + # 'https://fleet-api.prd.na.vn.cloud.tesla.com/api/1/energy_sites/{energy_site_id}/backup' + payload = self.poll(f"api/1/energy_sites/{self.site_id}/backup", "POST", data) + # Invalidate cache + self.pwcachetime.pop(f"api/1/energy_sites/{self.site_id}/site_info", None) + return payload + + def set_operating_mode(self, mode: str): + data = {"default_real_mode": mode} + if mode not in ["self_consumption", "autonomous"]: + log.debug(f"Invalid mode: {mode}") + return False + # 'https://fleet-api.prd.na.vn.cloud.tesla.com/api/1/energy_sites/{energy_site_id}/operation' + payload = self.poll(f"api/1/energy_sites/{self.site_id}/operation", "POST", data) + # Invalidate cache + self.pwcachetime.pop(f"api/1/energy_sites/{self.site_id}/site_info", None) + return payload + + def get_operating_mode(self, force=False): + return self.keyval(self.get_site_info(force=force), "default_real_mode") + + def get_battery_reserve(self, force=False): + return self.keyval(self.get_site_info(force=force), "backup_reserve_percent") + + def getsites(self, force=False): + payload = self.poll("api/1/products", force=force) + return self.keyval(payload, "response") + + # Macros for common data + def solar_power(self): + return self.keyval(self.get_live_status(), "solar_power") + def grid_power(self): + return self.keyval(self.get_live_status(), "grid_power") + def battery_power(self): + return self.keyval(self.get_live_status(), "battery_power") + def load_power(self): + return self.keyval(self.get_live_status(), "load_power") + def home_power(self): + return self.keyval(self.get_live_status(), "load_power") + def site_name(self): + return self.keyval(self.get_site_info(), "site_name") + def battery_level(self, force=False): + return self.keyval(self.get_live_status(force=force), "percentage_charged") + def battery_reserve(self): + return self.get_battery_reserve() + def operating_mode(self): + return self.get_operating_mode() + def energy_left(self, force=False): + return self.keyval(self.get_site_status(force=force), "energy_left") # FIXME: This is not in the API + def total_pack_energy(self, force=False): + return self.keyval(self.get_site_status(force=force), "total_pack_energy") # FIXME: This is not in the API + def grid_status(self): + return self.keyval(self.get_live_status(), "grid_status") + def island_status(self): + return self.keyval(self.get_live_status(), "island_status") + def firmware_version(self): + return self.keyval(self.get_site_info(), "firmware_version") + + # Setup Environment + def setup(self): + # Print Header + print("\nTesla FleetAPI Setup") + print("--------------------") + print() + print("Step 1 - Register your application at https://developer.tesla.com/") + print("Step 2 - Run create_pem_key.py to create a PEM key file for your website.") + print(" Put the public key in {site}/.well-known/appspecific/com.tesla.3p.public-key.pem") + print("Step 3 - Run this script to generate a partner token, register your partner account,") + print(" generate a user token, and get the site_id and live data for your Tesla Powerwall.") + print() + + # Display current configuration if we have it + config = self.load_config() + if config: + print(f"Current Configuration - Loaded: {self.configfile}:") + for item in config: + val = config[item] + if isinstance(val, dict): + continue + if isinstance(val, str) and len(val) > 50: + val = val[:50] + "..." + val[-5:] + print(f" {item}: {val}") + # Ask user if they wish to overwrite the configuration + overwrite = input("\nDo you want to overwrite this configuration? [y/N]: ") + if not overwrite.lower().startswith("y"): + print("Exiting...") + return False + else: + print(f"No configuration found - Creating: {self.configfile}") + # Get the client_id and client_secret from the userl + print("\nStep 3 - Enter your Tesla FleetAPI credentials...") + client_id = input(f" Enter Client ID [{self.CLIENT_ID}]: ") + if client_id: + self.CLIENT_ID = client_id + client_secret = input(f" Enter Client Secret [{self.CLIENT_SECRET}]: ") + if client_secret: + self.CLIENT_SECRET = client_secret + domain = input(f" Enter Domain [{self.DOMAIN}]: ") + if domain: + self.DOMAIN = domain + redirect_uri = f"https://{self.DOMAIN}/access" + if self.REDIRECT_URI: + redirect_uri = self.REDIRECT_URI + r = input(f" Enter Redirect URI [{redirect_uri}]: ") + if r: + self.REDIRECT_URI = r + else: + self.REDIRECT_URI = redirect_uri + # Select AUDIENCE from region list + if not self.AUDIENCE: + self.AUDIENCE = list(fleet_api_urls.values())[0] + current_audience = 1 + print(" Select your region:") + for i, region in enumerate(fleet_api_urls): + if self.AUDIENCE == fleet_api_urls[region]: + print(f" * {i+1}. {region} [{fleet_api_urls[region]}] (current)") + current_audience = i+1 + else: + print(f" {i+1}. {region} [{fleet_api_urls[region]}]") + region = input(f" Enter Region [{current_audience}]: ") + if region: + self.AUDIENCE = list(fleet_api_urls.values())[int(region)-1] + print() + # Save the configuration + self.save_config() + print(" Configuration saved") + print() + + # Generate Partner Token + print("Step 3A - Generating a partner authentication token...") + # Verify that the PEM key file exists + print(" Verifying PEM Key file...") + verify_url = f"https://{self.DOMAIN}/.well-known/appspecific/com.tesla.3p.public-key.pem" + response = requests.get(verify_url) + if response.status_code != 200: + print(f"ERROR: Could not verify PEM key file at {verify_url}") + print(f" Make sure you have created the PEM key file and uploaded it to your website.") + print() + print("Run create_pem_key.py to create a PEM key file for your website.") + return False + print(f" * Success: PEM Key file verified at {verify_url}.") + print() + # Check to see if already cached + if self.partner_token: + print(" Using cached token.") + log.debug(f"Cached partner token: {self.partner_token}") + else: + # If not cached, generate a new token + data = { + 'grant_type': 'client_credentials', + 'client_id': self.CLIENT_ID, + 'client_secret': self.CLIENT_SECRET, + 'scope': SCOPE, + 'audience': self.AUDIENCE + } + headers = { + 'Content-Type': 'application/x-www-form-urlencoded' + } + log.debug(f"POST: https://auth.tesla.com/oauth2/v3/token {json.dumps(data)}") + response = requests.post('https://auth.tesla.com/oauth2/v3/token', + data=data, headers=headers) + log.debug(f"Response Code: {response.status_code}") + partner_token = response.json().get("access_token") + self.partner_token = partner_token + print(f" Got Token: {partner_token[:40]}...\n") + log.debug(f"Partner Token: {partner_token}") + # Save the configuration + self.save_config() + print(" Configuration saved") + print() + + # Register Partner Account + print("Step 3B - Registering your partner account...") + if self.partner_account: + print(f" Already registered. Skipping...") + else: + # If not registered, register + url = f"{self.AUDIENCE}/api/1/partner_accounts" + headers = { + 'Content-Type': 'application/json', + 'Authorization': 'Bearer ' + self.partner_token + } + data = { + 'domain': self.DOMAIN, + } + log.debug(f"POST: {url} {json.dumps(data)}") + response = requests.post(url, headers=headers, data=json.dumps(data)) + log.debug(f" Response Code: {response.status_code}") + self.partner_account = response.json() + log.debug(f"Partner Account: {json.dumps(self.partner_account, indent=4)}\n") + # Save the configuration + self.save_config() + print(" Configuration saved") + print() + + # Generate User Token + print("Step 3C - Generating a one-time authentication token...") + if self.access_token and self.refresh_token: + print(f" Replacing cached tokens...") + scope = urllib.parse.quote(SCOPE) + state = self.random_string(64) + url = f"https://auth.tesla.com/oauth2/v3/authorize?&client_id={self.CLIENT_ID}&locale=en-US&prompt=login&redirect_uri={self.REDIRECT_URI}&response_type=code&scope={scope}&state={state}" + # Prompt user to login to Tesla account and authorize access + print(" Login to your Tesla account to authorize access.") + print(f" Go to this URL: {url}") + # If on Mac, automatically open the URL in the default browser + if sys.platform == 'darwin': + import subprocess + subprocess.call(['open', url]) + print("\nAfter authorizing access, copy the code from the URL and paste it below.") + code = input(" Enter the code: ") + # Check to see if user pasted URL or just the code + if code.startswith("http"): + code = code.split("code=")[1].split("&")[0] + print() + log.debug(f"Code: {code}") + + # Step 3D - Exchange the authorization code for a token + # The access_token will be used as the Bearer token + # in the Authorization header when making API requests. + print("Step 3D - Exchange the authorization code for a token") + data = { + 'grant_type': 'authorization_code', + 'client_id': self.CLIENT_ID, + 'client_secret': self.CLIENT_SECRET, + 'code': code, + 'audience': self.AUDIENCE, + 'redirect_uri': self.REDIRECT_URI, + 'scope': SCOPE + } + headers = { + 'Content-Type': 'application/x-www-form-urlencoded' + } + log.debug(f"POST: https://auth.tesla.com/oauth2/v3/token {json.dumps(data)}") + response = requests.post('https://auth.tesla.com/oauth2/v3/token', + data=data, headers=headers) + log.debug(f"Response Code: {response.status_code}") + # Extract access_token and refresh_token from this response + access_token = response.json().get('access_token') + refresh_token = response.json().get('refresh_token') + print("\n Tokens generated.") + print(f" * Access Token: {access_token}") + print(f" * Refresh Token: {refresh_token}\n") + self.access_token = access_token + self.refresh_token = refresh_token + # Save the configuration + self.save_config() + print(" Configuration saved") + print() + + # Select Site ID + print("Step 4 - Select site_id for your Tesla Powerwall...") + if self.site_id: + print(f" Previous site_id: {self.site_id}") + # Get list of sites + sites = self.getsites() + sel = 0 + # If not set, pick first site + if not self.site_id: + self.site_id = sites[0]['energy_site_id'] + sel = 1 + log.debug(sites) + print(" Sites:") + for i, site in enumerate(sites): + if self.site_id == site['energy_site_id']: + print(f" * {i+1}. {site['energy_site_id']} - {site['site_name']} (current)") + sel = i+1 + else: + print(f" {i+1}. {site['energy_site_id']} - {site['site_name']}") + # If only one site, use it + if len(sites) == 1: + print(f" Using site: {sites[0]['energy_site_id']}") + self.site_id = sites[0]['energy_site_id'] + sel = 1 + else: + site = input(f" Enter Site ID [{sel}]: ") + if site: + self.site_id = sites[int(site)-1]['energy_site_id'] + print() + log.debug(f"Site ID: {self.site_id}") + # Save the configuration + self.save_config() + print(" Configuration saved") + print() + + # Get Site Info + print("Step 5 - Verifying Access...") + site_info = self.get_site_info() + # List all the site info + print() + print(" Site Info:") + for key in site_info: + print(f" * {key}: {site_info[key]}") + print() + print(" Live Status:") + live_status = self.get_live_status() + for key in live_status: + print(f" * {key}: {live_status[key]}") + print() + print("Setup complete.") + print("You can now use this script to manage your Tesla Powerwall.") + print() + return True + +# End of FleetAPI Class diff --git a/pypowerwall/fleetapi/mock_data.py b/pypowerwall/fleetapi/mock_data.py new file mode 100644 index 0000000..248600a --- /dev/null +++ b/pypowerwall/fleetapi/mock_data.py @@ -0,0 +1,179 @@ +POWERWALLS = ('{"enumerating": false, "updating": false, "checking_if_offgrid": false, "running_phase_detection": ' + 'false, "phase_detection_last_error": "no phase information", "bubble_shedding": false, ' + '"on_grid_check_error": "on grid check not run", "grid_qualifying": false, "grid_code_validating": ' + 'false, "phase_detection_not_available": true, "powerwalls": [{"Type": "", "PackagePartNumber": ' + '"2012170-25-E", "PackageSerialNumber": "TG1234567890G1", "type": "SolarPowerwall", "grid_state": ' + '"Grid_Uncompliant", "grid_reconnection_time_seconds": 0, "under_phase_detection": false, "updating": ' + 'false, "commissioning_diagnostic": {"name": "Commissioning", "category": "InternalComms", ' + '"disruptive": false, "inputs": null, "checks": [{"name": "CAN connectivity", "status": "fail", ' + '"start_time": "2023-12-16T08:34:17.3068631-08:00", "end_time": "2023-12-16T08:34:17.3068696-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}, {"name": "Enable switch", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.306875474-08:00", "end_time": "2023-12-16T08:34:17.306880724-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}, {"name": "Internal communications", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.306886099-08:00", "end_time": "2023-12-16T08:34:17.306891223-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}, {"name": "Firmware up-to-date", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.306896598-08:00", "end_time": "2023-12-16T08:34:17.306901723-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}], "alert": false}, "update_diagnostic": {"name": "Firmware Update", "category": ' + '"InternalComms", "disruptive": true, "inputs": null, "checks": [{"name": "Solar Inverter firmware", ' + '"status": "not_run", "start_time": null, "end_time": null, "progress": 0, "results": null, ' + '"debug": null, "checks": null}, {"name": "Solar Safety firmware", "status": "not_run", "start_time": ' + 'null, "end_time": null, "progress": 0, "results": null, "debug": null, "checks": null}, ' + '{"name": "Grid code", "status": "not_run", "start_time": null, "end_time": null, "progress": 0, ' + '"results": null, "debug": null, "checks": null}, {"name": "Powerwall firmware", "status": "not_run", ' + '"start_time": null, "end_time": null, "progress": 0, "results": null, "debug": null, ' + '"checks": null}, {"name": "Battery firmware", "status": "not_run", "start_time": null, "end_time": ' + 'null, "progress": 0, "results": null, "debug": null, "checks": null}, {"name": "Inverter firmware", ' + '"status": "not_run", "start_time": null, "end_time": null, "progress": 0, "results": null, ' + '"debug": null, "checks": null}, {"name": "Grid code", "status": "not_run", "start_time": null, ' + '"end_time": null, "progress": 0, "results": null, "debug": null, "checks": null}], "alert": false}, ' + '"bc_type": null, "in_config": true}, {"Type": "", "PackagePartNumber": "3012170-05-B", ' + '"PackageSerialNumber": "TG1234567890G1", "type": "ACPW", "grid_state": "Grid_Uncompliant", ' + '"grid_reconnection_time_seconds": 0, "under_phase_detection": false, "updating": false, ' + '"commissioning_diagnostic": {"name": "Commissioning", "category": "InternalComms", "disruptive": ' + 'false, "inputs": null, "checks": [{"name": "CAN connectivity", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.320856307-08:00", "end_time": "2023-12-16T08:34:17.320940302-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}, {"name": "Enable switch", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.320949301-08:00", "end_time": "2023-12-16T08:34:17.320955301-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}, {"name": "Internal communications", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.320960676-08:00", "end_time": "2023-12-16T08:34:17.320966176-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}, {"name": "Firmware up-to-date", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.32097155-08:00", "end_time": "2023-12-16T08:34:17.3209768-08:00", "message": ' + '"Cannot perform this action with site controller running. From landing page, either \\"STOP ' + 'SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, "checks": null}], ' + '"alert": false}, "update_diagnostic": {"name": "Firmware Update", "category": "InternalComms", ' + '"disruptive": true, "inputs": null, "checks": [{"name": "Powerwall firmware", "status": "not_run", ' + '"start_time": null, "end_time": null, "progress": 0, "results": null, "debug": null, ' + '"checks": null}, {"name": "Battery firmware", "status": "not_run", "start_time": null, "end_time": ' + 'null, "progress": 0, "results": null, "debug": null, "checks": null}, {"name": "Inverter firmware", ' + '"status": "not_run", "start_time": null, "end_time": null, "progress": 0, "results": null, ' + '"debug": null, "checks": null}, {"name": "Grid code", "status": "not_run", "start_time": null, ' + '"end_time": null, "progress": 0, "results": null, "debug": null, "checks": null}], "alert": false}, ' + '"bc_type": null, "in_config": true}], "gateway_din": "1232100-00-E--TG1234567890G1", ' + '"sync": {"updating": false, "commissioning_diagnostic": {"name": "Commissioning", "category": ' + '"InternalComms", "disruptive": false, "inputs": null, "checks": [{"name": "CAN connectivity", ' + '"status": "fail", "start_time": "2023-12-16T08:34:17.321101293-08:00", "end_time": ' + '"2023-12-16T08:34:17.321107918-08:00", "message": "Cannot perform this action with site controller ' + 'running. From landing page, either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {' + '}, "debug": {}, "checks": null}, {"name": "Firmware up-to-date", "status": "fail", "start_time": ' + '"2023-12-16T08:34:17.321113792-08:00", "end_time": "2023-12-16T08:34:17.321118917-08:00", ' + '"message": "Cannot perform this action with site controller running. From landing page, ' + 'either \\"STOP SYSTEM\\" or \\"RUN WIZARD\\" to proceed.", "results": {}, "debug": {}, ' + '"checks": null}], "alert": false}, "update_diagnostic": {"name": "Firmware Update", "category": ' + '"InternalComms", "disruptive": true, "inputs": null, "checks": [{"name": "Synchronizer firmware", ' + '"status": "not_run", "start_time": null, "end_time": null, "progress": 0, "results": null, ' + '"debug": null, "checks": null}, {"name": "Islanding configuration", "status": "not_run", ' + '"start_time": null, "end_time": null, "progress": 0, "results": null, "debug": null, ' + '"checks": null}, {"name": "Grid code", "status": "not_run", "start_time": null, "end_time": null, ' + '"progress": 0, "results": null, "debug": null, "checks": null}], "alert": false}}, "msa": null, ' + '"states": null}') + +METERS_SITE = ('[{"id":0,"location":"site","type":"synchrometerX","cts":[true,true,false,false],' + '"inverted":[false,false,false,false],' + '"connection":{"short_id":"1232100-00-E--TG123456789E4G",' + '"device_serial":"JBL12345Y1F012synchrometerX","https_conf":{}},' + '"Cached_readings":{"last_communication_time":"2023-12-16T11:48:34.135766872-08:00",' + '"instant_power":2495,"instant_reactive_power":-212,' + '"instant_apparent_power":2503.9906149983867,"frequency":0,' + '"energy_exported":4507438.170261594,"energy_imported":6995047.554439916,' + '"instant_average_voltage":210.8945063295865,"instant_average_current":20.984,' + '"i_a_current":13.3045,"i_b_current":7.6795,"i_c_current":0,' + '"last_phase_voltage_communication_time":"2023-12-16T11:48:34.035339849-08:00",' + '"v_l1n":121.72,"v_l2n":121.78,' + '"last_phase_power_communication_time":"2023-12-16T11:48:34.135766872-08:00",' + '"real_power_a":1584,"real_power_b":911,"reactive_power_a":-129,"reactive_power_b":-83,' + '"last_phase_energy_communication_time":"0001-01-01T00:00:00Z",' + '"serial_number":"JBL12345Y1F012","version":"fa0c1ad02efda3","timeout":1500000000,' + '"instant_total_current":20.984}}]') + +SOLARS_BRANDS = ('["ABB","Ablerex Electronics","Advanced Energy Industries","Advanced Solar Photonics","AE Solar ' + 'Energy","AEconversion Gmbh","AEG Power Solutions","Aero-Sharp","Afore New Energy Technology Shanghai ' + 'Co","Agepower Limit","Alpha ESS Co","Alpha Technologies","Altenergy Power System","American Electric ' + 'Technologies","AMETEK Solidstate Control","Andalay Solar","Apparent","Asian Power Devices",' + '"AU Optronics","Auxin Solar","Ballard Power Systems","Beacon Power","Beijing Hua Xin Liu He ' + 'Investment (Australia) Pty","Beijing Kinglong New Energy","Bergey Windpower","Beyond Building ' + 'Group","Beyond Building Systems","BYD Auto Industry Company Limited","Canadian Solar",' + '"Carlo Gavazzi","CFM Equipment Distributors","Changzhou Nesl Solartech","Chiconypower","Chilicon",' + '"Chilicon Power","Chint Power Systems America","Chint Solar Zhejiang","Concept by US",' + '"Connect Renewable Energy","Danfoss","Danfoss Solar","Darfon Electronics","DASS tech","Delta Energy ' + 'Systems","Destin Power","Diehl AKO Stiftung","Diehl AKO Stiftung \u0026 KG","Direct Grid ' + 'Technologies","Dow Chemical","DYNAPOWER COMPANY","E-Village Solar","EAST GROUP CO LTD","Eaton",' + '"Eguana Technologies","Elettronica Santerno","Eltek","Emerson Network Power","Enecsys",' + '"Energy Storage Australia Pty","EnluxSolar","Enphase Energy","Eoplly New Energy Technology",' + '"EPC Power","ET Solar Industry","ETM Electromatic","Exeltech","Flextronics Industrial","Flextronics ' + 'International USA","Fronius","FSP Group","GAF","GE Energy","Gefran","Geoprotek","Global Mainstream ' + 'Dynamic Energy Technology","Green Power Technologies","GreenVolts","GridPoint","Growatt",' + '"Gsmart Ningbo Energy Storage Technology Co","Guangzhou Sanjing Electric Co","Hangzhou Sunny Energy ' + 'Science and Technology Co","Hansol Technics","Hanwha Q CELLS \u0026 Advanced Materials Corporation",' + '"Heart Transverter","Helios","HiQ Solar","HiSEL Power","Home Director","Hoymiles Converter ' + 'Technology","Huawei Technologies","Huawei Technologies Co","HYOSUNG","i-Energy Corporation",' + '"Ideal Power","Ideal Power Converters","IMEON ENERGY","Ingeteam","Involar","INVOLAR","INVT Solar ' + 'Technology Shenzhen Co","iPower","IST Energy","Jema Energy","Jiangsu GoodWe Power Supply Technology ' + 'Co","Jiangsu Weiheng Intelligent Technology Co","Jiangsu Zeversolar New Energy","Jiangsu Zeversolar ' + 'New Energy Co","Jiangyin Hareon Power","Jinko Solar","KACO","Kehua Hengsheng Co","Kostal Solar ' + 'Electric","LeadSolar Energy","Leatec Fine Ceramics","LG Electronics","Lixma Tech","Mage Solar",' + '"Mage Solar USA","Mariah Power","MIL-Systems","Ming Shen Energy Technology","Mohr Power",' + '"Motech Industries","NeoVolta","Nextronex Energy Systems","Nidec ASI","Ningbo Ginlong Technologies",' + '"Ningbo Ginlong Technologies Co","Northern Electric","ONE SUN MEXICO DE C.V.","Open Energy",' + '"OPTI International","OPTI-Solar","OutBack Power Technologies","Panasonic Corporation Eco Solutions ' + 'Company","Perfect Galaxy","Petra Solar","Petra Systems","Phoenixtec Power","Phono Solar Technology",' + '"Pika Energy","Power Electronics","Power-One","Powercom","PowerWave Energy Pty","Princeton Power ' + 'Systems","PurpleRubik New Energy Technology Co","PV Powered","Redback Technologies Limited",' + '"RedEarth Energy Storage Pty","REFU Elektronik","Renac Power Technology Co","Renergy",' + '"Renesola Zhejiang","Renovo Power Systems","Resonix","Rhombus Energy Solutions","Ritek Corporation",' + '"Sainty Solar","Samil Power","SanRex","SANYO","Sapphire Solar Pty","Satcon Technology",' + '"SatCon Technology","Schneider","Schneider Inverters USA","Schuco USA","Selectronic Australia",' + '"Senec GmbH","Shanghai Sermatec Energy Technology Co","Shanghai Trannergy Power Electronics Co",' + '"Sharp","Shenzhen BYD","Shenzhen Growatt","Shenzhen Growatt Co","Shenzhen INVT Electric Co",' + '"SHENZHEN KSTAR NEW ENERGY COMPANY LIMITED","Shenzhen Litto New Energy Co","Shenzhen Sinexcel ' + 'Electric","Shenzhen Sinexcel Electric Co","Shenzhen SOFARSOLAR Co","Siemens Industry",' + '"Silicon Energy","Sineng Electric Co","SMA","Sol-Ark","Solar Juice Pty","Solar Liberty",' + '"Solar Power","Solarbine","SolarBridge Technologies","SolarCity","SolarEdge Technologies",' + '"Solargate","Solaria Corporation","Solarmax","SolarWorld","SolaX Power Co","SolaX Power Network ' + 'Technology (Zhe jiang)","SolaX Power Network Technology Zhejiang Co","Solectria Renewables","Solis",' + '"Sonnen GmbH","Sonnetek","Southwest Windpower","Sparq Systems","Sputnik Engineering","STARFISH HERO ' + 'CO","Sungrow Power Supply","Sungrow Power Supply Co","Sunna Tech","SunPower","SunPower (Original ' + 'Mfr.Fronius)","Sunset","Sustainable Energy Technologies","Sustainable Solar Services",' + '"Suzhou Hypontech Co","Suzhou Solarwii Micro Grid Technology Co","Sysgration","Tabuchi Electric",' + '"Talesun Solar","Tesla","The Trustee for Soltaro Unit Trust","TMEIC","TOPPER SUN Energy Tech",' + '"Toshiba International","Trannergy","Trina Energy Storage Solutions (Jiangsu)","Trina Energy Storage ' + 'Solutions Jiangsu Co","Trina Solar Co","Ubiquiti Networks International","United Renewable Energy ' + 'Co","Westinghouse Solar","Windterra Systems","Xantrex Technology","Xiamen Kehua Hengsheng",' + '"Xiamen Kehua Hengsheng Co","Xslent Energy Technologies","Yaskawa Solectria Solar","Yes! Solar",' + '"Zhongli Talesun Solar","ZIGOR","シャープ (Sharp)","パナソニック (Panasonic)","三菱電機 (Mitsubishi)",' + '"京セラ (Kyocera)","東芝 (Toshiba)","長州産業 (Choshu Sangyou)","ｶﾅﾃﾞｨｱﾝ ｿｰﾗｰ"]') + +METERS = ('[{"serial":"VAH1234AB1234","short_id":"73533","type":"neurio_w2_tcp","connected":true,' + '"cts":[{"type":"solarRGM","valid":[true,false,false,false],"inverted":[false,false,false,false],' + '"real_power_scale_factor":2}],"ip_address":"PWRview-73533","mac":"01-23-45-56-78-90"},' + '{"serial":"JBL12345Y1F012synchrometerY","short_id":"1232100-00-E--TG123456789EGG","type":"synchrometerY"},' + '{"serial":"JBL12345Y1F012synchrometerX","short_id":"1232100-00-E--TG123456789EGG","type":"synchrometerX",' + '"cts":[{"type":"site","valid":[true,true,false,false],"inverted":[false,false,false,false]}]}]') + +INSTALLER = ('{"company":"Tesla","customer_id":"","phone":"","email":"","location":"","mounting":"","wiring":"",' + '"backup_configuration":"Whole Home","solar_installation":"New","solar_installation_type":"PV Panel",' + '"run_sitemaster":true,"verified_config":true,"installation_types":["Residential"]}') + +NETWORKS = ('[{"network_name":"ethernet_tesla_internal_default","interface":"EthType","enabled":true,"dhcp":true,' + '"extra_ips":[{"ip":"192.168.90.2","netmask":24}],"active":true,"primary":true,"lastTeslaConnected":true,' + '"lastInternetConnected":true,"iface_network_info":{"network_name":"ethernet_tesla_internal_default",' + '"ip_networks":[{"IP":"","Mask":"////AA=="}],"gateway":"","interface":"EthType",' + '"state":"DeviceStateReady","state_reason":"DeviceStateReasonNone","signal_strength":0,"hw_address":""}},' + '{"network_name":"gsm_tesla_internal_default","interface":"GsmType","enabled":true,"dhcp":null,' + '"active":true,"primary":false,"lastTeslaConnected":false,"lastInternetConnected":false,' + '"iface_network_info":{"network_name":"gsm_tesla_internal_default","ip_networks":[{"IP":"",' + '"Mask":"/////w=="}],"gateway":"","interface":"GsmType","state":"DeviceStateReady",' + '"state_reason":"DeviceStateReasonNone","signal_strength":71,"hw_address":""}}]') diff --git a/pypowerwall/fleetapi/pypowerwall_fleetapi.py b/pypowerwall/fleetapi/pypowerwall_fleetapi.py new file mode 100644 index 0000000..8f2f938 --- /dev/null +++ b/pypowerwall/fleetapi/pypowerwall_fleetapi.py @@ -0,0 +1,816 @@ +import json +import logging +import os +import time +from typing import Optional, Union, List + +from pypowerwall.fleetapi.fleetapi import FleetAPI, CONFIGFILE +from pypowerwall.fleetapi.decorators import not_implemented_mock_data +from pypowerwall.fleetapi.exceptions import * +from pypowerwall.fleetapi.mock_data import * +from pypowerwall.fleetapi.stubs import * +from pypowerwall.pypowerwall_base import PyPowerwallBase + +log = logging.getLogger(__name__) + +# Defaults +COUNTER_MAX = 64 # Max counter value for SITE_DATA API +SITE_CONFIG_TTL = 59 # Site config cache TTL in seconds + +fleet_api_urls = { + "North America, Asia-Pacific": "https://fleet-api.prd.na.vn.cloud.tesla.com", + "Europe, Middle East, Africa": "https://fleet-api.prd.eu.vn.cloud.tesla.com", + "China": "https://fleet-api.prd.cn.vn.cloud.tesla.cn" +} + + +def set_debug(debug=False, quiet=False, color=True): + logging.basicConfig(format='%(levelname)s: %(message)s') + if not quiet: + log.setLevel(logging.INFO) + if color: + logging.basicConfig(format='\x1b[31;1m%(levelname)s: %(message)s\x1b[0m') + if debug: + log.setLevel(logging.DEBUG) + else: + log.setLevel(logging.NOTSET) + + +def lookup(data, keylist): + """ + Lookup a value in a nested dictionary or return None if not found. + data - nested dictionary + keylist - list of keys to traverse + """ + if len(keylist) == 1: + return data.get(keylist[0]) + for key in keylist: + if key in data: + data = data[key] + else: + return None + return data + + +# noinspection PyMethodMayBeStatic +class PyPowerwallFleetAPI(PyPowerwallBase): + def __init__(self, email: Optional[str], pwcacheexpire: int = 5, timeout: int = 5, siteid: Optional[int] = None, + authpath: str = ""): + super().__init__(email) + self.fleet = None + self.apilock = {} # holds lock flag for pending api requests + self.siteindex = 0 # site index to use + self.siteid = siteid # site id to use + self.counter = 0 # counter for SITE_DATA API + self.timeout = timeout + self.poll_api_map = self.init_poll_api_map() + self.post_api_map = self.init_post_api_map() + self.authpath = authpath + self.configfile = os.path.join(self.authpath, CONFIGFILE) + self.auth = {'AuthCookie': 'local', 'UserRecord': 'local'} # Bogus local auth record + + # Initialize FleetAPI + self.fleet = FleetAPI(configfile=self.configfile, site_id=self.siteid, + pwcacheexpire=pwcacheexpire, timeout=self.timeout) + + # Load Configuration + if not os.path.isfile(self.fleet.configfile): + log.debug(f" -- fleetapi: Configuration file not found: {self.configfile} - run setup") + + # Set siteid + if self.siteid is None: + self.siteid = self.fleet.site_id + + log.debug(f" -- fleetapi: Using site {self.siteid} for {self.email}") + + def init_post_api_map(self) -> dict: + return { + "/api/operation": self.post_api_operation, + } + + def init_poll_api_map(self) -> dict: + # API map for local to cloud call conversion + return { + # Somewhat Real Actions + "/api/devices/vitals": self.get_api_devices_vitals, + "/api/meters/aggregates": self.get_api_meters_aggregates, + "/api/operation": self.get_api_operation, + "/api/site_info": self.get_api_site_info, + "/api/site_info/site_name": self.get_api_site_info_site_name, + "/api/status": self.get_api_status, + "/api/system_status": self.get_api_system_status, + "/api/system_status/grid_status": self.get_api_system_status_grid_status, + "/api/system_status/soe": self.get_api_system_status_soe, + "/vitals": self.get_vitals, + # Possible Actions + "/api/login/Basic": self.api_login_basic, + "/api/logout": self.api_logout, + # Mock Actions + "/api/auth/toggle/supported": self.get_api_auth_toggle_supported, + "/api/customer": self.get_api_customer, + "/api/customer/registration": self.get_api_customer_registration, + "/api/installer": self.get_api_installer, + "/api/meters": self.get_api_meters, + "/api/meters/readings": self.get_api_unimplemented_timeout, + "/api/meters/site": self.get_api_meters_site, + "/api/meters/solar": self.get_unimplemented_api, + "/api/networks": self.get_api_unimplemented_timeout, + "/api/powerwalls": self.get_api_powerwalls, + "/api/site_info/grid_codes": self.get_api_unimplemented_timeout, + "/api/sitemaster": self.get_api_sitemaster, + "/api/solar_powerwall": self.get_api_solar_powerwall, + "/api/solars": self.get_api_solars, + "/api/solars/brands": self.get_api_solars_brands, + "/api/synchrometer/ct_voltage_references": self.get_api_synchrometer_ct_voltage_references, + "/api/system/update/status": self.get_api_system_update_status, + "/api/system_status/grid_faults": self.get_api_system_status_grid_faults, + "/api/troubleshooting/problems": self.get_api_troubleshooting_problems, + } + + def authenticate(self): + log.debug('Tesla fleetapi mode enabled') + # Check to see if we can connect to the cloud + if not self.connect(): + err = "Unable to connect to Tesla FleetAPI - run pypowerwall fleetapi setup" + log.debug(err) + raise ConnectionError(err) + + def connect(self): + """ + Connect to Tesla FleetAPI + """ + # Get site info + sites = self.getsites() + self.siteindex = 0 + if sites is None or len(sites) == 0: + log.error("No sites found for %s" % self.email) + return False + # Find siteindex - Lookup energy_site_id in sites + if self.siteid is None: + self.siteid = sites[0]['energy_site_id'] # default to first site + self.siteindex = 0 + else: + found = False + for idx, site in enumerate(sites): + if site['energy_site_id'] == self.siteid: + self.siteindex = idx + found = True + break + if not found: + log.error("Site %r not found for %s" % (self.siteid, self.email)) + return False + # Set site + self.siteid = sites[self.siteindex].get('energy_site_id') + log.debug(f"Connected to Tesla FleetAPI - Site {self.siteid} " + f"({sites[self.siteindex]['site_name']}) for {self.email}") + return True + + # Function to map Powerwall API to Tesla FleetAPI Data + def poll(self, api: str, force: bool = False, + recursive: bool = False, raw: bool = False) -> Optional[Union[dict, list, str, bytes]]: + """ + Map Powerwall API to Tesla FleetAPI Data + """ + # API Map - Determine what data we need based on Powerwall APIs + log.debug(f" -- fleetapi: Request for {api}") + + func = self.poll_api_map.get(api) + if func: + kwargs = { + 'force': force, + 'recursive': recursive, + 'raw': raw + } + return func(**kwargs) + else: + return {"ERROR": f"Unknown API: {api}"} + + def post(self, api: str, payload: Optional[dict], din: Optional[str], + recursive: bool = False, raw: bool = False) -> Optional[Union[dict, list, str, bytes]]: + """ + Map Powerwall API to Tesla FleetAPI Data + """ + # API Map - Determine what data we need based on Powerwall APIs + log.debug(f" -- fleetapi: Request for {api}") + + func = self.post_api_map.get(api) + if func: + kwargs = { + 'payload': payload, + 'din': din + } + res = func(**kwargs) + if res: + # invalidate appropriate read cache on (more or less) successful call to writable API + super()._invalidate_cache(api) + return res + else: + # raise PyPowerwallFleetAPINotImplemented(api) + # or pass a custom error response: + return {"ERROR": f"Unknown API: {api}"} + + def getsites(self): + """ + Get list of Tesla Energy sites + """ + if self.siteid is None: + return None + try: + sitelist = self.fleet.getsites() + except Exception as err: + log.error(f"Failed to retrieve sitelist - {repr(err)}") + return None + return sitelist + + def change_site(self, siteid): + """ + Change the site to the one that matches the siteid + """ + # Check that siteid is a valid number + try: + siteid = int(siteid) + except Exception as err: + log.error("Invalid siteid - %s" % repr(err)) + return False + # Check for valid site index + sites = self.getsites() + if sites is None or len(sites) == 0: + log.error("No sites found for %s" % self.email) + return False + # Set siteindex - Find siteid in sites + for idx, site in enumerate(sites): + if site['energy_site_id'] == siteid: + self.siteid = siteid + self.siteindex = idx + self.siteid = sites[self.siteindex] + log.debug(f"Changed site to {self.siteid} ({sites[self.siteindex]['site_name']}) for {self.email}") + return True + log.error("Site %d not found for %s" % (siteid, self.email)) + return False + + # FleetAPI Functions + def get_site_info(self): + """ + { + 'id': '1234000-00-E--TG12345678904G', + 'site_name': 'TeslaEnergyGateway', + 'backup_reserve_percent': 20, + 'default_real_mode': 'self_consumption', + 'installation_date': '2021-09-25T15: 53: 47-07: 00', + 'user_settings': { + 'go_off_grid_test_banner_enabled': False, + 'storm_mode_enabled': False, + 'powerwall_onboarding_settings_set': True, + 'powerwall_tesla_electric_interested_in': False, + 'vpp_tour_enabled': True, + 'sync_grid_alert_enabled': True, + 'breaker_alert_enabled': False + }, + 'components': { + 'solar': True, + 'solar_type': 'pv_panel', + 'battery': True, + 'grid': True, + 'backup': True, + 'gateway': 'teg', + 'load_meter': True, + 'tou_capable': True, + 'storm_mode_capable': True, + 'flex_energy_request_capable': False, + 'car_charging_data_supported': False, + 'off_grid_vehicle_charging_reserve_supported': True, + 'vehicle_charging_performance_view_enabled': False, + 'vehicle_charging_solar_offset_view_enabled': False, + 'battery_solar_offset_view_enabled': True, + 'solar_value_enabled': True, + 'energy_value_header': 'EnergyValue', + 'energy_value_subheader': 'EstimatedValue', + 'energy_service_self_scheduling_enabled': True, + 'show_grid_import_battery_source_cards': True, + 'set_islanding_mode_enabled': True, + 'wifi_commissioning_enabled': True, + 'backup_time_remaining_enabled': True, + 'battery_type': 'solar_powerwall', + 'configurable': True, + 'grid_services_enabled': False, + 'gateways': [ + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '1232100-00-E--TG12345678904G', + 'serial_number': 'TG12345678904G', + 'part_number': '1232100-00-E', + 'part_type': 10, + 'part_name': 'TeslaBackupGateway2', + 'is_active': True, + 'site_id': 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxx', + 'firmware_version': '24.4.00fe780c9', + 'updated_datetime': '2024-05-11T09: 20: 26.225Z' + } + ], + 'batteries': [ + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '2012170-25-E--TG12345678904G', + 'serial_number': 'TG12345678904G', + 'part_number': '2012170-25-E', + 'part_type': 2, + 'part_name': 'Powerwall2', + 'nameplate_max_charge_power': 5400, + 'nameplate_max_discharge_power': 5400, + 'nameplate_energy': 13500 + }, + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '3012170-05-B--TG12345678904G', + 'serial_number': 'TG12345678904G', + 'part_number': '3012170-05-B', + 'part_type': 2, + 'part_name': 'Powerwall2', + 'nameplate_max_charge_power': 5400, + 'nameplate_max_discharge_power': 5400, + 'nameplate_energy': 13500 + } + ], + 'inverters': [ + { + 'device_id': 'xxxxxxxx-xxxxx-xxx-xxxx-xxxxxxxxxxxx', + 'din': '1530000-00-F--CN12345678901J', + 'part_number': '1538100-00-F', + 'part_type': 7, + 'part_name': 'Non-TeslaInverter', + 'is_active': True, + 'site_id': 'xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxxx', + } + ], + 'edit_setting_permission_to_export': True, + 'edit_setting_grid_charging': True, + 'edit_setting_energy_exports': True, + 'system_alerts_enabled': True + }, + 'version': '24.4.00fe780c9', + 'battery_count': 2, + 'tariff_content': {} + } + """ + return self.fleet.get_site_info() + + def get_live_status(self): + """ + { + 'solar_power': 0, + 'percentage_charged': 55.164177150990625, + 'backup_capable': True, + 'battery_power': 4080, + 'load_power': 4080, + 'grid_status': 'Active', + 'grid_services_active': False, + 'grid_power': 0, + 'grid_services_power': 0, + 'generator_power': 0, + 'island_status': 'on_grid', + 'storm_mode_active': False, + 'timestamp': '2024-05-11T22:43:20-07:00', + 'wall_connectors': [] + } + """ + return self.fleet.get_live_status() + + def get_time_remaining(self, force: bool = False) -> Optional[float]: + """ + Get backup time remaining from Tesla FleetAPI + TODO + """ + return self.fleet.get_backup_time_remaining() + + def get_api_system_status_soe(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + percentage_charged = self.fleet.battery_level(force=force) or 0 + data = { + "percentage": percentage_charged + } + return data + + def get_api_status(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + config = self.fleet.get_site_info(force=force) + if config is None: + data = None + else: + data = { + "din": config.get("id"), # 1232100-00-E--TGxxxxxxxxxxxx + "start_time": config.get("installation_date"), # "2023-10-13 04:01:45 +0800" + "up_time_seconds": None, # "1541h38m20.998412744s" + "is_new": False, + "version": config.get("version"), # 23.28.2 27626f98 + "git_hash": "27626f98a66cad5c665bbe1d4d788cdb3e94fd34", + "commission_count": 0, + "device_type": lookup(config, ("components", "gateway")), # teg + "teg_type": "unknown", + "sync_type": "v2.1", + "cellular_disabled": False, + "can_reboot": True + } + return data + + def get_api_system_status_grid_status(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + power = self.fleet.get_live_status(force=force) + if power is None: + data = None + else: + if power.get("grid_status") == "Active": + grid_status = "SystemGridConnected" + else: # off_grid or off_grid_unintentional + grid_status = "SystemIslandedActive" + data = { + "grid_status": grid_status, # SystemIslandedActive or SystemTransitionToGrid + "grid_services_active": power.get("grid_services_active") + # true when participating in VPP event + } + return data + + def get_api_site_info_site_name(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + config = self.fleet.get_site_info(force=force) + if config is None: + data = None + else: + sitename = config.get("site_name") + tz = config.get("installation_time_zone") + data = { + "site_name": sitename, + "timezone": tz + } + return data + + def get_api_site_info(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + config = self.fleet.get_site_info(force=force) + if config is None: + data = None + else: + nameplate_power = int(config.get("nameplate_power") or 0) / 1000 + nameplate_energy = int(config.get("nameplate_energy") or 0) / 1000 + max_site_meter_power_ac = config.get("max_site_meter_power_ac") + min_site_meter_power_ac = config.get("min_site_meter_power_ac") + utility = config.get("tariff_content", {}).get("utility") + sitename = config.get("site_name") + tz = config.get("installation_time_zone") + data = { + "max_system_energy_kWh": nameplate_energy, + "max_system_power_kW": nameplate_power, + "site_name": sitename, + "timezone": tz, + "max_site_meter_power_kW": max_site_meter_power_ac, + "min_site_meter_power_kW": min_site_meter_power_ac, + "nominal_system_energy_kWh": nameplate_energy, + "nominal_system_power_kW": nameplate_power, + "panel_max_current": None, + "grid_code": { + "grid_code": None, + "grid_voltage_setting": None, + "grid_freq_setting": None, + "grid_phase_setting": None, + "country": None, + "state": None, + "utility": utility + } + } + return data + + # noinspection PyUnusedLocal + def get_api_devices_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + # Protobuf payload - not implemented - use /vitals instead + data = None + log.warning("Protobuf payload - not implemented for /api/devices/vitals - use /vitals instead") + return data + + def get_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + # Simulated Vitals + force = kwargs.get('force', False) + config = self.fleet.get_site_info(force=force) + power = self.fleet.get_live_status(force=force) + if config is None or power is None: + data = None + else: + din = config.get("id") + parts = din.split("--") + if len(parts) == 2: + part_number = parts[0] + serial_number = parts[1] + else: + part_number = None + serial_number = None + version = config.get("version") + # Get grid status + # also "grid_status": "Active" + island_status = power.get("island_status") + if island_status == "on_grid": + alert = "SystemConnectedToGrid" + elif island_status == "off_grid_intentional": + alert = "ScheduledIslandContactorOpen" + elif island_status == "off_grid": + alert = "UnscheduledIslandContactorOpen" + else: + alert = "" + if power.get("grid_status") == "Active": + alert = "SystemConnectedToGrid" + data = { + f'STSTSM--{part_number}--{serial_number}': { + 'partNumber': part_number, + 'serialNumber': serial_number, + 'manufacturer': 'Simulated', + 'firmwareVersion': version, + 'lastCommunicationTime': int(time.time()), + 'teslaEnergyEcuAttributes': { + 'ecuType': 207 + }, + 'STSTSM-Location': 'Simulated', + 'alerts': [ + alert + ] + } + } + return data + + def get_api_meters_aggregates(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + config = self.fleet.get_site_info(force=force) + power = self.fleet.get_live_status(force=force) + if config is None or power is None: + data = None + else: + timestamp = power.get("timestamp") + solar_power = power.get("solar_power") + battery_power = power.get("battery_power") + load_power = power.get("load_power") + grid_power = power.get("grid_power") + battery_count = config.get("battery_count") + inverters = lookup(config, ("components", "inverters")) + + if inverters is not None: + solar_inverters = len(inverters) + elif lookup(config, ("components", "solar")): + solar_inverters = 1 + else: + solar_inverters = 0 + data = API_METERS_AGGREGATES_STUB + data['site'].update({ + "last_communication_time": timestamp, + "instant_power": grid_power, + }) + data['battery'].update({ + "last_communication_time": timestamp, + "instant_power": battery_power, + "num_meters_aggregated": battery_count, + }) + data['load'].update({ + "last_communication_time": timestamp, + "instant_power": load_power, + + }) + data['solar'].update({ + "last_communication_time": timestamp, + "instant_power": solar_power, + "num_meters_aggregated": solar_inverters, + }) + return data + + def get_api_operation(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + config = self.fleet.get_site_info(force=force) + if config is None: + data = None + else: + default_real_mode = config.get("default_real_mode") + backup_reserve_percent = config.get("backup_reserve_percent") or 0 + backup = (backup_reserve_percent + (5 / 0.95)) * 0.95 + data = { + "real_mode": default_real_mode, + "backup_reserve_percent": backup + } + return data + + def get_api_system_status(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + force = kwargs.get('force', False) + power = self.fleet.get_live_status(force=force) + config = self.fleet.get_site_info(force=force) + if power is None or config is None: + data = None + else: + solar_power = power.get("solar_power") + grid_services_power = power.get("grid_services_power") + battery_count = config.get("battery_count") + total_pack_energy = self.fleet.total_pack_energy(force=force) + energy_left = self.fleet.energy_left(force=force) + nameplate_power = config.get("nameplate_power") + + if power.get("island_status") == "on_grid": + grid_status = "SystemGridConnected" + else: # off_grid or off_grid_unintentional + grid_status = "SystemIslandedActive" + # "grid_status": "Active" + if power.get("grid_status") == "Active": + grid_status = "SystemGridConnected" + data = API_SYSTEM_STATUS_STUB # TODO: see inside API_SYSTEM_STATUS_STUB definition + data.update({ + "nominal_full_pack_energy": total_pack_energy, + "nominal_energy_remaining": energy_left, + "max_charge_power": nameplate_power, + "max_discharge_power": nameplate_power, + "max_apparent_power": nameplate_power, + "grid_services_power": grid_services_power, + "system_island_state": grid_status, + "available_blocks": battery_count, + "solar_real_power_limit": solar_power, + "blocks_controlled": battery_count, + }) + + return data + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def api_logout(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return {"status": "ok"} + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def api_login_basic(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return {"status": "ok"} + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_meters_site(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads(METERS_SITE) + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_unimplemented_api(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return None + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_unimplemented_timeout(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return "TIMEOUT!" + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_auth_toggle_supported(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return {"toggle_auth_supported": True} + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_sitemaster(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return {"status": "StatusUp", "running": True, "connected_to_tesla": True, "power_supply_mode": False, + "can_reboot": "Yes"} + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_powerwalls(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads(POWERWALLS) + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_customer_registration(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads('{"privacy_notice":null,"limited_warranty":null,"grid_services":null,"marketing":null,' + '"registered":true,"timed_out_registration":false}') + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_system_update_status(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads('{"state":"/update_succeeded","info":{"status":["nonactionable"]},' + '"current_time":1702756114429,"last_status_time":1702753309227,"version":"23.28.2 27626f98",' + '"offline_updating":false,"offline_update_error":"","estimated_bytes_per_second":null}') + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_system_status_grid_faults(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads('[]') + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_solars(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads('[{"brand":"Tesla","model":"Solar Inverter 7.6","power_rating_watts":7600}]') + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_solars_brands(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads(SOLARS_BRANDS) + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_customer(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return {"registered": True} + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_meters(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads(METERS) + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_installer(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads(INSTALLER) + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_synchrometer_ct_voltage_references(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads('{"ct1":"Phase1","ct2":"Phase2","ct3":"Phase1"}') + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_troubleshooting_problems(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return json.loads('{"problems":[]}') + + # noinspection PyUnusedLocal + @not_implemented_mock_data + def get_api_solar_powerwall(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]: + return {} + + def setup(self, email=None): + """ + Set up the Tesla FleetAPI connection + """ + return self.fleet.setup() + + def close_session(self): + return True + + def vitals(self) -> Optional[dict]: + return self.poll('/vitals') + + def post_api_operation(self, **kwargs): + payload = kwargs.get('payload', {}) + din = kwargs.get('din') + resp = {} + + if not payload.get('backup_reserve_percent') and not payload.get('real_mode'): + raise PyPowerwallFleetAPIInvalidPayload("/api/operation payload missing required parameters. Either " + "'backup_reserve_percent or 'real_mode', or both must present.") + + if din: + log.warning("FleetAPI mode operates on entire site, not din. Ignoring din parameter.") + + if payload.get('backup_reserve_percent') is not None: + backup_reserve_percent = payload['backup_reserve_percent'] + if backup_reserve_percent == False: + backup_reserve_percent = 0 + op_level = self.fleet.set_battery_reserve(backup_reserve_percent) + resp['set_backup_reserve_percent'] = { + 'backup_reserve_percent': payload['backup_reserve_percent'], + 'result': op_level + } + if payload.get('real_mode') is not None: + real_mode = payload['real_mode'] + op_mode = self.fleet.set_operating_mode(real_mode) + resp['set_operation'] = { + 'real_mode': payload['real_mode'], + 'result': op_mode + } + return resp + + +if __name__ == "__main__": + set_debug(quiet=False, debug=True, color=True) + + fleet = PyPowerwallFleetAPI() + + if not fleet.connect(): + log.info("Failed to connect to Tesla FleetAPI") + fleet.setup() + if not fleet.connect(): + log.critical("Failed to connect to Tesla FleetAPI") + exit(1) + + log.info("Connected to Tesla FleetAPI") + + log.info("Site Data") + tsites = fleet.getsites() + log.info(tsites) + + # print("\Battery") + # r = fleet.get_battery() + # print(r) + + # print("\Site Power") + # r = fleet.get_site_power() + # print(r) + + # print("\Site Config") + # r = fleet.get_site_config() + # print(r) + + # Test Poll + # '/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'] + 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/synchrometer/ct_voltage_references', + '/vitals'] + for i in items: + log.info(f"poll({i}):") + log.info(fleet.poll(i)) diff --git a/pypowerwall/fleetapi/stubs.py b/pypowerwall/fleetapi/stubs.py new file mode 100644 index 0000000..f18ca2e --- /dev/null +++ b/pypowerwall/fleetapi/stubs.py @@ -0,0 +1,121 @@ +API_METERS_AGGREGATES_STUB = { + "site": { + "last_communication_time": None, + "instant_power": None, + "instant_reactive_power": 0, + "instant_apparent_power": 0, + "frequency": 0, + "energy_exported": 0, + "energy_imported": 0, + "instant_average_voltage": 0, + "instant_average_current": 0, + "i_a_current": 0, + "i_b_current": 0, + "i_c_current": 0, + "last_phase_voltage_communication_time": "0001-01-01T00:00:00Z", + "last_phase_power_communication_time": "0001-01-01T00:00:00Z", + "last_phase_energy_communication_time": "0001-01-01T00:00:00Z", + "timeout": 1500000000, + "num_meters_aggregated": 1, + "instant_total_current": None + }, + "battery": { + "last_communication_time": None, + "instant_power": None, + "instant_reactive_power": 0, + "instant_apparent_power": 0, + "frequency": 0, + "energy_exported": 0, + "energy_imported": 0, + "instant_average_voltage": 0, + "instant_average_current": 0, + "i_a_current": 0, + "i_b_current": 0, + "i_c_current": 0, + "last_phase_voltage_communication_time": "0001-01-01T00:00:00Z", + "last_phase_power_communication_time": "0001-01-01T00:00:00Z", + "last_phase_energy_communication_time": "0001-01-01T00:00:00Z", + "timeout": 1500000000, + "num_meters_aggregated": None, + "instant_total_current": 0 + }, + "load": { + "last_communication_time": None, + "instant_power": None, + "instant_reactive_power": 0, + "instant_apparent_power": 0, + "frequency": 0, + "energy_exported": 0, + "energy_imported": 0, + "instant_average_voltage": 0, + "instant_average_current": 0, + "i_a_current": 0, + "i_b_current": 0, + "i_c_current": 0, + "last_phase_voltage_communication_time": "0001-01-01T00:00:00Z", + "last_phase_power_communication_time": "0001-01-01T00:00:00Z", + "last_phase_energy_communication_time": "0001-01-01T00:00:00Z", + "timeout": 1500000000, + "instant_total_current": 0 + }, + "solar": { + "last_communication_time": None, + "instant_power": None, + "instant_reactive_power": 0, + "instant_apparent_power": 0, + "frequency": 0, + "energy_exported": 0, + "energy_imported": 0, + "instant_average_voltage": 0, + "instant_average_current": 0, + "i_a_current": 0, + "i_b_current": 0, + "i_c_current": 0, + "last_phase_voltage_communication_time": "0001-01-01T00:00:00Z", + "last_phase_power_communication_time": "0001-01-01T00:00:00Z", + "last_phase_energy_communication_time": "0001-01-01T00:00:00Z", + "timeout": 1000000000, + "num_meters_aggregated": None, + "instant_total_current": 0 + } +} + +API_SYSTEM_STATUS_STUB = { # TODO: Fill in 0 values + "command_source": "Configuration", + "battery_target_power": 0, + "battery_target_reactive_power": 0, + "nominal_full_pack_energy": None, + "nominal_energy_remaining": None, + "max_power_energy_remaining": 0, # TODO: Calculate + "max_power_energy_to_be_charged": 0, # TODO: Calculate + "max_charge_power": None, + "max_discharge_power": None, + "max_apparent_power": None, + "instantaneous_max_discharge_power": 0, + "instantaneous_max_charge_power": 0, + "instantaneous_max_apparent_power": 0, + "hardware_capability_charge_power": 0, + "hardware_capability_discharge_power": 0, + "grid_services_power": None, + "system_island_state": None, + "available_blocks": None, + "available_charger_blocks": 0, + "battery_blocks": [], # TODO: Populate with battery blocks + "ffr_power_availability_high": 0, + "ffr_power_availability_low": 0, + "load_charge_constraint": 0, + "max_sustained_ramp_rate": 0, + "grid_faults": [], # TODO: Populate with grid faults + "can_reboot": "Yes", + "smart_inv_delta_p": 0, + "smart_inv_delta_q": 0, + "last_toggle_timestamp": "2023-10-13T04:08:05.957195-07:00", + "solar_real_power_limit": None, + "score": 10000, + "blocks_controlled": None, + "primary": True, + "auxiliary_load": 0, + "all_enable_lines_high": True, + "inverter_nominal_usable_power": 0, + "expected_energy_remaining": 0 +} diff --git a/pypowerwall/local/pypowerwall_local.py b/pypowerwall/local/pypowerwall_local.py index 2d8ec37..532de4c 100644 --- a/pypowerwall/local/pypowerwall_local.py +++ b/pypowerwall/local/pypowerwall_local.py @@ -30,6 +30,7 @@ def __init__(self, host: str, password: str, email: str, timezone: str, timeout: self.session = None self.pwcachetime = {} # holds the cached data timestamps for api self.pwcacheexpire = pwcacheexpire # seconds to expire cache + self.pwcache = {} # holds the cached data for api self.pwcooldown = 0 # rate limit cooldown time - pause api calls self.vitals_api = True # vitals api is available for local mode diff --git a/pypowerwall/aux.py b/pypowerwall/regex.py similarity index 100% rename from pypowerwall/aux.py rename to pypowerwall/regex.py diff --git a/tools/fleetapi/README.md b/tools/fleetapi/README.md index 4375494..198d3b1 100644 --- a/tools/fleetapi/README.md +++ b/tools/fleetapi/README.md @@ -16,7 +16,7 @@ Step 1 - Sign in to Tesla Developer Portal and make an App Access Request: See [ * CLIENT_ID - This will be provided to you by Tesla when your request is approved. * CLIENT_SECRET - Same as above. * DOMAIN - The domain name of a website your own and control. -* REDIRECT_URI - This is the URL that Tesla will direct you to after you authenticate. This landing URL (on your website) will extract the GET variable `code`, which is a one-time use authorization code needed to generate the Bearer auth and Refresh token used to access your Tesla Powerwall energy devices. Place the [index.html](./index.html) file in a folder under this domain and use this as the REDIRECT_URI path in the setup below. Alternatively, you can just copy the URL from the 404 page during the authorzation process (the code is in the URL). +* REDIRECT_URI - This is the URL that Tesla will direct you to after you authenticate. This landing URL (on your website) will extract the GET variable `code`, which is a one-time use authorization code needed to generate the Bearer auth and Refresh token used to access your Tesla Powerwall energy devices. Place the [index.html](./index.html) file in a folder under this domain and use this as the REDIRECT_URI path in the setup below. Alternatively, you can just copy the URL from the 404 page during the authorization process (the code is in the URL). Step 2 - Run the `create_pem_key.py` script and place the **public** key on your website at the URL: https://{DOMAIN}/.well-known/appspecific/com.tesla.3p.public-key.pem