v0.10.0 - New Device Vitals

.github/workflows/simtest.yml

@@ -10,11 +10,11 @@ jobs:
     name: "Python ${{ matrix.python-version }}"
     runs-on: "ubuntu-20.04"
     env:
-      USING_COVERAGE: '3.6,3.8'
+      USING_COVERAGE: '3.8'
 
     strategy:
       matrix:
-        python-version: ["3.6", "3.7", "3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
 
     services:
       simulator:

.github/workflows/test.yml

@@ -10,11 +10,11 @@ jobs:
     name: "Python ${{ matrix.python-version }}"
     runs-on: "ubuntu-20.04"
     env:
-      USING_COVERAGE: '3.6,3.8'
+      USING_COVERAGE: '3.8'
 
     strategy:
       matrix:
-        python-version: ["3.6", "3.7", "3.8", "3.9", "3.10", "3.11"]
+        python-version: ["3.7", "3.8", "3.9", "3.10", "3.11"]
 
     steps:
       - uses: "actions/checkout@v2"

.gitignore

@@ -156,3 +156,4 @@ proxy/teslapy
 .auth
 tools/tedapi/status.json
 tools/tedapi/config.json
+j

RELEASE.md

@@ -1,5 +1,39 @@
 # RELEASE NOTES
 
+## v0.10.0 - New Device Vitals
+
+* Add support for `/tedapi` API access on Gateway (requires connectivity to 192.168.91.1 GW and Gateway Password) with access to "config" and "status" data.
+* Adds drop-in replacement for depreciated `/vitals` API and payload using the new TEDAPI class. This allows easy access to Powerwall device vitals.
+* Proxy update to t58 to support TEDAPI with environmental variable `PW_GW_PWD` for Gateway Password. Also added FleetAPI, Cloud and TEDAPI specific GET calls, `/fleetapi`, `/cloud`, and `/tedapi` respectively.
+
+```python
+# How to Activate the TEDAPI Mode
+import pypowerwall
+
+gw_pwd = "GW_PASSWORD" # Gateway Passowrd usually on QR code on Gateway
+
+host = "192.168.91.1" # Direct Connect to GW
+pw = pypowerwall.Powerwall(host,password,email,timezone,gw_pwd=gw_pwd)
+print(pw.vitals())
+```
+
+```python
+# New TEDAPI Class
+import pypowerwall.tedapi
+
+tedapi = pypowerwall.tedapi.TEDAPI("GW_PASSWORD")
+
+config = tedapi.get_config()
+status = tedapi.get_status()
+
+meterAggregates = status.get('control', {}).get('meterAggregates', [])
+for meter in meterAggregates:
+    location = meter.get('location', 'Unknown').title()
+    realPowerW = int(meter.get('realPowerW', 0))
+    print(f"   - {location}: {realPowerW}W")
+
+```
+
 ## v0.9.1 - Bug Fixes and Updates
 
 * Fix bug in time_remaining_hours() and convert print statements in FleetAPI to log messages.

proxy/RELEASE.md

@@ -1,5 +1,10 @@
 ## pyPowerwall Proxy Release Notes
 
+### Proxy t58 (2 Jun 2024)
+
+* Add support for pypowerwall v0.10.0 and TEDAPI with environmental variable `PW_GW_PWD` for Gateway Password. This unlocks new device vitals metrics (as seen with `/vitals`). It requires the user to have access to the Powerwall Gateway at 192.168.91.1, either via WiFi for by adding a route to their host or network. 
+* Add FleetAPI, Cloud and TEDAPI specific GET calls, `/fleetapi`, `/cloud`, and `/tedapi` respectively.
+
 ### Proxy t57 (15 May 2024)
 
 * Add pypowerwall v0.9.0 capabilities, specifically supporting Tesla FleetAPI for cloud connections (main data and control).

proxy/requirements.txt

@@ -1,2 +1,2 @@
-pypowerwall==0.9.1
+pypowerwall==0.10.0
 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 = "t57"
+BUILD = "t58"
 ALLOWLIST = [
     '/api/status', '/api/site_info/site_name', '/api/meters/site',
     '/api/meters/solar', '/api/sitemaster', '/api/powerwalls',
@@ -94,6 +94,7 @@
     cf = os.path.join(authpath, ".powerwall")
 cachefile = os.getenv("PW_CACHE_FILE", cf)
 control_secret = os.getenv("PW_CONTROL_SECRET", "")
+gw_pwd = os.getenv("PW_GW_PWD", None)
 
 # Global Stats
 proxystats = {
@@ -112,6 +113,7 @@
     'site_name': "",
     'cloudmode': False,
     'fleetapi': False,
+    'tedapi': False,
     'siteid': None,
     'counter': 0
 }
@@ -170,7 +172,7 @@ def get_value(a, key):
                                timeout, pool_maxsize, siteid=siteid,
                                authpath=authpath, authmode=authmode,
                                cachefile=cachefile, auto_select=True, 
-                               retry_modes=True)
+                               retry_modes=True, gw_pwd=gw_pwd)
 except Exception as e:
     log.error(e)
     log.error("Fatal Error: Unable to connect. Please fix config and restart.")
