Merge pull request #100 from jasonacox/v0.10.3

v0.10.3 - TEDAPI Connect

README.md

@@ -43,7 +43,23 @@ 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).
+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, the Powerwall 3 does not have a local API but you can access it via the cloud (see options 2 and 3).
+
+Extended Device Vitals Metrics: With version v0.10.0+, pypowerwall can be set to access the TEDAPI on the Gateway to pull additional metrics. However, you will need the Gateway Password (often found on the QR sticker on the Powerwall Gateway). Additionally, your computer will need network access to the Gateway IP (192.168.91.1). You can have your computer join the Gateway local WiFi or you can add a route:
+
+```bash
+# Example - Change 192.168.0.100 to the IP address of Powerwall Gateway on your LAN
+
+# Linux Ubuntu and RPi - Can add to /etc/rc.local for persistence 
+sudo ip route add 192.168.91.1 via 192.168.0.100
+
+# MacOS 
+sudo route add -host 192.168.91.1 192.168.0.100 # Temporary 
+networksetup -setadditionalroutes Wi-Fi 192.168.91.1 255.255.255.255 192.168.0.100 # Persistent
+
+# Windows - Using persistence flag - Administrator Shell
+route -p add 192.168.91.1 mask 255.255.255.255 192.168.0.100
+```
 
 ### FleetAPI Setup - Option 2
 

RELEASE.md

@@ -1,5 +1,21 @@
 # RELEASE NOTES
 
+## v0.10.3 - TEDAPI Connect
+
+* Update `setup.py` to include dependencies on `protobuf>=3.20.0`.
+* Add TEDAPI `connect()` logic to better validate Gateway endpoint access.
+* Add documentation for TEDAPI setup.
+* Update CLI to support TEDAPI calls.
+
+```bash
+# Connect to TEDAPI and pull data
+python3 -m pypowerwall tedapi
+
+# Direct call to TEDAPI class test function (optional password)
+python3 -m pypowerwall.tedapi GWPASSWORD
+python3 -m pypowerwall.tedapi --debug
+```
+
 ## v0.10.2 - FleetAPI Hotfix
 
 * Fix FleetAPI setup script as raised in https://github.com/jasonacox/pypowerwall/issues/98.

proxy/RELEASE.md

@@ -1,5 +1,9 @@
 ## pyPowerwall Proxy Release Notes
 
+### Proxy t60 (9 Jun 2024)
+
+* Add error handling for `/csv` API to accommodate `None` data points.
+
 ### Proxy t59 (8 Jun 2024)
 
 * Minor fix to send less ambiguous debug information during client disconnects.

proxy/requirements.txt

@@ -1,2 +1,2 @@
-pypowerwall==0.10.2
+pypowerwall==0.10.3
 bs4==0.0.2

proxy/server.py

@@ -55,7 +55,7 @@
 from transform import get_static, inject_js
 from urllib.parse import urlparse, parse_qs
 
-BUILD = "t59"
+BUILD = "t60"
 ALLOWLIST = [
     '/api/status', '/api/site_info/site_name', '/api/meters/site',
     '/api/meters/solar', '/api/sitemaster', '/api/powerwalls',
@@ -330,7 +330,7 @@ def do_GET(self):
         elif self.path == '/csv':
             # Grid,Home,Solar,Battery,Level - CSV
             contenttype = 'text/plain; charset=utf-8'
-            batterylevel = pw.level()
+            batterylevel = pw.level() or 0
             grid = pw.grid() or 0
             solar = pw.solar() or 0
             battery = pw.battery() or 0

pypowerwall/__init__.py

@@ -17,7 +17,7 @@
 
  Classes
     Powerwall(host, password, email, timezone, pwcacheexpire, timeout, poolmaxsize, 
-        cloudmode, siteid, authpath, authmode)
+        cloudmode, siteid, authpath, authmode, cachefile, fleetapi, auto_select, retry_modes, gw_pwd)
 
  Parameters
     host                      # Hostname or IP of the Tesla gateway
@@ -36,6 +36,8 @@
     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)
+    retry_modes = False       # If True, retry connection to Powerwall
+    gw_pwd = None             # TEG Gateway password (used for local mode access to tedapi)
     
  Functions 
     poll(api, json, force)    # Return data from Powerwall api (dict if json=True, bypass cache force=True)
@@ -82,7 +84,7 @@
 from typing import Union, Optional
 import time
 
-version_tuple = (0, 10, 2)
+version_tuple = (0, 10, 3)
 version = __version__ = '%d.%d.%d' % version_tuple
 __author__ = 'jasonacox'
 

pypowerwall/__main__.py

