diff --git a/.github/workflows/simtest.yml b/.github/workflows/simtest.yml index 8ef8963..786ea8a 100644 --- a/.github/workflows/simtest.yml +++ b/.github/workflows/simtest.yml @@ -14,7 +14,7 @@ jobs: strategy: matrix: - python-version: ["3.5", "3.6", "3.7", "3.8", "3.9", "3.10", "3.11"] + python-version: ["3.6", "3.7", "3.8", "3.9", "3.10", "3.11"] services: simulator: @@ -33,7 +33,7 @@ jobs: python -VV python -m site python -m pip install --upgrade pip setuptools wheel - pip install --upgrade requests protobuf + pip install --upgrade requests protobuf teslapy - name: "Run test.py on ${{ matrix.python-version }}" run: "python example.py" diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index f6d57c8..01b25cf 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -14,7 +14,7 @@ jobs: strategy: matrix: - python-version: ["3.5", "3.6", "3.7", "3.8", "3.9", "3.10", "3.11"] + python-version: ["3.6", "3.7", "3.8", "3.9", "3.10", "3.11"] steps: - uses: "actions/checkout@v2" @@ -27,7 +27,7 @@ jobs: python -VV python -m site python -m pip install --upgrade pip setuptools wheel - pip install --upgrade requests protobuf + pip install --upgrade requests protobuf teslapy - name: "Run test.py on ${{ matrix.python-version }}" run: "python test.py" diff --git a/.gitignore b/.gitignore index 5e71af7..6de9584 100644 --- a/.gitignore +++ b/.gitignore @@ -144,3 +144,7 @@ tools/set-mode.auth tools/set-mode.conf tools/tedapi/request.bin tools/tedapi/app* +.pypowerwall.auth +.pypowerwall.site +proxy/pypowerwall +proxy/teslapy diff --git a/README.md b/README.md index 414a3c1..c747066 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ [![Python Version](https://img.shields.io/pypi/pyversions/pypowerwall)](https://img.shields.io/pypi/pyversions/pypowerwall) [![PyPI Downloads](https://static.pepy.tech/badge/pypowerwall/month)](https://static.pepy.tech/badge/pypowerwall/month) -Python module to interface with Tesla Energy Gateways for Powerwall and solar power data. Currently supporting Powerwall, Powerwall 2 and Powerwall+ systems. +Python module to interface with Tesla Energy Gateways for Powerwall and solar power data. Currently supporting local access to Powerwall, Powerwall 2 and Powerwall+ systems and Tesla Owner cloud API for all systems including Solar Only and Powerwall 3 systems. ## Description @@ -33,10 +33,13 @@ You can clone this repo or install the package with pip. Once installed, pyPowe ```bash # Install pyPowerwall -python -m pip install pypowerwall +python3 -m pip install pypowerwall # Scan Network for Powerwalls -python -m pypowerwall scan +python3 -m pypowerwall scan + +# (optional) Setup to use Tesla Owners cloud API +python3 -m pypowerwall setup ``` FreeBSD users can install from ports or pkg [FreshPorts](https://www.freshports.org/net-mgmt/py-pypowerwall): @@ -51,7 +54,7 @@ Via ports: # cd /usr/ports/net-mgmt/py-pypowerwall/ && make install clean ``` -Note: pyPowerwall installation will attempt to install these required python packages: _requests_ and _protobuf_. +Note: pyPowerwall installation will attempt to install these required python packages: _requests_, _protobuf_ and _teslapy_. ## Programming with pyPowerwall @@ -64,11 +67,17 @@ and call function to poll data. Here is an example: # Optional: Turn on Debug Mode # pypowerwall.set_debug(True) - # Credentials for your Powerwall - Customer Login Data + # 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 = "" + email='email@example.com' + host = "" + timezone = "America/Los_Angeles" # Your local timezone # Connect to Powerwall pw = pypowerwall.Powerwall(host,password,email,timezone) @@ -114,6 +123,17 @@ and call function to poll data. Here is an example: Classes Powerwall(host, password, email, timezone, pwcacheexpire, timeout, poolmaxsize) + Parameters + host # Hostname or IP of the Tesla gateway + password # Customer password for gateway + email # (required) Customer email for gateway / cloud + timezone # Desired timezone + pwcacheexpire = 5 # Set API cache timeout in seconds + timeout = 5 # Timeout for HTTPS calls in seconds + 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) + Functions poll(api, json, force) # Return data from Powerwall api (dict if json=True, bypass cache force=True) level() # Return battery power level percentage @@ -134,20 +154,16 @@ 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 and login to Powerwall + is_connected() # Returns True if able to connect to Powerwall + get_reserve(scale) # Get Battery Reserve Percentage + get_time_remaining() # Get the backup time remaining on the battery - Parameters - host # (required) hostname or IP of the Tesla gateway - password # (required) password for logging into the gateway - email # (required) email used for logging into the gateway - timezone # (required) desired timezone - pwcacheexpire = 5 # Set API cache timeout in seconds - timeout = 10 # Timeout for HTTPS calls in seconds - poolmaxsize = 10 # Pool max size for http connection re-use (persistent connections disabled if zero) ``` ## Tools diff --git a/RELEASE.md b/RELEASE.md index 5b535b0..39da526 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -1,9 +1,34 @@ # RELEASE NOTES +## v0.7.1 - Tesla Cloud Mode + +* Simulate Powerwall Energy Gateway via Tesla Cloud API calls. In `cloudmode` API calls to pypowerwall APIs will result in calls made to the Tesla API to fetch the data. + +Cloud Mode Setup - Use pypowerwall to fetch your Tesla Owners API Token + +```bash +python3 -m pypowerwall setup + +# Token and site information stored in .pypowerwall.auth and .pypowerwall.site +``` + +Cloud Mode Code Example + +```python +import pypowerwall +pw = pypowerwall.Powerwall(email="email@example.com",cloudmode=True) +pw.power() +# Output: {'site': 2977, 'solar': 1820, 'battery': -3860, 'load': 937} +pw.poll('/api/system_status/soe') +# Output: '{"percentage": 26.403205103271222}' +``` + +* Added new API function to compute estimated backup time remaining on the battery: `get_time_remaining()` + ## v0.6.4 - Power Flow Animation Proxy t29 Updates -* Default page rendered by proxy (http://pypowerwall/) will render Powerflow Animation +* Default page rendered by proxy (http://pypowerwall:8675/) will render Powerflow Animation * Animation assets (html, css, js, images, fonts, svg) will render from local filesystem instead of pulling from Powerwall TEG portal. * Start prep for possible API removals from Powerwall TEG portal (see NOAPI settings) diff --git a/proxy/Dockerfile b/proxy/Dockerfile index 8bc2881..2131f73 100644 --- a/proxy/Dockerfile +++ b/proxy/Dockerfile @@ -1,6 +1,6 @@ FROM python:3.10-alpine WORKDIR /app -RUN pip3 install pypowerwall==0.6.4 bs4 +RUN pip3 install pypowerwall==0.7.1 bs4 COPY . . CMD ["python3", "server.py"] EXPOSE 8675 diff --git a/proxy/RELEASE.md b/proxy/RELEASE.md index 7d4580a..7443aa7 100644 --- a/proxy/RELEASE.md +++ b/proxy/RELEASE.md @@ -1,5 +1,13 @@ ## pyPowerwall Proxy Release Notes +### Proxy t35 (29 Dec 2023) + +* Add `cloudmode` support for pypowerwall v0.7.1. + +### Proxy t32 (20 Dec 2023) + +* Fix "flashing animation" problem by matching `hash` variable in index.html to firmware version `git_hash`. + ### Proxy t29 (16 Dec 2023) * Default page rendered by proxy (http://pypowerwall/) will render Powerflow Animation diff --git a/proxy/server-r.py b/proxy/server-r.py deleted file mode 100755 index 517d5ba..0000000 --- a/proxy/server-r.py +++ /dev/null @@ -1,342 +0,0 @@ -#!/usr/bin/env python -# pyPowerWall Module - Proxy Server Tool -# -*- coding: utf-8 -*- -""" - Python module to interface with Tesla Solar Powerwall Gateway - - Author: Jason A. Cox - For more information see https://github.com/jasonacox/pypowerwall - - Proxy Server Tool - This tool will proxy API calls to /api/meters/aggregates and - /api/system_status/soe - You can containerize it and run it as - an endpoint for tools like telegraf to pull metrics. - - This proxy also supports pyPowerwall data for /vitals and /strings - -""" -import pypowerwall -from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer -from socketserver import ThreadingMixIn -import os -import json -import time -import sys -import resource -import requests -import ssl -from transform import get_static, inject_js - -BUILD = "r18" -ALLOWLIST = [ - '/api/status', '/api/site_info/site_name', '/api/meters/site', - '/api/meters/solar', '/api/sitemaster', '/api/powerwalls', - '/api/customer/registration', '/api/system_status', '/api/system_status/grid_status', - '/api/system/update/status', '/api/site_info', '/api/system_status/grid_faults', - '/api/operation', '/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', - '/api/troubleshooting/problems' - ] -web_root = os.path.join(os.path.dirname(__file__), "web") - -# Configuration for Proxy - Check for environmental variables -# and always use those if available (required for Docker) -bind_address = os.getenv("PW_BIND_ADDRESS", "") -password = os.getenv("PW_PASSWORD", "password") -email = os.getenv("PW_EMAIL", "email@example.com") -host = os.getenv("PW_HOST", "hostname") -timezone = os.getenv("PW_TIMEZONE", "America/Los_Angeles") -debugmode = os.getenv("PW_DEBUG", "no") -cache_expire = int(os.getenv("PW_CACHE_EXPIRE", "5")) -timeout = int(os.getenv("PW_TIMEOUT", "10")) -pool_maxsize = int(os.getenv("PW_POOL_MAXSIZE", "10")) -https_mode = os.getenv("PW_HTTPS", "no") -port = int(os.getenv("PW_PORT", "8675")) -style = os.getenv("PW_STYLE", "clear") + ".js" -restrict = os.getenv("PW_RESTRICT", "") - -# Global Stats -proxystats = {} -proxystats['pypowerwall'] = "%s Proxy %s" % (pypowerwall.version, BUILD) -proxystats['gets'] = 0 -proxystats['errors'] = 0 -proxystats['timeout'] = 0 -proxystats['uri'] = {} -proxystats['ts'] = int(time.time()) # Timestamp for Now -proxystats['start'] = int(time.time()) # Timestamp for Start -proxystats['clear'] = int(time.time()) # Timestamp of lLast Stats Clear - -if https_mode == "yes": - # run https mode with self-signed cert - cookiesuffix = "path=/;SameSite=None;Secure;" - httptype = "HTTPS" -elif https_mode == "http": - # run http mode but simulate https for proxy behind https proxy - cookiesuffix = "path=/;SameSite=None;Secure;" - httptype = "HTTP" -else: - # run in http mode - cookiesuffix = "path=/;" - httptype = "HTTP" - -if(debugmode == "yes"): - pypowerwall.set_debug(True) - sys.stderr.write("pyPowerwall [%s] Proxy Server [%s] Started - %s Port %d - DEBUG\n" % - (pypowerwall.version, BUILD, httptype, port)) -else: - sys.stderr.write("pyPowerwall [%s] Proxy Server [%s] Started - %s Port %d\n" % - (pypowerwall.version, BUILD, httptype, port)) - sys.stderr.flush() - -# Connect to Powerwall -pw = pypowerwall.Powerwall(host,password,email,timezone,cache_expire,timeout,pool_maxsize) - -# Cached assets from Powerwall web interface passthrough -web_cache = {} - -class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): - daemon_threads = True - pass - -class handler(BaseHTTPRequestHandler): - def log_message(self, format, *args): - if debugmode == "yes": - sys.stderr.write("%s - - [%s] %s\n" % - (self.address_string(), - self.log_date_time_string(), - format%args)) - else: - pass - - def address_string(self): - # replace function to avoid lookup delays - host, hostport = self.client_address[:2] - return host - - def do_GET(self): - allowed = True - self.send_response(200) - message = "ERROR!" - contenttype = 'application/json' - if restrict != "": - # Restrict access to users with secret - allowed = False - cookies = self.headers.get('Cookie') - for c in cookies.split(";"): - kv = c.split("=") - if len(kv) >= 2: - if kv[0].strip() == 'PW_SECRET' and kv[1].strip() == restrict: - # Found secret cookie - allowed = True - break - if not allowed: - # Secret not found in cookies - did they send secret? - sesame = "/?s=%s" % restrict - contenttype = 'text/html' - if self.path.startswith(sesame): - self.send_header("Set-Cookie", "PW_SECRET={};{}".format(restrict, cookiesuffix)) - message = '\n\n' - message += '\n

Access Granted

\n' - else: - # Send form for user to enter secret - message = "\n\n
" - message += "\n" - message += "
\n" - message += "\n" - if allowed: - if self.path == '/aggregates' or self.path == '/api/meters/aggregates': - # Meters - JSON - message = pw.poll('/api/meters/aggregates') - elif self.path == '/soe': - # Battery Level - JSON - message = pw.poll('/api/system_status/soe') - elif self.path == '/api/system_status/soe': - # Force 95% Scale - level = pw.level(scale=True) - message = json.dumps({"percentage":level}) - elif self.path == '/csv': - # Grid,Home,Solar,Battery,Level - CSV - contenttype = 'text/plain; charset=utf-8' - batterylevel = pw.level() - grid = pw.grid() - solar = pw.solar() - battery = pw.battery() - home = pw.home() - message = "%0.2f,%0.2f,%0.2f,%0.2f,%0.2f\n" \ - % (grid, home, solar, battery, batterylevel) - elif self.path == '/vitals': - # Vitals Data - JSON - message = pw.vitals(jsonformat=True) - elif self.path == '/strings': - # Strings Data - JSON - message = pw.strings(jsonformat=True) - elif self.path == '/stats': - # Give Internal Stats - proxystats['ts'] = int(time.time()) - proxystats['mem'] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss - message = json.dumps(proxystats) - elif self.path == '/stats/clear': - # Clear Internal Stats - proxystats['gets'] = 0 - proxystats['errors'] = 0 - proxystats['uri'] = {} - proxystats['clear'] = int(time.time()) - message = json.dumps(proxystats) - elif self.path == '/temps': - # Temps of Powerwalls - message = pw.temps(jsonformat=True) - elif self.path == '/temps/pw': - # Temps of Powerwalls with Simple Keys - pwtemp = {} - idx = 1 - temps = pw.temps() - for i in temps: - key = "PW%d_temp" % idx - pwtemp[key] = temps[i] - idx = idx + 1 - message = json.dumps(pwtemp) - elif self.path == '/alerts': - # Alerts - message = pw.alerts(jsonformat=True) - elif self.path == '/freq': - # Frequency, Current, Voltage and Grid Status - fcv = {} - idx = 1 - vitals = pw.vitals() - for device in vitals: - d = vitals[device] - if device.startswith('TEPINV'): - # PW freq - fcv["PW%d_name" % idx] = device - fcv["PW%d_PINV_Fout" % idx] = d['PINV_Fout'] - fcv["PW%d_PINV_VSplit1" % idx] = d['PINV_VSplit1'] - fcv["PW%d_PINV_VSplit2" % idx] = d['PINV_VSplit2'] - idx = idx + 1 - if device.startswith('TESYNC') or device.startswith('TEMSA'): - # Island and Meter Metrics from Backup Gateway or Backup Switch - for i in d: - if i.startswith('ISLAND') or i.startswith('METER'): - fcv[i] = d[i] - fcv["grid_status"] = pw.grid_status(type="numeric") - message = json.dumps(fcv) - elif self.path == '/pod': - # Battery Data - pod = {} - idx = 1 - vitals = pw.vitals() - for device in vitals: - d = vitals[device] - if device.startswith('TEPOD'): - pod["PW%d_name" % idx] = device - pod["PW%d_POD_ActiveHeating" % idx] = int(d['POD_ActiveHeating']) - pod["PW%d_POD_ChargeComplete" % idx] = int(d['POD_ChargeComplete']) - pod["PW%d_POD_ChargeRequest" % idx] = int(d['POD_ChargeRequest']) - pod["PW%d_POD_DischargeComplete" % idx] = int(d['POD_DischargeComplete']) - pod["PW%d_POD_PermanentlyFaulted" % idx] = int(d['POD_PermanentlyFaulted']) - pod["PW%d_POD_PersistentlyFaulted" % idx] = int(d['POD_PersistentlyFaulted']) - pod["PW%d_POD_enable_line" % idx] = int(d['POD_enable_line']) - pod["PW%d_POD_available_charge_power" % idx] = d['POD_available_charge_power'] - pod["PW%d_POD_available_dischg_power" % idx] = d['POD_available_dischg_power'] - pod["PW%d_POD_nom_energy_remaining" % idx] = d['POD_nom_energy_remaining'] - pod["PW%d_POD_nom_energy_to_be_charged" % idx] = d['POD_nom_energy_to_be_charged'] - pod["PW%d_POD_nom_full_pack_energy" % idx] = d['POD_nom_full_pack_energy'] - idx = idx + 1 - pod["backup_reserve_percent"] = pw.get_reserve() - message = json.dumps(pod) - elif self.path == '/version': - # Firmware Version - v = {} - v["version"] = pw.version() - val = pw.version().split(" ")[0] - while len(val.split('.')) < 3: - val = val + ".0" - l = [int(x, 10) for x in val.split('.')] - l.reverse() - v["vint"] = sum(x * (100 ** i) for i, x in enumerate(l)) - message = json.dumps(v) - elif self.path == '/help': - contenttype = 'text/plain; charset=utf-8' - message = 'HELP: See https://github.com/jasonacox/pypowerwall/blob/main/proxy/HELP.md' - elif self.path in ALLOWLIST: - # Allowed API Call - message = pw.poll(self.path) - else: - # Set auth headers required for web application - self.send_header("Set-Cookie", "AuthCookie={};{}".format(pw.auth['AuthCookie'], cookiesuffix)) - self.send_header("Set-Cookie", "UserRecord={};{}".format(pw.auth['UserRecord'], cookiesuffix)) - - # Serve static assets from web root first, if found. - fcontent, ftype = get_static(web_root, self.path) - if fcontent: - self.send_header('Content-type','{}'.format(ftype)) - self.end_headers() - self.wfile.write(fcontent) - return - - # Proxy request to Powerwall web server and cache. - cache_item = web_cache.get(self.path, None) - if not cache_item: - proxy_path = self.path - if proxy_path.startswith("/"): - proxy_path = proxy_path[1:] - pw_url = "https://{}/{}".format(pw.host, proxy_path) - print("INFO: Proxy request: {}".format(pw_url)) - r = pw.session.get( - url=pw_url, - cookies=pw.auth, - verify=False, - stream=True, - timeout=pw.timeout - ) - fcontent = r.content - ftype = r.headers['content-type'] - web_cache[self.path] = (fcontent, ftype) - else: - fcontent, ftype = cache_item - - # Inject transformations - if self.path.split('?')[0] == "/": - if os.path.exists(os.path.join(web_root, style)): - fcontent = bytes(inject_js(fcontent, style), 'utf-8') - - self.send_header('Content-type','{}'.format(ftype)) - self.end_headers() - self.wfile.write(fcontent) - return - - # Count - if message is None: - proxystats['timeout'] = proxystats['timeout'] + 1 - message = "TIMEOUT!" - elif message == "ERROR!": - proxystats['errors'] = proxystats['errors'] + 1 - message = "ERROR!" - else: - proxystats['gets'] = proxystats['gets'] + 1 - if self.path in proxystats['uri']: - proxystats['uri'][self.path] = proxystats['uri'][self.path] + 1 - else: - proxystats['uri'][self.path] = 1 - - # Send headers and payload - self.send_header('Content-type',contenttype) - self.send_header('Content-Length', str(len(message))) - self.end_headers() - self.wfile.write(bytes(message, "utf8")) - -with ThreadingHTTPServer((bind_address, port), handler) as server: - if(https_mode == "yes"): - # Activate HTTPS - server.socket = ssl.wrap_socket (server.socket, - certfile=os.path.join(os.path.dirname(__file__), 'localhost.pem'), - server_side=True, ssl_version=ssl.PROTOCOL_TLSv1_2, ca_certs=None, - do_handshake_on_connect=True) - - try: - server.serve_forever() - except: - print(' CANCEL \n') - sys.stderr.write("pyPowerwall Proxy Stopped\n") - sys.stderr.flush() - os._exit(0) diff --git a/proxy/server.py b/proxy/server.py index 89b2f08..cf4023c 100755 --- a/proxy/server.py +++ b/proxy/server.py @@ -12,7 +12,21 @@ /api/system_status/soe - You can containerize it and run it as an endpoint for tools like telegraf to pull metrics. - This proxy also supports pyPowerwall data for /vitals and /strings + Local Powerwall Mode + The default mode for this proxy is to connect to a local Powerwall + to pull data. This works with the Tesla Energy Gateway (TEG) for + Powerwall 1, 2 and +. It will also support pulling /vitals and /strings + data if available. + Set: PW_HOST to Powerwall Address and PW_PASSWORD to use this mode. + + Cloud Mode + An optional mode is to connect to the Tesla Cloud to pull data. This + requires that you have a Tesla Account and have registered your + Tesla Solar System or Powerwall with the Tesla App. It requires that + you run the setup 'python -m pypowerwall setup' process to create the + required API keys and tokens. This mode doesn't support /vitals or + /strings data. + Set: PW_EMAIL and leave PW_HOST blank to use this mode. """ import pypowerwall @@ -28,8 +42,7 @@ import ssl from transform import get_static, inject_js -BUILD = "t29" -NOAPI = False # Set to True to serve bogus data for API calls to Powerwall +BUILD = "t35" ALLOWLIST = [ '/api/status', '/api/site_info/site_name', '/api/meters/site', '/api/meters/solar', '/api/sitemaster', '/api/powerwalls', @@ -47,16 +60,17 @@ bind_address = os.getenv("PW_BIND_ADDRESS", "") password = os.getenv("PW_PASSWORD", "password") email = os.getenv("PW_EMAIL", "email@example.com") -host = os.getenv("PW_HOST", "hostname") +host = os.getenv("PW_HOST", "") timezone = os.getenv("PW_TIMEZONE", "America/Los_Angeles") debugmode = os.getenv("PW_DEBUG", "no") cache_expire = int(os.getenv("PW_CACHE_EXPIRE", "5")) browser_cache = int(os.getenv("PW_BROWSER_CACHE", "0")) -timeout = int(os.getenv("PW_TIMEOUT", "10")) +timeout = int(os.getenv("PW_TIMEOUT", "5")) pool_maxsize = int(os.getenv("PW_POOL_MAXSIZE", "15")) https_mode = os.getenv("PW_HTTPS", "no") port = int(os.getenv("PW_PORT", "8675")) style = os.getenv("PW_STYLE", "clear") + ".js" +siteid = os.getenv("PW_SITEID", None) # Global Stats proxystats = {} @@ -69,6 +83,11 @@ proxystats['start'] = int(time.time()) # Timestamp for Start proxystats['clear'] = int(time.time()) # Timestamp of lLast Stats Clear proxystats['uptime'] = "" +proxystats['mem'] = 0 +proxystats['site_name'] = "" +proxystats['cloudmode'] = False +proxystats['siteid'] = 0 +proxystats['counter'] = 0 if https_mode == "yes": # run https mode with self-signed cert @@ -108,13 +127,25 @@ def get_value(a, key): # Connect to Powerwall # TODO: Add support for multiple Powerwalls -# TODO: Add support for solar-only systems -pw = pypowerwall.Powerwall(host,password,email,timezone,cache_expire,timeout,pool_maxsize) -if not pw or NOAPI or not host or host.lower() == "none": - NOAPI = True - log.info("pyPowerwall Proxy Server - NOAPI Mode") +try: + pw = pypowerwall.Powerwall(host,password,email,timezone,cache_expire,timeout,pool_maxsize) +except Exception as e: + log.error(e) + log.error("Fatal Error: Unable to connect. Please fix config and restart.") + while True: + time.sleep(5) # Infinite loop to keep container running +if pw.cloudmode: + log.info("pyPowerwall Proxy Server - Cloud Mode") + log.info("Connected to Site ID %s (%s)" % (pw.Tesla.siteid, pw.site_name())) + if siteid is not None and siteid != str(pw.Tesla.siteid): + log.info("Switch to Site ID %s" % siteid) + if not pw.Tesla.change_site(siteid): + log.error("Fatal Error: Unable to connect. Please fix config and restart.") + while True: + time.sleep(5) # Infinite loop to keep container running else: - log.info("pyPowerwall Proxy Server - Connected to %s" % host) + log.info("pyPowerwall Proxy Server - Local Mode") + log.info("Connected to Energy Gateway %s (%s)" % (host, pw.site_name())) class ThreadingHTTPServer(ThreadingMixIn, HTTPServer): daemon_threads = True @@ -173,6 +204,11 @@ def do_GET(self): delta = proxystats['ts'] - proxystats['start'] proxystats['uptime'] = str(datetime.timedelta(seconds=delta)) proxystats['mem'] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + proxystats['site_name'] = pw.site_name() + proxystats['cloudmode'] = pw.cloudmode + if pw.cloudmode and pw.Tesla is not None: + proxystats['siteid'] = pw.Tesla.siteid + proxystats['counter'] = pw.Tesla.counter message = json.dumps(proxystats) elif self.path == '/stats/clear': # Clear Internal Stats @@ -199,18 +235,21 @@ def do_GET(self): # Alerts message = pw.alerts(jsonformat=True) elif self.path == '/alerts/pw': - # Alerts in dictionary/object format - pwalerts = {} - idx = 1 - alerts = pw.alerts() - for alert in alerts: - pwalerts[alert] = 1 - message = json.dumps(pwalerts) + # Alerts in dictionary/object format + pwalerts = {} + idx = 1 + alerts = pw.alerts() + if alerts is None: + message = None + else: + for alert in alerts: + pwalerts[alert] = 1 + message = json.dumps(pwalerts) elif self.path == '/freq': # Frequency, Current, Voltage and Grid Status fcv = {} idx = 1 - vitals = pw.vitals() + vitals = pw.vitals() or {} for device in vitals: d = vitals[device] if device.startswith('TEPINV'): @@ -231,7 +270,7 @@ def do_GET(self): # Battery Data pod = {} idx = 1 - vitals = pw.vitals() + vitals = pw.vitals() or {} for device in vitals: d = vitals[device] if device.startswith('TEPOD'): @@ -250,24 +289,38 @@ def do_GET(self): pod["PW%d_POD_nom_full_pack_energy" % idx] = get_value(d, 'POD_nom_full_pack_energy') idx = idx + 1 pod["backup_reserve_percent"] = pw.get_reserve() + d = pw.system_status() or {} + pod["nominal_full_pack_energy"] = get_value(d,'nominal_full_pack_energy') + pod["nominal_energy_remaining"] = get_value(d,'nominal_energy_remaining') + pod["time_remaining_hours"] = pw.get_time_remaining() message = json.dumps(pod) elif self.path == '/version': # Firmware Version - v = {} - v["version"] = pw.version() - val = pw.version().split(" ")[0] - val = ''.join(i for i in val if i.isdigit() or i in './\\') - while len(val.split('.')) < 3: - val = val + ".0" - l = [int(x, 10) for x in val.split('.')] - l.reverse() - v["vint"] = sum(x * (100 ** i) for i, x in enumerate(l)) - message = json.dumps(v) + version = pw.version() + if version is None: + message = None + else: + v = {} + v["version"] = version + val = v["version"].split(" ")[0] + val = ''.join(i for i in val if i.isdigit() or i in './\\') + while len(val.split('.')) < 3: + val = val + ".0" + l = [int(x, 10) for x in val.split('.')] + l.reverse() + v["vint"] = sum(x * (100 ** i) for i, x in enumerate(l)) + message = json.dumps(v) elif self.path == '/help': # Display friendly help screen link and stats proxystats['ts'] = int(time.time()) delta = proxystats['ts'] - proxystats['start'] proxystats['uptime'] = str(datetime.timedelta(seconds=delta)) + proxystats['mem'] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + proxystats['site_name'] = pw.site_name() + proxystats['cloudmode'] = pw.cloudmode + if pw.cloudmode and pw.Tesla is not None: + proxystats['siteid'] = pw.Tesla.siteid + proxystats['counter'] = pw.Tesla.counter contenttype = 'text/html' message = '\n\n' message += '\n' @@ -285,22 +338,7 @@ def do_GET(self): str(datetime.datetime.fromtimestamp(time.time()))) elif self.path in ALLOWLIST: # Allowed API Calls - Proxy to Powerwall - if NOAPI: - # If we are in NOAPI mode serve up bogus data - filename = "/bogus/" + self.path[1:].replace('/', '.') + ".json" - fcontent, ftype = get_static(web_root, filename) - if not fcontent: - fcontent = bytes("{}", 'utf-8') - log.debug("No offline content found for: {}".format(filename)) - else: - log.debug("Served from local filesystem: {}".format(filename)) - self.send_header('Content-type','{}'.format(ftype)) - self.end_headers() - self.wfile.write(fcontent) - return - else: - # Proxy request to Powerwall - message = pw.poll(self.path) + message = pw.poll(self.path) else: # Everything else - Set auth headers required for web application proxystats['gets'] = proxystats['gets'] + 1 @@ -310,34 +348,48 @@ def do_GET(self): # Serve static assets from web root first, if found. if self.path == "/" or self.path == "": self.path = "/index.html" - fcontent, ftype = get_static(web_root, self.path) + fcontent, ftype = get_static(web_root, self.path) + # Replace {VARS} with current data + status = pw.status() + # convert fcontent to string + fcontent = fcontent.decode("utf-8") + fcontent = fcontent.replace("{VERSION}", status["version"]) + fcontent = fcontent.replace("{HASH}", status["git_hash"]) + fcontent = fcontent.replace("{EMAIL}", email) + fcontent = fcontent.replace("{STYLE}", style) + # convert fcontent back to bytes + fcontent = bytes(fcontent, 'utf-8') + else: + fcontent, ftype = get_static(web_root, self.path) if fcontent: - log.debug("Served from local web root: {}".format(self.path)) - self.send_header('Content-type','{}'.format(ftype)) - self.end_headers() - self.wfile.write(fcontent) - return - - # Proxy request to Powerwall web server. - proxy_path = self.path - if proxy_path.startswith("/"): - proxy_path = proxy_path[1:] - pw_url = "https://{}/{}".format(pw.host, proxy_path) - log.debug("Proxy request to: {}".format(pw_url)) - r = pw.session.get( - url=pw_url, - cookies=pw.auth, - verify=False, - stream=True, - timeout=pw.timeout - ) - fcontent = r.content - ftype = r.headers['content-type'] + log.debug("Served from local web root: {} type {}".format(self.path,ftype)) + # If not found, serve from Powerwall web server + elif pw.cloudmode: + log.debug("Cloud Mode - File not found: {}".format(self.path)) + fcontent = bytes("Not Found", 'utf-8') + ftype = "text/plain" + else: + # Proxy request to Powerwall web server. + proxy_path = self.path + if proxy_path.startswith("/"): + proxy_path = proxy_path[1:] + pw_url = "https://{}/{}".format(pw.host, proxy_path) + log.debug("Proxy request to: {}".format(pw_url)) + r = pw.session.get( + url=pw_url, + cookies=pw.auth, + verify=False, + stream=True, + timeout=pw.timeout + ) + fcontent = r.content + ftype = r.headers['content-type'] + # Allow browser caching, if user permits, only for CSS, JavaScript and PNG images... if browser_cache > 0 and (ftype == 'text/css' or ftype == 'application/javascript' or ftype == 'image/png'): self.send_header("Cache-Control", "max-age={}".format(browser_cache)) else: - self.send_header("Cache-Control", "no-cache, no-store") + self.send_header("Cache-Control", "no-cache, no-store") # Inject transformations if self.path.split('?')[0] == "/": diff --git a/proxy/transform.py b/proxy/transform.py index 5504ae9..1d7ace6 100644 --- a/proxy/transform.py +++ b/proxy/transform.py @@ -27,7 +27,23 @@ def get_static(web_root, fpath): ftype = "image/png" elif freq.lower().endswith(".html"): ftype = "text/html" - else: + elif freq.lower().endswith(".otf"): + ftype = "font/opentype" + elif freq.lower().endswith(".woff"): + ftype = "font/woff" + elif freq.lower().endswith(".woff2"): + ftype = "font/woff2" + elif freq.lower().endswith(".ttf"): + ftype = "font/ttf" + elif freq.lower().endswith(".svg"): + ftype = "image/svg+xml" + elif freq.lower().endswith(".eot"): + ftype = "application/vnd.ms-fontobject" + elif freq.lower().endswith(".json"): + ftype = "application/json" + elif freq.lower().endswith(".xml"): + ftype = "application/xml" + else: ftype = "text/plain" with open(freq, 'rb') as f: diff --git a/proxy/web/index.html b/proxy/web/index.html index b8122f6..02bd44d 100644 --- a/proxy/web/index.html +++ b/proxy/web/index.html @@ -10,8 +10,8 @@ window.__INITIAL_STATE__ = { "configuration": { "isNew": false, - "version": "23.28.2 27626f98", - "hash": "27626f98a66cad5c665bbe1d4d788cdb3e94fd33", + "version": "{VERSION}", + "hash": "{HASH}", "deviceType": "teg", "isSolarPowerwall": false, "bootstrapped": true @@ -27,7 +27,7 @@ "authentication": { "loginType": "customer", "selectedLoginType": "customer", - "username": "jason@jasonacox.com", + "username": "{EMAIL}", "lastLoginAt": 1700501180000, "toggle_auth_supported": true } @@ -39,6 +39,6 @@
- + diff --git a/pypowerwall/__init__.py b/pypowerwall/__init__.py index 597075c..3a2a059 100644 --- a/pypowerwall/__init__.py +++ b/pypowerwall/__init__.py @@ -7,14 +7,26 @@ For more information see https://github.com/jasonacox/pypowerwall Features - * Works with Tesla Energy Gateways - Powerwall+ + * Works with Tesla Energy Gateways - Powerwall+ * Simple access through easy to use functions using customer credentials * Will cache authentication to reduce load on Powerwall Gateway * Will cache responses for 5s to limit number of calls to Powerwall Gateway * Will re-use http connections to Powerwall Gateway for reduced load and faster response times + * Can use Tesla Cloud API instead of local Powerwall Gateway (if enabled) Classes - Powerwall(host, password, email, timezone, pwcacheexpire, timeout, poolmaxsize) + Powerwall(host, password, email, timezone, pwcacheexpire, timeout, poolmaxsize, cloudmode) + + Parameters + host # Hostname or IP of the Tesla gateway + password # Customer password for gateway + email # (required) Customer email for gateway / cloud + timezone # Desired timezone + pwcacheexpire = 5 # Set API cache timeout in seconds + timeout = 5 # Timeout for HTTPS calls in seconds + 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) Functions poll(api, json, force) # Return data from Powerwall api (dict if json=True, bypass cache force=True) @@ -41,15 +53,12 @@ # - "string": "UP", "DOWN", "SYNCING" # - "numeric": -1 (Syncing), 0 (DOWN), 1 (UP) is_connected() # Returns True if able to connect and login to Powerwall + get_reserve(scale) # Get Battery Reserve Percentage + get_time_remaining() # Get the backup time remaining on the battery - Parameters - host # (required) hostname or IP of the Tesla gateway - password # (required) password for logging into the gateway - email # (required) email used for logging into the gateway - timezone # (required) desired timezone - pwcacheexpire = 5 # Set API cache timeout in seconds - timeout = 10 # Timeout for HTTPS calls in seconds - poolmaxsize = 10 # Pool max size for http connection re-use (persistent connections disabled if zero) + Requirements + This module requires the following modules: requests, protobuf, teslapy + pip install requests protobuf teslapy """ import json, time import requests @@ -58,8 +67,9 @@ import logging import sys from . import tesla_pb2 # Protobuf definition for vitals +from . import cloud # Tesla Cloud API -version_tuple = (0, 6, 4) +version_tuple = (0, 7, 1) version = __version__ = '%d.%d.%d' % version_tuple __author__ = 'jasonacox' @@ -86,7 +96,7 @@ class ConnectionError(Exception): pass class Powerwall(object): - def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="America/Los_Angeles", pwcacheexpire=5, timeout=10, poolmaxsize=10): + def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="America/Los_Angeles", pwcacheexpire=5, timeout=5, poolmaxsize=10, cloudmode=False): """ Represents a Tesla Energy Gateway Powerwall device. @@ -99,6 +109,7 @@ def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="A pwcacheexpire = Seconds to expire cached entries timeout = Seconds for the timeout on http requests poolmaxsize = Pool max size for http connection re-use (persistent connections disabled if zero) + cloudmode = If True, use Tesla cloud for data (default is False) """ @@ -108,34 +119,47 @@ def __init__(self, host="", password="", email="nobody@nowhere.com", timezone="A self.password = password self.email = email self.timezone = timezone - self.timeout = timeout # 10s timeout for http calls + self.timeout = timeout # 5s timeout for http calls self.poolmaxsize = poolmaxsize # pool max size for http connection re-use self.auth = {} # caches authentication cookies self.pwcachetime = {} # holds the cached data timestamps for api self.pwcache = {} # holds the cached data for api self.pwcacheexpire = pwcacheexpire # seconds to expire cache - - if self.poolmaxsize > 0: - # Create session object for http connection re-use - self.session = requests.Session() - a = requests.adapters.HTTPAdapter(pool_maxsize=self.poolmaxsize) - self.session.mount('https://', a) + self.cloudmode = cloudmode # cloud mode or local mode (default) + self.Tesla = None # cloud object for cloud connection + + # Check for cloud mode + if self.cloudmode or self.host == "": + self.cloudmode = True + log.debug('Tesla cloud mode enabled') + self.Tesla = cloud.TeslaCloud(self.email, pwcacheexpire, timeout) + # Check to see if we can connect to the cloud + if not self.Tesla.connect(): + err = "Unable to connect to Tesla Cloud - run pypowerwall setup" + log.debug(err) + raise ConnectionError(err) + self.auth = {'AuthCookie': 'local', 'UserRecord': 'local'} else: - # Disable http persistent connections - self.session = requests - - # Load cached auth session - try: - f = open(self.cachefile, "r") - self.auth = json.load(f) - log.debug('loaded auth from cache file %s' % self.cachefile) - except: - log.debug('no auth cache file') - pass - - # Create new session - if self.auth == {}: - self._get_session() + log.debug('Tesla local mode enabled') + if self.poolmaxsize > 0: + # Create session object for http connection re-use + self.session = requests.Session() + a = requests.adapters.HTTPAdapter(pool_maxsize=self.poolmaxsize) + self.session.mount('https://', a) + else: + # Disable http persistent connections + self.session = requests + # Load cached auth session + try: + f = open(self.cachefile, "r") + self.auth = json.load(f) + log.debug('loaded auth from cache file %s' % self.cachefile) + except: + log.debug('no auth cache file') + pass + # Create new session + if self.auth == {}: + self._get_session() def _get_session(self): # Login and create a new session @@ -165,6 +189,9 @@ def _get_session(self): def _close_session(self): # Log out + if self.cloudmode: + self.Tesla.logout() + return url = "https://%s/api/logout" % self.host g = self.session.get(url, cookies=self.auth, verify=False, timeout=self.timeout) self.auth = {} @@ -193,13 +220,19 @@ def poll(self, api='/api/site_info/site_name', jsonformat=False, raw=False, recu recursive = If True, this is a recursive call and do not allow additional recursive calls force = If True, bypass the cache and make the API call to the gateway """ - # Query powerwall and return payload as string - - # First check to see if in cache + # Check to see if we are in cloud mode + if self.cloudmode: + if jsonformat: + return self.Tesla.poll(api) + else: + return json.dumps(self.Tesla.poll(api)) + + # Query powerwall and return payload fetch = True + # Check cache if(api in self.pwcache and api in self.pwcachetime): # is it expired? - if(time.time() - self.pwcachetime[api] < self.pwcacheexpire): + if(time.perf_counter() - self.pwcachetime[api] < self.pwcacheexpire): payload = self.pwcache[api] # We do the override here to ensure that we cache the force entry if force: @@ -240,7 +273,7 @@ def poll(self, api='/api/site_info/site_name', jsonformat=False, raw=False, recu else: payload = r.text self.pwcache[api] = payload - self.pwcachetime[api] = time.time() + self.pwcachetime[api] = time.perf_counter() if(jsonformat): try: data = json.loads(payload) @@ -260,14 +293,13 @@ def level(self, scale=False): Note: Tesla App reserves 5% of battery = ( (batterylevel / 0.95) - (5 / 0.95) ) """ # Return power level percentage for battery - level = 0 payload = self.poll('/api/system_status/soe', jsonformat=True) - if(payload is not None and 'percentage' in payload): + if payload is not None and 'percentage' in payload: level = payload['percentage'] - if scale: - return ((level / 0.95) - (5 / 0.95)) - else: + if scale: + level = (level / 0.95) - (5 / 0.95) return level + return None def power(self): """ @@ -304,6 +336,12 @@ def vitals(self, jsonformat=False): Args: jsonformat = If True, return JSON format otherwise return Python Dictionary """ + if self.cloudmode: + if jsonformat: + return json.dumps(self.Tesla.poll('/vitals')) + else: + return self.Tesla.poll('/vitals') + # Pull vitals payload - binary protobuf stream = self.poll('/api/devices/vitals') if(not stream): @@ -404,9 +442,7 @@ def strings(self, jsonformat=False, verbose=False): result = {} devicemap = ['','1','2','3','4','5','6','7','8'] deviceidx = 0 - v = self.vitals(jsonformat=False) - if(not v): - return None + v = self.vitals(jsonformat=False) or {} for device in v: if device.split('--')[0] == 'PVAC': # Check for PVS data @@ -513,6 +549,8 @@ def status(self, param=None, jsonformat=False): cellular_disabled = payload['cellular_disabled'] """ payload = self.poll('/api/status', jsonformat=True) + if payload is None: + return None if param is None: if jsonformat: return json.dumps(payload, indent=4, sort_keys=True) @@ -540,7 +578,7 @@ def din(self): def temps(self, jsonformat=False): """ Temperatures of Powerwalls """ temps = {} - devices = self.vitals() + devices = self.vitals() or {} for device in devices: if device.startswith('TETHC'): try: @@ -562,7 +600,7 @@ def alerts(self, jsonformat=False, alertsonly=True): jsonformat = If True, return JSON format otherwise return Python Dictionary """ alerts = [] - devices = self.vitals() + devices = self.vitals() or {} for device in devices: if 'alerts' in devices[device]: for i in devices[device]['alerts']: @@ -586,15 +624,14 @@ def get_reserve(self, scale=True): scale = If True (default) use Tesla's 5% reserve calculation Tesla App reserves 5% of battery = ( (batterylevel / 0.95) - (5 / 0.95) ) """ - data = self.poll('/api/operation') - if data is None: - return None - data = json.loads(data) - percent = float(data['backup_reserve_percent']) - if scale: - # Get percentage based on Tesla App scale - percent = float((percent / 0.95) - (5 / 0.95)) - return percent + data = self.poll('/api/operation', jsonformat=True) + if data is not None and 'backup_reserve_percent' in data: + percent = float(data['backup_reserve_percent']) + if scale: + # Get percentage based on Tesla App scale + percent = float((percent / 0.95) - (5 / 0.95)) + return percent + return None def grid_status(self, type="string"): """ @@ -625,7 +662,7 @@ def grid_status(self, type="string"): return gridmap[grid_status][type] except: # The payload from powerwall was not valid - log.debug("ERROR Invalid return value received from gateway: " + str(payload.grid_status)) + log.debug('ERROR unable to parse payload for grid_status: %r' % payload) return None def system_status(self, jsonformat=False): @@ -719,4 +756,29 @@ def battery_blocks(self, jsonformat=False): if jsonformat: return json.dumps(result, indent=4, sort_keys=True) else: - return result \ No newline at end of file + return result + + def get_time_remaining(self): + """ + Get the backup time remaining on the battery + + Returns: + The time remaining in hours + """ + if self.cloudmode: + d = self.Tesla.get_time_remaining() + # {'response': {'time_remaining_hours': 7.909122698326978}} + if d is None: + return None + if 'response' in d and 'time_remaining_hours' in d['response']: + return d['response']['time_remaining_hours'] + + # Compute based on battery level and load + d = self.system_status() or {} + if 'nominal_energy_remaining' in d and d['nominal_energy_remaining'] is not None: + load = self.load() or 0 + if load > 0: + return d['nominal_energy_remaining']/load + # Default + return None + \ No newline at end of file diff --git a/pypowerwall/__main__.py b/pypowerwall/__main__.py index 9448416..ac6a024 100644 --- a/pypowerwall/__main__.py +++ b/pypowerwall/__main__.py @@ -15,8 +15,10 @@ import pypowerwall import sys from . import scan +from . import cloud # Global Variables +AUTHFILE = ".pypowerwall.auth" timeout = 1.0 state = 0 color = True @@ -26,6 +28,8 @@ continue elif(i.lower() == "scan"): state = 0 + elif(i.lower() == "setup"): + state = 1 elif(i.lower() == "-nocolor"): color = False else: @@ -38,9 +42,16 @@ if(state == 0): scan.scan(color, timeout) -# State 1 = Future +# State 1 = Cloud Mode Setup if(state == 1): - print("Future Feature") + print("pyPowerwall [%s] - Cloud Mode Setup\n" % (pypowerwall.version)) + # Run Setup + c = cloud.TeslaCloud(None) + if c.setup(): + print("Setup Complete. Auth file %s ready to use." % (AUTHFILE)) + else: + print("ERROR: Failed to setup Tesla Cloud Mode") + exit(1) # State 2 = Show Usage if(state == 2): @@ -49,6 +60,7 @@ print(" python -m pypowerwall [command] [] [-nocolor] [-h]") print("") print(" command = scan Scan local network for Powerwall gateway.") + print(" command = setup Setup Tesla Login for Cloud Mode access.") print(" timeout Seconds to wait per host [Default=%0.1f]" % (timeout)) print(" -nocolor Disable color text output.") print(" -h Show usage.") diff --git a/pypowerwall/cloud.py b/pypowerwall/cloud.py new file mode 100644 index 0000000..1da61bb --- /dev/null +++ b/pypowerwall/cloud.py @@ -0,0 +1,990 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" + Python library to pull live Powerwall or Solar energy site data from + Tesla Owner API (Tesla Cloud). + + Authors: Jason A. Cox and Michael Birse + For more information see https://github.com/jasonacox/pypowerwall + + Classes + TeslaCloud(email, timezone, pwcacheexpire, timeout, siteid) + + Parameters + email # (required) email used for logging into the gateway + timezone # (required) desired timezone + pwcacheexpire = 5 # Set API cache timeout in seconds + timeout = 5 # Timeout for HTTPS calls in seconds + siteid = None # (optional) energy_site_id to use + + Functions + setup() # Set up the Tesla Cloud connection + connect() # Connect to Tesla Cloud + change_site(siteid) # Select another site - energy_site_id + get_battery() # Get battery data from Tesla Cloud + get_site_power() # Get site power data from Tesla Cloud + get_site_config() # Get site configuration data from Tesla Cloud + get_time_remaining() # Get backup time remaining from Tesla Cloud + poll(api) # Map Powerwall API to Tesla Cloud Data + + Requirements + This module requires the python teslapy module. Install with: + pip install teslapy + +""" +import sys +import os +import time +import logging +import json +try: + from teslapy import Tesla, JsonDict, Battery, SolarPanel +except: + sys.exit("ERROR: Missing python teslapy module. Run 'pip install teslapy'.") + +AUTHFILE = ".pypowerwall.auth" # Stores auth session information +SITEFILE = ".pypowerwall.site" # Stores site id +COUNTER_MAX = 64 # Max counter value for SITE_DATA API +SITE_CONFIG_TTL = 59 # Site config cache TTL in seconds + +# pypowerwall cloud module version +version_tuple = (0, 0, 2) +version = __version__ = '%d.%d.%d' % version_tuple +__author__ = 'jasonacox' + +log = logging.getLogger(__name__) +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): + if(color): + logging.basicConfig(format='\x1b[31;1m%(levelname)s:%(message)s\x1b[0m',level=logging.DEBUG) + else: + logging.basicConfig(format='%(levelname)s:%(message)s',level=logging.DEBUG) + log.setLevel(logging.DEBUG) + log.debug("%s [%s]\n" % (__name__, __version__)) + 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 + """ + for key in keylist: + if key in data: + data = data[key] + else: + return None + return data + +class TeslaCloud: + def __init__(self, email, pwcacheexpire=5, timeout=5, siteid=None): + self.authfile = AUTHFILE + self.sitefile = SITEFILE + self.email = email + self.timeout = timeout + self.site = None + self.tesla = None + self.apilock = {} # holds lock flag for pending cloud api requests + 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.siteindex = 0 # site index to use + self.siteid = siteid # site id to use + self.counter = 0 # counter for SITE_DATA API + + if self.siteid is None: + # Check for site file + if os.path.exists(self.sitefile): + with open(self.sitefile) as file: + try: + self.siteid = int(file.read()) + except: + self.siteid = 0 + else: + self.siteindex = 0 + + # Check for auth file + if not os.path.exists(self.authfile): + log.debug("WARNING: Missing auth file %s - run setup" % self.authfile) + + def connect(self): + """ + Connect to Tesla Cloud via teslapy + """ + # Create Tesla instance + if not os.path.exists(self.authfile): + log.error("Missing auth file %s - run setup" % self.authfile) + return False + self.tesla = Tesla(self.email, cache_file=self.authfile, timeout=self.timeout) + # Check to see if we have a cached token + if not self.tesla.authorized: + # Login to Tesla account and cache token + state = self.tesla.new_state() + code_verifier = self.tesla.new_code_verifier() + try: + self.tesla.fetch_token(authorization_response=self.tesla.authorization_url(state=state, code_verifier=code_verifier)) + except Exception as err: + log.error("Login failure - %s" % repr(err)) + return False + # Get site info + sites = self.getsites() + 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 %d not found for %s" % (self.siteid, self.email)) + return False + # Set site + self.site = sites[self.siteindex] + log.debug(f"Connected to Tesla Cloud - Site {self.siteid} ({sites[self.siteindex]['site_name']}) for {self.email}") + return True + + def getsites(self): + """ + Get list of Tesla Energy sites + """ + if self.tesla is None: + return None + try: + sitelist = self.tesla.battery_list() + self.tesla.solar_list() + 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.site = 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 + + # Functions to get data from Tesla Cloud + + def _site_api(self, name, ttl, **kwargs): + """ + Private function to get site data from Tesla Cloud using + TeslaPy API. This function uses a lock to prevent threads + from sending multiple requests to Tesla Cloud at the same time. + It also caches the data for ttl seconds. + + Arguments: + name - TeslaPy API name + ttl - Cache expiration time in seconds + kwargs - Variable arguments to pass to API call + + Returns (response, cached) + response - TeslaPy API response + cached - True if cached data was returned + """ + if self.tesla is None: + log.debug(f" -- cloud: No connection to Tesla Cloud") + return (None, False) + # Check for lock and wait if api request already sent + if name in self.apilock: + locktime = time.perf_counter() + while self.apilock[name]: + time.sleep(0.2) + if time.perf_counter() >= locktime + self.timeout: + log.debug(f" -- cloud: Timeout waiting for {name}") + return (None, False) + # Check to see if we have cached data + if name in self.pwcache: + if self.pwcachetime[name] > time.perf_counter() - ttl: + log.debug(f" -- cloud: Returning cached {name} data") + return (self.pwcache[name], True) + try: + # Set lock + self.apilock[name] = True + response = self.site.api(name, **kwargs) + except Exception as err: + log.error(f"Failed to retrieve {name} - {repr(err)}") + response = None + else: + log.debug(f" -- cloud: Retrieved {name} data") + self.pwcache[name] = response + self.pwcachetime[name] = time.perf_counter() + finally: + # Release lock + self.apilock[name] = False + return (response, False) + + def get_battery(self): + """ + Get site battery data from Tesla Cloud + + "response": { + "resource_type": "battery", + "site_name": "Tesla Energy Gateway", + "gateway_id": "1232100-00-E--TGxxxxxxxxxxxx", + "energy_left": 21276.894736842103, + "total_pack_energy": 25939, + "percentage_charged": 82.02665768472995, + "battery_type": "ac_powerwall", + "backup_capable": true, + "battery_power": -220, + "go_off_grid_test_banner_enabled": null, + "storm_mode_enabled": false, + "powerwall_onboarding_settings_set": true, + "powerwall_tesla_electric_interested_in": null, + "vpp_tour_enabled": null, + "sync_grid_alert_enabled": true, + "breaker_alert_enabled": true + } + """ + # GET api/1/energy_sites/{site_id}/site_status + (response, _) = self._site_api("SITE_SUMMARY", + self.pwcacheexpire, language="en") + return response + + def get_site_power(self): + """ + Get site power data from Tesla Cloud + + "response": { + "solar_power": 1290, + "energy_left": 21276.894736842103, + "total_pack_energy": 25939, + "percentage_charged": 82.02665768472995, + "backup_capable": true, + "battery_power": -220, + "load_power": 1070, + "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": "2023-12-17T14:23:31-08:00", + "wall_connectors": [] + } + """ + # GET api/1/energy_sites/{site_id}/live_status?counter={counter}&language=en + (response, cached) = self._site_api("SITE_DATA", + self.pwcacheexpire, counter=self.counter+1, language="en") + if not cached: + self.counter = (self.counter + 1) % COUNTER_MAX + return response + + def get_site_config(self): + """ + Get site configuration data from Tesla Cloud + + "response": { + "id": "1232100-00-E--TGxxxxxxxxxxxx", + "site_name": "Tesla Energy Gateway", + "backup_reserve_percent": 80, + "default_real_mode": "self_consumption", + "installation_date": "xxxx-xx-xx", + "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": "Energy Value", + "energy_value_subheader": "Estimated Value", + "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, + "rate_plan_manager_supported": true, + "battery_type": "solar_powerwall", + "configurable": true, + "grid_services_enabled": false, + "inverters": [ + { + "device_id": "xxxxxxxxxxxxxxxxxx", + "din": "xxxxxxxxx", + "is_active": true, + "site_id": "xxxxxxxxxxxxxxxxxx", + } + ], + "edit_setting_permission_to_export": true, + "edit_setting_grid_charging": true, + "edit_setting_energy_exports": true + }, + "version": "23.28.2 27626f98", + "battery_count": 2, + "tariff_content": { # removed for brevity + }, + "tariff_id": "SCE-TOU-PRIME", + "nameplate_power": 10800, + "nameplate_energy": 27000, + "installation_time_zone": "America/Los_Angeles", + "off_grid_vehicle_charging_reserve_percent": 65, + "max_site_meter_power_ac": 1000000000, + "min_site_meter_power_ac": -1000000000, + "geolocation": { + "latitude": XX.XXXXXXX, + "longitude": XX.XXXXXXX, + "source": "Site Address Preference" + }, + "address": { + "address_line1": "xxxxxx", + "city": "xxxxxx", + "state": "xx", + "zip": "xxxxx", + "country": "xx" + }, + "vpp_backup_reserve_percent": 80 + } + } + + """ + # GET api/1/energy_sites/{site_id}/site_info + (response, _) = self._site_api("SITE_CONFIG", + SITE_CONFIG_TTL, language="en") + return response + + def get_time_remaining(self): + """ + Get backup time remaining from Tesla Cloud + + {'response': {'time_remaining_hours': 7.909122698326978}} + """ + # GET api/1/energy_sites/{site_id}/backup_time_remaining + (response, _) = self._site_api("ENERGY_SITE_BACKUP_TIME_REMAINING", + self.pwcacheexpire, language="en") + return response + + # Function to map Powerwall API to Tesla Cloud Data + + def poll(self, api): + """ + Map Powerwall API to Tesla Cloud Data + + """ + if self.tesla is None: + return None + # API Map - Determine what data we need based on Powerwall APIs + log.debug(f" -- cloud: Request for {api}") + + ## Dynamic Values + if api == '/api/status': + # TOOO: Fix start_time and up_time_seconds + config = self.get_site_config() + if config is None: + data = None + else: + data = { + "din": lookup(config, ("response", "id")), # 1232100-00-E--TGxxxxxxxxxxxx + "start_time": lookup(config, ("response", "installation_date")), # "2023-10-13 04:01:45 +0800" + "up_time_seconds": None, # "1541h38m20.998412744s" + "is_new": False, + "version": lookup(config, ("response", "version")), # 23.28.2 27626f98 + "git_hash": "27626f98a66cad5c665bbe1d4d788cdb3e94fd34", + "commission_count": 0, + "device_type": lookup(config, ("response", "components", "gateway")), # teg + "teg_type": "unknown", + "sync_type": "v2.1", + "cellular_disabled": False, + "can_reboot": True + } + + elif api == '/api/system_status/grid_status': + power = self.get_site_power() + if power is None: + data = None + else: + if lookup(power, ("response", "island_status")) == "on_grid": + grid_status = "SystemGridConnected" + else: # off_grid or off_grid_unintentional + grid_status = "SystemIslandedActive" + data = { + "grid_status": grid_status, # SystemIslandedActive or SystemTransitionToGrid + "grid_services_active": lookup(power, ("response", "grid_services_active")) # true when participating in VPP event + } + + elif api == '/api/site_info/site_name': + config = self.get_site_config() + if config is None: + data = None + else: + sitename = lookup(config, ("response", "site_name")) + tz = lookup(config, ("response", "installation_time_zone")) + data = { + "site_name": sitename, + "timezone": tz + } + + elif api == '/api/site_info': + config = self.get_site_config() + if config is None: + data = None + else: + nameplate_power = int(lookup(config, ("response", "nameplate_power")) or 0) / 1000 + nameplate_energy = int(lookup(config, ("response", "nameplate_energy")) or 0) / 1000 + max_site_meter_power_ac = lookup(config, ("response", "max_site_meter_power_ac")) + min_site_meter_power_ac = lookup(config, ("response", "min_site_meter_power_ac")) + utility = lookup(config, ("response", "tariff_content", "utility")) + sitename = lookup(config, ("response", "site_name")) + tz = lookup(config, ("response", "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 + } + } + + elif api == '/api/devices/vitals': + # Protobuf payload - not implemented - use /vitals instead + data = None + + elif api == '/vitals': + # Simulated Vitals + config = self.get_site_config() + power = self.get_site_power() + if config is None or power is None: + data = None + else: + din = lookup(config, ("response", "id")) + parts = din.split("--") + if len(parts) == 2: + partNumber = parts[0] + serialNumber = parts[1] + else: + partNumber = None + serialNumber = None + version = lookup(config, ("response", "version")) + # Get grid status + island_status = lookup(power, ("response", "island_status")) + if island_status == "on_grid": + alert = "SystemConnectedToGrid" + elif island_status == "off_grid_intentional": + alert = "ScheduledIslandContactorOpen" + else: + alert = "UnscheduledIslandContactorOpen" + data = { + f'STSTSM--{partNumber}--{serialNumber}': { + 'partNumber': partNumber, + 'serialNumber': serialNumber, + 'manufacturer': 'Simulated', + 'firmwareVersion': version, + 'lastCommunicationTime': int(time.time()), + 'teslaEnergyEcuAttributes': { + 'ecuType': 207 + }, + 'STSTSM-Location': 'Simulated', + 'alerts': [ + alert + ] + } + } + + elif api in ['/api/system_status/soe']: + battery = self.get_battery() + if battery is None: + data = None + else: + percentage_charged = lookup(battery, ("response", "percentage_charged")) or 0 + # percentage_charged is scaled to keep 5% buffer at bottom + soe = (percentage_charged + (5 / 0.95)) * 0.95 + data = { + "percentage": soe + } + + elif api == '/api/meters/aggregates': + config = self.get_site_config() + power = self.get_site_power() + if config is None or power is None: + data = None + else: + timestamp = lookup(power, ("response", "timestamp")) + solar_power = lookup(power, ("response", "solar_power")) + battery_power = lookup(power, ("response", "battery_power")) + load_power = lookup(power, ("response", "load_power")) + grid_power = lookup(power, ("response", "grid_power")) + battery_count = lookup(config, ("response", "battery_count")) + inverters = lookup(config, ("response", "components", "inverters")) + if inverters is not None: + solar_inverters = len(inverters) + elif lookup(config, ("response", "components", "solar")): + solar_inverters = 1 + else: + solar_inverters = 0 + data = { + "site": { + "last_communication_time": timestamp, + "instant_power": grid_power, + "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": timestamp, + "instant_power": battery_power, + "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": battery_count, + "instant_total_current": 0 + }, + "load": { + "last_communication_time": timestamp, + "instant_power": load_power, + "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": timestamp, + "instant_power": solar_power, + "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": solar_inverters, + "instant_total_current": 0 + } + } + + elif api == '/api/operation': + config = self.get_site_config() + if config is None: + data = None + else: + default_real_mode = lookup(config, ("response", "default_real_mode")) + backup_reserve_percent = lookup(config, ("response", "backup_reserve_percent")) or 0 + # backup_reserve_percent is scaled to keep 5% buffer at bottom + backup = (backup_reserve_percent + (5 / 0.95)) * 0.95 + data = { + "real_mode": default_real_mode, + "backup_reserve_percent": backup + } + + elif api == '/api/system_status': + power = self.get_site_power() + config = self.get_site_config() + battery = self.get_battery() + if power is None or config is None or battery is None: + data = None + else: + timestamp = lookup(power, ("response", "timestamp")) + solar_power = lookup(power, ("response", "solar_power")) + battery_power = lookup(power, ("response", "battery_power")) + load_power = lookup(power, ("response", "load_power")) + grid_services_power = lookup(power, ("response", "grid_services_power")) + grid_status = lookup(power, ("response", "grid_status")) + grid_services_active = lookup(power, ("response", "grid_services_active")) + battery_count = lookup(config, ("response", "battery_count")) + total_pack_energy = lookup(battery, ("response", "total_pack_energy")) + energy_left = lookup(battery, ("response", "energy_left")) + nameplate_power = lookup(config, ("response", "nameplate_power")) + nameplate_energy = lookup(config, ("response", "nameplate_energy")) + if lookup(power, ("response", "island_status")) == "on_grid": + grid_status = "SystemGridConnected" + else: # off_grid or off_grid_unintentional + grid_status = "SystemIslandedActive" + data = { # TODO: Fill in 0 values + "command_source": "Configuration", + "battery_target_power": 0, + "battery_target_reactive_power": 0, + "nominal_full_pack_energy": total_pack_energy, + "nominal_energy_remaining": energy_left, + "max_power_energy_remaining": 0, # TODO: Calculate + "max_power_energy_to_be_charged": 0, # TODO: Calculate + "max_charge_power": nameplate_power, + "max_discharge_power": nameplate_power, + "max_apparent_power": nameplate_power, + "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": grid_services_power, + "system_island_state": grid_status, + "available_blocks": battery_count, + "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": solar_power, + "score": 10000, + "blocks_controlled": battery_count, + "primary": True, + "auxiliary_load": 0, + "all_enable_lines_high": True, + "inverter_nominal_usable_power": 0, + "expected_energy_remaining": 0 + } + + ## Possible Actions + elif api == '/api/logout': + data = '{"status":"ok"}' + elif api == '/api/login/Basic': + data = '{"status":"ok"}' + + ## Static Mock Values + elif api == '/api/meters/site': + data = json.loads('[{"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}}]') + + elif api == '/api/meters/solar': + data = None + + elif api == '/api/auth/toggle/supported': + data = json.loads('{"toggle_auth_supported":true}') + + elif api == '/api/sitemaster': + data = json.loads('{"status":"StatusUp","running":true,"connected_to_tesla":true,"power_supply_mode":false,"can_reboot":"Yes"}') + + elif api == '/api/powerwalls': + data = json.loads('{"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}') + + elif api == '/api/customer/registration': + data = json.loads('{"privacy_notice":null,"limited_warranty":null,"grid_services":null,"marketing":null,"registered":true,"timed_out_registration":false}') + + elif api == '/api/system/update/status': + data = 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}') + + elif api == '/api/system_status/grid_faults': + data = json.loads('[]') + + elif api == '/api/site_info/grid_codes': + data = "TIMEOUT!" + + elif api == '/api/solars': + data = json.loads('[{"brand":"Tesla","model":"Solar Inverter 7.6","power_rating_watts":7600}]') + + elif api == '/api/solars/brands': + data = json.loads('["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)","ｶﾅﾃﾞｨｱﾝ ｿｰﾗｰ"]') + + elif api == '/api/customer': + data = json.loads('{"registered":true}') + + elif api == '/api/meters': + data = json.loads('[{"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]}]}]') + + elif api == '/api/installer': + data = json.loads('{"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"]}') + + elif api == '/api/networks': + data = json.loads('[{"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":""}}]') + + elif api == '/api/system/networks': + data = "TIMEOUT!" + + elif api == '/api/meters/readings': + data = "TIMEOUT!" + + elif api == '/api/synchrometer/ct_voltage_references': + data = json.loads('{"ct1":"Phase1","ct2":"Phase2","ct3":"Phase1"}') + + elif api == '/api/troubleshooting/problems': + data = json.loads('{"problems":[]}') + + else: + data = {"ERROR": f"Unknown API: {api}"} + + return data + + def setup(self): + """ + Set up the Tesla Cloud connection + """ + print("Tesla Account Setup") + print("-" * 60) + tuser = "" + # Check for .pypowerwall.auth file + if os.path.isfile(AUTHFILE): + print(" Found existing Tesla Cloud setup file ({})".format(AUTHFILE)) + with open(AUTHFILE) as json_file: + try: + data = json.load(json_file) + tuser = list(data.keys())[0] + print(f" Using Tesla User: {tuser}") + # Ask user if they want to overwrite the existing file + response = input("\n Overwrite existing file? [y/N]: ").strip() + if response.lower() == "y": + tuser = "" + os.remove(AUTHFILE) + else: + self.email = tuser + except Exception as err: + tuser = "" + + if tuser == "": + # Create new AUTHFILE + while True: + response = input("\n Email address: ").strip() + if "@" not in response: + print(" - Error: Invalid email address\n") + else: + tuser = response + break + + # Update the Tesla User + self.email = tuser + + # Create Tesla instance + tesla = Tesla(self.email, cache_file=AUTHFILE) + + if not tesla.authorized: + # Login to Tesla account and cache token + state = tesla.new_state() + code_verifier = tesla.new_code_verifier() + + try: + print("Open the below address in your browser to login.\n") + print(tesla.authorization_url(state=state, code_verifier=code_verifier)) + except Exception as err: + log.error(f"Connection failure - {repr(err)}") + + print("\nAfter login, paste the URL of the 'Page Not Found' webpage below.\n") + + tesla.close() + tesla = Tesla(self.email, state=state, code_verifier=code_verifier, cache_file=AUTHFILE) + + if not tesla.authorized: + try: + tesla.fetch_token(authorization_response=input("Enter URL after login: ")) + print("-" * 60) + except Exception as err: + log.error(f"Connection failure - {repr(err)}") + return False + + # Connect to Tesla Cloud + self.siteid = None + if not self.connect(): + print("\nERROR: Failed to connect to Tesla Cloud") + return False + + sites = self.getsites() + if sites is None or len(sites) == 0: + print("\nERROR: No sites found for %s" % self.email) + return False + + print(f"\n{len(sites)} Sites Found (* = default)") + print("-"*60) + + # Check for existing site file + if os.path.isfile(SITEFILE): + with open(SITEFILE) as file: + try: + self.siteid = int(file.read()) + except: + self.siteid = 0 + + idx = 1 + self.siteindex = 0 + siteids = [] + for s in sites: + if s["energy_site_id"] == self.siteid: + sitelabel = "*" + self.siteindex = idx - 1 + else: + sitelabel = " " + siteids.append(s["energy_site_id"]) + print(" %s%d - %s (%s) - Type: %s" % (sitelabel, idx, s["site_name"], + s["energy_site_id"], s["resource_type"])) + idx += 1 + # Ask user to select a site + while True: + response = input(f"\n Select a site [{self.siteindex+1}]: ").strip() + if response.isdigit(): + idx = int(response) + if idx >= 1 and idx < (len(sites)+1): + self.siteindex = idx -1 + break + else: + print(f" - Invalid: {response} is not a valid site number") + else: + break + # Lookup the site id + self.siteid = siteids[self.siteindex] + self.site = sites[self.siteindex] + print("\nSelected site %d - %s (%s)" % (self.siteindex+1, sites[self.siteindex]["site_name"], self.siteid)) + # Write the site id to the sitefile + with open(SITEFILE, "w") as f: + f.write(str(self.siteid)) + + return True + +if __name__ == "__main__": + + # Test code + set_debug(False) + # Check for .pypowerwall.auth file + if os.path.isfile(AUTHFILE): + # Read the json file + with open(AUTHFILE) as json_file: + try: + data = json.load(json_file) + tuser = list(data.keys())[0] + print(f"Using Tesla User: {tuser}") + except Exception as err: + tuser = None + + while not tuser: + response = input("Tesla User Email address: ").strip() + if "@" not in response: + print("Invalid email address\n") + else: + tuser = response + break + + cloud = TeslaCloud(tuser) + + if not cloud.connect(): + print("Failed to connect to Tesla Cloud") + cloud.setup() + if not cloud.connect(): + print("Failed to connect to Tesla Cloud") + exit(1) + + print("Connected to Tesla Cloud") + + #print("\nSite Data") + #sites = cloud.getsites() + #print(sites) + + #print("\Battery") + #r = cloud.get_battery() + #print(r) + + #print("\Site Power") + #r = cloud.get_site_power() + #print(r) + + #print("\Site Config") + #r = cloud.get_site_config() + #print(r) + + # Test Poll + # '/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'] + for i in items: + print(f"poll({i}):") + print(cloud.poll(i)) + print("\n") diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..79ead2f --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +# +requests +protobuf +teslapy \ No newline at end of file diff --git a/setup.py b/setup.py index 2d876b0..c52e395 100644 --- a/setup.py +++ b/setup.py @@ -18,9 +18,9 @@ install_requires=[ 'requests', 'protobuf', + 'teslapy', ], classifiers=[ - "Programming Language :: Python :: 2", "Programming Language :: Python :: 3", "License :: OSI Approved :: MIT License", "Operating System :: OS Independent",