@@ -202,6 +204,9 @@ def get_value(a, key):
     proxystats['mode'] = "Local"
     log.info("pyPowerwall Proxy Server - Local Mode")
     log.info("Connected to Energy Gateway %s (%s)" % (host, site_name.strip()))
+    if pw.tedapi:
+        proxystats['tedapi'] = True
+        log.info("TEDAPI Mode Enabled for Device Vitals")
 
 pw_control = None
 if control_secret:
@@ -461,11 +466,11 @@ def do_GET(self):
                     pod["PW%d_i_out" % idx] = get_value(block, "i_out")
                     pod["PW%d_energy_charged" % idx] = get_value(block, "energy_charged")
                     pod["PW%d_energy_discharged" % idx] = get_value(block, "energy_discharged")
-                    pod["PW%d_off_grid" % idx] = int(get_value(block, "off_grid"))
-                    pod["PW%d_vf_mode" % idx] = int(get_value(block, "vf_mode"))
-                    pod["PW%d_wobble_detected" % idx] = int(get_value(block, "wobble_detected"))
-                    pod["PW%d_charge_power_clamped" % idx] = int(get_value(block, "charge_power_clamped"))
-                    pod["PW%d_backup_ready" % idx] = int(get_value(block, "backup_ready"))
+                    pod["PW%d_off_grid" % idx] = int(get_value(block, "off_grid") or 0)
+                    pod["PW%d_vf_mode" % idx] = int(get_value(block, "vf_mode") or 0)
+                    pod["PW%d_wobble_detected" % idx] = int(get_value(block, "wobble_detected") or 0)
+                    pod["PW%d_charge_power_clamped" % idx] = int(get_value(block, "charge_power_clamped") or 0)
+                    pod["PW%d_backup_ready" % idx] = int(get_value(block, "backup_ready") or 0)
                     pod["PW%d_OpSeqState" % idx] = get_value(block, "OpSeqState")
                     pod["PW%d_version" % idx] = get_value(block, "version")
                     idx = idx + 1
@@ -476,13 +481,13 @@ def do_GET(self):
                 v = vitals[device]
                 if device.startswith('TEPOD'):
                     pod["PW%d_name" % idx] = device
-                    pod["PW%d_POD_ActiveHeating" % idx] = int(get_value(v, 'POD_ActiveHeating'))
-                    pod["PW%d_POD_ChargeComplete" % idx] = int(get_value(v, 'POD_ChargeComplete'))
-                    pod["PW%d_POD_ChargeRequest" % idx] = int(get_value(v, 'POD_ChargeRequest'))
-                    pod["PW%d_POD_DischargeComplete" % idx] = int(get_value(v, 'POD_DischargeComplete'))
-                    pod["PW%d_POD_PermanentlyFaulted" % idx] = int(get_value(v, 'POD_PermanentlyFaulted'))
-                    pod["PW%d_POD_PersistentlyFaulted" % idx] = int(get_value(v, 'POD_PersistentlyFaulted'))
-                    pod["PW%d_POD_enable_line" % idx] = int(get_value(v, 'POD_enable_line'))
+                    pod["PW%d_POD_ActiveHeating" % idx] = int(get_value(v, 'POD_ActiveHeating') or 0)
+                    pod["PW%d_POD_ChargeComplete" % idx] = int(get_value(v, 'POD_ChargeComplete') or 0)
+                    pod["PW%d_POD_ChargeRequest" % idx] = int(get_value(v, 'POD_ChargeRequest') or 0)
+                    pod["PW%d_POD_DischargeComplete" % idx] = int(get_value(v, 'POD_DischargeComplete') or 0)
+                    pod["PW%d_POD_PermanentlyFaulted" % idx] = int(get_value(v, 'POD_PermanentlyFaulted') or 0)
+                    pod["PW%d_POD_PersistentlyFaulted" % idx] = int(get_value(v, 'POD_PersistentlyFaulted') or 0)
+                    pod["PW%d_POD_enable_line" % idx] = int(get_value(v, 'POD_enable_line') or 0)
                     pod["PW%d_POD_available_charge_power" % idx] = get_value(v, 'POD_available_charge_power')
                     pod["PW%d_POD_available_dischg_power" % idx] = get_value(v, 'POD_available_dischg_power')
                     pod["PW%d_POD_nom_energy_remaining" % idx] = get_value(v, 'POD_nom_energy_remaining')