@@ -41,6 +41,8 @@
 
 setup_args = subparsers.add_parser("fleetapi", help='Setup Tesla FleetAPI for Cloud Mode access')
 
+setup_args = subparsers.add_parser("tedapi", help='Test TEDAPI connection to Powerwall Gateway')
+
 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}]")
@@ -96,6 +98,7 @@
     else:
         print("ERROR: Failed to setup Tesla Cloud Mode")
         exit(1)
+
 # FleetAPI Mode Setup
 elif command == 'fleetapi':
     from pypowerwall import PyPowerwallFleetAPI
@@ -108,6 +111,12 @@
     else:
         print("Setup Aborted.")
         exit(1)
+
+# TEDAPI Test
+elif command == 'tedapi':
+    from pypowerwall.tedapi.__main__ import run_tedapi_test
+    run_tedapi_test(auto=True, debug=args.debug)
+
 # Run Scan
 elif command == 'scan':
     from pypowerwall import scan
@@ -118,6 +127,7 @@
     hosts = args.hosts
     timeout = args.timeout
     scan.scan(color, timeout, hosts, ip)
+
 # Set Powerwall Mode
 elif command == 'set':
     # If no arguments, print usage
@@ -146,6 +156,7 @@
         current = float(pw.level())
         print("Setting Powerwall Reserve to Current Charge Level %s" % current)
         pw.set_reserve(current)
+
 # Get Powerwall Mode
 elif command == 'get':
     import pypowerwall

pypowerwall/tedapi/__init__.py

@@ -7,7 +7,16 @@
  TEDAPI on 192.168.91.1 as used by the Tesla One app.
 
  Class:
-    TEDAPI - Tesla TEDAPI Class
+    TEDAPI(gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, timeout: int = 5,
+              pwconfigexpire: int = 300, host: str = GW_IP) - Initialize TEDAPI
+    
+ Parameters:
+    gw_pwd - Powerwall Gateway Password
+    debug - Enable Debug Output
+    pwcacheexpire - Cache Expiration in seconds
+    timeout - API Timeout in seconds
+    pwconfigexpire - Configuration Cache Expiration in seconds
+    host - Powerwall Gateway IP Address (default: 192.168.91.1)
 
  Functions:
     get_din() - Get the DIN from the Powerwall Gateway
@@ -39,6 +48,7 @@
 import time
 from pypowerwall import __version__
 import math
+import sys
 
 # TEDAPI Fixed Gateway IP Address
 GW_IP = "192.168.91.1"
@@ -48,6 +58,8 @@
 
 # Setup Logging
 log = logging.getLogger(__name__)
+log.debug('%s version %s', __name__, __version__)
+log.debug('Python %s on %s', sys.version, sys.platform)
 
 # Utility Functions
 def lookup(data, keylist):
@@ -64,29 +76,42 @@ def lookup(data, keylist):
     return data
 
 # TEDAPI Class
-
 class TEDAPI:
-    def __init__(self, gw_pwd, debug=False, pwcacheexpire: int = 5, timeout: int = 5, pwconfigexpire: int = 300) -> None:
+    def __init__(self, gw_pwd: str, debug: bool = False, pwcacheexpire: int = 5, timeout: int = 5, 
+                 pwconfigexpire: int = 300, host: str = GW_IP) -> None:
         self.debug = debug 
         self.pwcachetime = {}  # holds the cached data timestamps for api
         self.pwcacheexpire = pwcacheexpire  # seconds to expire status cache
         self.pwconfigexpire = pwconfigexpire  # seconds to expire config cache
         self.pwcache = {}  # holds the cached data for api
         self.timeout = timeout
         self.pwcooldown = 0
+        self.gw_ip = host
         self.din = None
+        self.pw3 = False # Powerwall 3 Gateway only supports TEDAPI
         if not gw_pwd:
             raise ValueError("Missing gw_pwd")
         if self.debug:
-            log.setLevel(logging.DEBUG)
+            self.set_debug(True)
         self.gw_pwd = gw_pwd
         # Connect to Powerwall Gateway
         if not self.connect():
             log.error("Failed to connect to Powerwall Gateway")
-            raise ValueError("Failed to connect to Powerwall Gateway")
-
+
     # TEDAPI Functions
 