@@ -544,6 +549,38 @@ def do_GET(self):
             # Simulate old API call and respond with empty list
             message = '{"problems": []}'
             # message = pw.poll('/api/troubleshooting/problems') or '{"problems": []}'
+        elif self.path.startswith('/tedapi'):
+            # TEDAPI Specific Calls
+            if pw.tedapi:
+                message = '{"error": "Use /tedapi/config, /tedapi/status"}'
+                if self.path == '/tedapi/config':
+                    message = json.dumps(pw.tedapi.get_config())
+                if self.path == '/tedapi/status':
+                    message = json.dumps(pw.tedapi.get_status())
+            else:
+                message = '{"error": "TEDAPI not enabled"}'
+        elif self.path.startswith('/cloud'):
+            # Cloud API Specific Calls
+            if pw.cloudmode and not pw.fleetapi:
+                message = '{"error": "Use /cloud/battery, /cloud/power, /cloud/config"}'
+                if self.path == '/cloud/battery':
+                    message = json.dumps(pw.client.get_battery())
+                if self.path == '/cloud/power':
+                    message = json.dumps(pw.client.get_site_power())
+                if self.path == '/cloud/config':
+                    message = json.dumps(pw.client.get_site_config())
+            else:
+                message = '{"error": "Cloud API not enabled"}'
+        elif self.path.startswith('/fleetapi'):
+            # FleetAPI Specific Calls
+            if pw.fleetapi:
+                message = '{"error": "Use /fleetapi/info, /fleetapi/status"}'
+                if self.path == '/fleetapi/info':
+                    message = json.dumps(pw.client.get_site_info())
+                if self.path == '/fleetapi/status':
+                    message = json.dumps(pw.client.get_live_status())
+            else:
+                message = '{"error": "FleetAPI not enabled"}'
         elif self.path in DISABLED:
             # Disabled API Calls
             message = '{"status": "404 Response - API Disabled"}'

pypowerwall/__init__.py

@@ -82,6 +82,10 @@
 from typing import Union, Optional
 import time
 
+version_tuple = (0, 10, 0)
+version = __version__ = '%d.%d.%d' % version_tuple
+__author__ = 'jasonacox'
+
 # noinspection PyPackageRequirements
 import urllib3
 
@@ -96,9 +100,6 @@
 
 urllib3.disable_warnings()  # Disable SSL warnings
 
-version_tuple = (0, 9, 1)
-version = __version__ = '%d.%d.%d' % version_tuple
-__author__ = 'jasonacox'
 
 log = logging.getLogger(__name__)
 log.debug('%s version %s', __name__, __version__)
@@ -121,7 +122,7 @@ class Powerwall(object):
     def __init__(self, host="", password="", email="nobody@nowhere.com",
                  timezone="America/Los_Angeles", pwcacheexpire=5, timeout=5, poolmaxsize=10,
                  cloudmode=False, siteid=None, authpath="", authmode="cookie", cachefile=".powerwall",
-                 fleetapi=False, auto_select=False, retry_modes=False):
+                 fleetapi=False, auto_select=False, retry_modes=False, gw_pwd=None):
         """
         Represents a Tesla Energy Gateway Powerwall device.
 
@@ -142,6 +143,7 @@ def __init__(self, host="", password="", email="nobody@nowhere.com",
             fleetapi     = If True, use Tesla FleetAPI for data (default is False)
             auto_select  = If True, select the best available mode to connect (default is False)
             retry_modes  = If True, retry connection to Powerwall
+            gw_pwd       = TEG Gateway password (used for local mode access to tedapi)
         """
 
         # Attributes
@@ -165,6 +167,8 @@ def __init__(self, host="", password="", email="nobody@nowhere.com",
         self.fleetapi = fleetapi
         self.retry_modes = retry_modes
         self.mode = "unknown"
+        self.gw_pwd = gw_pwd # TEG Gateway password for TEDAPI mode
+        self.tedapi = False
 
         # Make certain assumptions here
         if not self.host:
@@ -226,9 +230,11 @@ def connect(self, retry=False) -> bool:
             if self.mode == "local":
                 try:
                     self.client = PyPowerwallLocal(self.host, self.password, self.email, self.timezone, self.timeout,
-                                               self.pwcacheexpire, self.poolmaxsize, self.authmode, self.cachefile)
+                                               self.pwcacheexpire, self.poolmaxsize, self.authmode, self.cachefile,
+                                               self.gw_pwd)
                     self.client.authenticate()
                     self.cloudmode = self.fleetapi = False