+    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 get_din(self, force=False):
         """
         Get the DIN from the Powerwall Gateway
@@ -102,13 +127,16 @@ def get_din(self, force=False):
             return None
         # Fetch DIN from Powerwall
         log.debug("Fetching DIN from Powerwall...")
-        url = f'https://{GW_IP}/tedapi/din'
+        url = f'https://{self.gw_ip}/tedapi/din'
         r = requests.get(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False)
         if r.status_code in BUSY_CODES:
             # Rate limited - Switch to cooldown mode for 5 minutes
             self.pwcooldown = time.perf_counter() + 300
             log.error('Possible Rate limited by Powerwall at - Activating 5 minute cooldown')
             return None
+        if r.status_code == 403:
+            log.error("Access Denied: Check your Gateway Password")
+            return None
         if r.status_code != 200:
             log.error(f"Error fetching DIN: {r.status_code}")
             return None
@@ -158,6 +186,11 @@ def get_config(self,force=False):
                 # Rate limited - return None
                 log.debug('Rate limit cooldown period - Pausing API calls')
                 return None
+        # Check Connection
+        if not self.din:
+            if not self.connect():
+                log.error("Not Connected - Unable to get configuration")
+                return None
         # Fetch Configuration from Powerwall
         log.debug("Get Configuration from Powerwall")
         # Build Protobuf to fetch config
@@ -168,7 +201,7 @@ def get_config(self,force=False):
         pb.message.config.send.num = 1
         pb.message.config.send.file = "config.json"
         pb.tail.value = 1
-        url = f'https://{GW_IP}/tedapi/v1'
+        url = f'https://{self.gw_ip}/tedapi/v1'
         r = requests.post(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False,
             headers={'Content-type': 'application/octet-string'},
             data=pb.SerializeToString())
@@ -245,6 +278,11 @@ def get_status(self, force=False):
                 # Rate limited - return None
                 log.debug('Rate limit cooldown period - Pausing API calls')
                 return None
+        # Check Connection
+        if not self.din:
+            if not self.connect():
+                log.error("Not Connected - Unable to get status")
+                return None
         # Fetch Current Status from Powerwall
         log.debug("Get Status from Powerwall")
         # Build Protobuf to fetch status
@@ -258,7 +296,7 @@ def get_status(self, force=False):
         pb.message.payload.send.code = b'0\201\206\002A\024\261\227\245\177\255\265\272\321r\032\250\275j\305\030\2300\266\022B\242\264pO\262\024vd\267\316\032\f\376\322V\001\f\177*\366\345\333g_/`\v\026\225_qc\023$\323\216y\276~\335A1\022x\002Ap\a_\264\037]\304>\362\356\005\245V\301\177*\b\307\016\246]\037\202\242\353I~\332\317\021\336\006\033q\317\311\264\315\374\036\365s\272\225\215#o!\315z\353\345z\226\365\341\f\265\256r\373\313/\027\037'
         pb.message.payload.send.b.value = "{}"
         pb.tail.value = 1
-        url = f'https://{GW_IP}/tedapi/v1'
+        url = f'https://{self.gw_ip}/tedapi/v1'
         r = requests.post(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False,
             headers={'Content-type': 'application/octet-string'},
             data=pb.SerializeToString())
@@ -290,20 +328,21 @@ def connect(self):
         Connect to the Powerwall Gateway
         """
         # Test IP Connection to Powerwall Gateway
-        log.debug(f"Testing Connection to Powerwall Gateway: {GW_IP}")
-        url = f'https://{GW_IP}'
-        try:
-            r = requests.get(url, verify=False, timeout=5)
-        except requests.exceptions.RequestException as e:
-            r = False
-            log.error("ERROR: Powerwall not Found",
-                      f"Try: sudo route add -host <Powerwall_IP> {GW_IP}")
-        if r:
-            # Attempt to fetch DIN from Powerwall
-            self.din = self.get_din()
-            return True
+        log.debug(f"Testing Connection to Powerwall Gateway: {self.gw_ip}")
+        url = f'https://{self.gw_ip}'
         self.din = None
-        return False
+        try:
+            resp = requests.get(url, verify=False, timeout=5)
+            if resp.status_code != 200:
+                # Connected but appears to be Powerwall 3
+                log.debug("Detected Powerwall 3 Gateway")
+                self.pw3 = True
+            self.din = self.get_din() 
+        except Exception as e:
+            log.error(f"Unable to connect to Powerwall Gateway {self.gw_ip}")
+            log.error("Please verify your your host has a route to the Gateway.")
+            log.error(f"Error Details: {e}")
+        return self.din
 
     # Handy Function to access Powerwall Status
 
@@ -377,7 +416,7 @@ def calculate_dc_power(V, I):
         header["VITALS"] = {
             "text": "Device vitals generated from Tesla Powerwall Gateway TEDAPI",
             "timestamp": time.time(),
-            "gateway": GW_IP,
+            "gateway": self.gw_ip,
             "pyPowerwall": __version__,
         }