+                    self.tedapi = self.client.tedapi
                     return True
                 except Exception as exc:
                     log.debug(f"Failed to connect using Local mode: {exc} - trying fleetapi mode.")

pypowerwall/cloud/pypowerwall_cloud.py

@@ -363,7 +363,7 @@ def get_site_power(self, force: bool = False):
                 "backup_capable": true,
                 "battery_power": -220,
                 "load_power": 1070,
-                "grid_status": "Active",
+                "grid_status": "Active", # Solar Only will use "Unknown"
                 "grid_services_active": false,
                 "grid_power": 0,
                 "grid_services_power": 0,
@@ -535,7 +535,10 @@ def get_api_system_status_grid_status(self, **kwargs) -> Optional[Union[dict, li
         if power is None:
             data = None
         else:
-            if lookup(power, ("response", "grid_status")) == "Active":
+            if lookup(power, ("response", "grid_status")) in ["Active", "Unknown"]:
+                grid_status = "SystemGridConnected"
+            elif not lookup(power, ("response", "grid_status")):
+                # If no grid_status, assume on_grid
                 grid_status = "SystemGridConnected"
             else:  # off_grid or off_grid_unintentional
                 grid_status = "SystemIslandedActive"
@@ -630,7 +633,7 @@ def get_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]:
                 alert = "UnscheduledIslandContactorOpen"
             else:
                 alert = ""
-                if lookup(power, ("response", "grid_status")) == "Active":
+                if lookup(power, ("response", "grid_status")) in ["Active", "Unknown"]:
                     alert = "SystemConnectedToGrid"
             data = {
                 f'STSTSM--{part_number}--{serial_number}': {
@@ -733,7 +736,7 @@ def get_api_system_status(self, **kwargs) -> Optional[Union[dict, list, str, byt
             else:  # off_grid or off_grid_unintentional
                 grid_status = "SystemIslandedActive"
                 # "grid_status": "Active"
-                if lookup(power, ("response", "grid_status")) == "Active":
+                if lookup(power, ("response", "grid_status")) in ["Active", "Unknown"]:
                     grid_status = "SystemGridConnected"
             data = API_SYSTEM_STATUS_STUB  # TODO: see inside API_SYSTEM_STATUS_STUB definition
             data.update({

pypowerwall/fleetapi/pypowerwall_fleetapi.py

@@ -362,7 +362,7 @@ def get_live_status(self):
             'backup_capable': True,
             'battery_power': 4080,
             'load_power': 4080,
-            'grid_status': 'Active',
+            'grid_status': 'Active', # Solar Only will use "Unknown"
             'grid_services_active': False,
             'grid_power': 0,
             'grid_services_power': 0,
@@ -422,7 +422,7 @@ def get_api_system_status_grid_status(self, **kwargs) -> Optional[Union[dict, li
         if power is None:
             data = None
         else:
-            if power.get("grid_status") == "Active":
+            if not not power.get("grid_status") or power.get("grid_status") in ["Active", "Unknown"]:
                 grid_status = "SystemGridConnected"
             else:  # off_grid or off_grid_unintentional
                 grid_status = "SystemIslandedActive"
@@ -517,7 +517,7 @@ def get_vitals(self, **kwargs) -> Optional[Union[dict, list, str, bytes]]:
                 alert = "UnscheduledIslandContactorOpen"
             else:
                 alert = ""
-                if power.get("grid_status") == "Active":
+                if power.get("grid_status") in ["Active", "Unknown"]:
                     alert = "SystemConnectedToGrid"
             data = {
                 f'STSTSM--{part_number}--{serial_number}': {
@@ -614,7 +614,7 @@ def get_api_system_status(self, **kwargs) -> Optional[Union[dict, list, str, byt
             else:  # off_grid or off_grid_unintentional
                 grid_status = "SystemIslandedActive"
                 # "grid_status": "Active"
-                if power.get("grid_status") == "Active":
+                if power.get("grid_status") in ["Active", "Unknown"]:
                     grid_status = "SystemGridConnected"
             data = API_SYSTEM_STATUS_STUB  # TODO: see inside API_SYSTEM_STATUS_STUB definition
             data.update({