Add Powerwall 3 Support #97

RELEASE.md

@@ -1,5 +1,21 @@
 # RELEASE NOTES
 
+## v0.10.4 - Powerwall 3 Support
+
+* Add local support for Powerwall 3 using TEDAPI. 
+* TEDAPI will activate in `hybrid` (using TEDAPI for vitals and existing local APIs for other metrics) or `full` (all data from TEDAPI) mode to provide better Powerwall 3 support.
+* The `full` mode will automatically activate when the customer `password` is blank and `gw_pwd` is set.
+
+```python
+import pypowerwall
+
+# Activate HYBRID mode (for Powerwall / 2 / + systems)
+pw = pypowerwall.Powerwall("192.168.91.1", password=PASSWORD, email=EMAIL, gw_pwd=PW_GW_PWD)
+
+# Activate FULL mode (for Powerwall 3 systems)
+pw = pypowerwall.Powerwall("192.168.91.1", gw_pwd=PW_GW_PWD)
+```
+
 ## v0.10.3 - TEDAPI Connect Update
 
 * Update `setup.py` to include dependencies on `protobuf>=3.20.0`.

proxy/RELEASE.md

@@ -1,5 +1,10 @@
 ## pyPowerwall Proxy Release Notes
 
+### Proxy t61 (9 Jun 2024)
+
+* Fix 404 bug that would throw error when user requested non-supported URI.
+* Add TEDAPI mode to stats.
+
 ### Proxy t60 (9 Jun 2024)
 
 * Add error handling for `/csv` API to accommodate `None` data points.

proxy/requirements.txt

@@ -1,2 +1,2 @@
-pypowerwall==0.10.3
+pypowerwall==0.10.4
 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 = "t60"
+BUILD = "t61"
 ALLOWLIST = [
     '/api/status', '/api/site_info/site_name', '/api/meters/site',
     '/api/meters/solar', '/api/sitemaster', '/api/powerwalls',
@@ -74,7 +74,7 @@
 # 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")
+password = os.getenv("PW_PASSWORD", "")
 email = os.getenv("PW_EMAIL", "email@example.com")
 host = os.getenv("PW_HOST", "")
 timezone = os.getenv("PW_TIMEZONE", "America/Los_Angeles")
@@ -114,6 +114,7 @@
     'cloudmode': False,
     'fleetapi': False,
     'tedapi': False,
+    'tedapi_mode': "off",
     'siteid': None,
     'counter': 0
 }
@@ -206,6 +207,7 @@ def get_value(a, key):
     log.info("Connected to Energy Gateway %s (%s)" % (host, site_name.strip()))
     if pw.tedapi:
         proxystats['tedapi'] = True
+        proxystats['tedapi_mode'] = pw.tedapi_mode
         log.info("TEDAPI Mode Enabled for Device Vitals")
 
 pw_control = None
@@ -635,24 +637,30 @@ def do_GET(self):
                     proxy_path = proxy_path[1:]
                 pw_url = "https://{}/{}".format(pw.host, proxy_path)
                 log.debug("Proxy request to: {}".format(pw_url))
-                if pw.authmode == "token":
-                    r = pw.client.session.get(
-                        url=pw_url,
-                        headers=pw.auth,
-                        verify=False,
-                        stream=True,
-                        timeout=pw.timeout
-                    )
-                else:
-                    r = pw.client.session.get(
-                        url=pw_url,
-                        cookies=pw.auth,
-                        verify=False,
-                        stream=True,
-                        timeout=pw.timeout
-                    )
-                fcontent = r.content
-                ftype = r.headers['content-type']
+                try:
+                    if pw.authmode == "token":
+                        r = pw.client.session.get(
+                            url=pw_url,
+                            headers=pw.auth,
+                            verify=False,
+                            stream=True,
+                            timeout=pw.timeout
+                        )
+                    else:
+                        r = pw.client.session.get(
+                            url=pw_url,
+                            cookies=pw.auth,
+                            verify=False,
+                            stream=True,
+                            timeout=pw.timeout
+                        )
+                    fcontent = r.content
+                    ftype = r.headers['content-type']
+                except AttributeError as e:
+                    # Display 404
+                    log.debug("File not found: {}".format(self.path))
+                    fcontent = bytes("Not Found", 'utf-8')
+                    ftype = "text/plain"
 
             # 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'):

pypowerwall/__init__.py

@@ -84,7 +84,7 @@
 from typing import Union, Optional
 import time
 
-version_tuple = (0, 10, 3)
+version_tuple = (0, 10, 4)
 version = __version__ = '%d.%d.%d' % version_tuple
 __author__ = 'jasonacox'
 
@@ -97,6 +97,7 @@
 from pypowerwall.local.pypowerwall_local import PyPowerwallLocal
 from pypowerwall.fleetapi.pypowerwall_fleetapi import PyPowerwallFleetAPI
 from pypowerwall.pypowerwall_base import parse_version, PyPowerwallBase
+from pypowerwall.tedapi.pypowerwall_tedapi import PyPowerwallTEDAPI
 from pypowerwall.fleetapi.fleetapi import CONFIGFILE
 from pypowerwall.cloud.pypowerwall_cloud import AUTHFILE
 
@@ -171,6 +172,7 @@ def __init__(self, host="", password="", email="nobody@nowhere.com",
         self.mode = "unknown"
         self.gw_pwd = gw_pwd # TEG Gateway password for TEDAPI mode
         self.tedapi = False
+        self.tedapi_mode = "off"  # off, full, hybrid
 
         # Make certain assumptions here
         if not self.host:
@@ -230,16 +232,28 @@ def connect(self, retry=False) -> bool:
                 time.sleep(30)
                 count = 0
             if self.mode == "local":
+                log.debug(f"password = {self.password}, gw_pwd = {self.gw_pwd}")
                 try:
-                    self.client = PyPowerwallLocal(self.host, self.password, self.email, self.timezone, self.timeout,
+                    if not self.password and self.gw_pwd:  # Use full TEDAPI mode
+                        log.debug("TEDAPI ** full **")
+                        self.tedapi_mode = "full"
+                        self.client = PyPowerwallTEDAPI(self.gw_pwd, pwcacheexpire=self.pwcacheexpire, 
+                                                        timeout=self.timeout, host=self.host)
+                    else:   
+                        self.tedapi_mode = "hybrid"                           
+                        self.client = PyPowerwallLocal(self.host, self.password, self.email, self.timezone, self.timeout,
                                                self.pwcacheexpire, self.poolmaxsize, self.authmode, self.cachefile,
                                                self.gw_pwd)
                     self.client.authenticate()
                     self.cloudmode = self.fleetapi = False
                     self.tedapi = self.client.tedapi
+                    if not self.tedapi:
+                        self.tedapi_mode = "off"
                     return True
                 except Exception as exc:
                     log.debug(f"Failed to connect using Local mode: {exc} - trying fleetapi mode.")
+                    self.tedapi = False
+                    self.tedapi_mode = "off"
                     self.mode = "fleetapi"
                     continue
             if self.mode == "fleetapi":

pypowerwall/local/pypowerwall_local.py

@@ -35,6 +35,7 @@ def __init__(self, host: str, password: str, email: str, timezone: str, timeout:
         self.vitals_api = True  # vitals api is available for local mode
         self.gw_pw = gw_pw  # Powerwall Gateway password for TEDAPI
         self.tedapi = None  # TEDAPI object
+        self.pw3 = False  # Powerwall 3 detected
 
     def authenticate(self):
         log.debug('Tesla local mode enabled')
@@ -77,6 +78,7 @@ def authenticate(self):
             self.tedapi = TEDAPI(self.gw_pw)
             if self.tedapi.connect():
                 log.debug('TEDAPI connected - Vitals metrics enabled')
+                self.pw3 = self.tedapi.pw3
             else:
                 log.debug('TEDAPI connection failed - continuing')
                 self.tedapi = None

pypowerwall/tedapi/__init__.py

@@ -346,17 +346,39 @@ def connect(self):
 
     # Handy Function to access Powerwall Status
 
+    def current_power(self, location=None, force=False):
+        """
+        Get the current power in watts for a location: 
+            BATTERY, SITE, LOAD, SOLAR, SOLAR_RGM, GENERATOR, CONDUCTOR
+        """
+        status = self.get_status(force)
+        power = lookup(status, ['control', 'meterAggregates'])
+        if not isinstance(power, list):
+            return None
+        if location:
+            for p in power:
+                if p.get('location') == location.upper():
+                    return p.get('realPowerW')
+        else:
+            # Build a dictionary of all locations
+            power = {}
+            for p in power:
+                power[p.get('location')] = p.get('realPowerW')
+        return power
+
+
     def backup_time_remaining(self, force=False):
         """
         Get the time remaining in hours
         """
         status = self.get_status(force)
         nominalEnergyRemainingWh = lookup(status, ['control', 'systemStatus', 'nominalEnergyRemainingWh'])
-        power = lookup(status, ['control', 'meterAggregates', 0, 'realPowerW'])
-        if not nominalEnergyRemainingWh or not power:
+        load = self.current_power('LOAD', force)
+        if not nominalEnergyRemainingWh or not load:
             return None
-        time_remaining = nominalEnergyRemainingWh / power 
+        time_remaining = nominalEnergyRemainingWh / load
         return time_remaining
+
 
     def battery_level(self, force=False):
         """
@@ -795,4 +817,68 @@ def calculate_dc_power(V, I):
         }
         return vitals
 
+
+    def get_blocks(self, force=False):
+        """
+        Get the list of battery blocks from the Powerwall Gateway
+        """
+        status = self.get_status(force)
+        config = self.get_config(force)
+
+        if not isinstance(status, dict) or not isinstance(config, dict):
+            return None
+        block = {} 
+        i = 0
+        # Loop through each THC device serial number
+        for p in lookup(status, ['esCan', 'bus', 'THC']) or {}:
+            if not p['packageSerialNumber']:
+                continue
+            packagePartNumber = p.get('packagePartNumber', str(i))
+            packageSerialNumber = p.get('packageSerialNumber', str(i))
+            # THC block
+            name = f"{packagePartNumber}--{packageSerialNumber}"
+            block[name] = {
+                "Type": "",
+                "PackagePartNumber": packagePartNumber,
+                "PackageSerialNumber": packageSerialNumber,
+                "disabled_reasons": [],
+                "pinv_state": None,
+                "pinv_grid_state": None,
+                "nominal_energy_remaining": None,
+                "nominal_full_pack_energy": None,
+                "p_out": None,
+                "q_out": None,
+                "v_out": None,
+                "f_out": None,
+                "i_out": None,
+                "energy_charged": None,
+                "energy_discharged": None,
+                "off_grid": None,
+                "vf_mode": None,
+                "wobble_detected": None,
+                "charge_power_clamped": None,
+                "backup_ready": None,
+                "OpSeqState": None,
+                "version": None
+            }
+            # POD block
+            pod = lookup(status, ['esCan', 'bus', 'POD'])[i]
+            energy_remaining = lookup(pod, ['POD_EnergyStatus', 'POD_nom_energy_remaining'])
+            full_pack_energy = lookup(pod, ['POD_EnergyStatus', 'POD_nom_full_pack_energy'])
+            block[name].update({
+                "nominal_energy_remaining": energy_remaining,
+                "nominal_full_pack_energy": full_pack_energy,
+            })
+            # INV block
+            pinv = lookup(status, ['esCan', 'bus', 'PINV'])[i]
+            block[name].update({
+                "f_out": lookup(pinv, ['PINV_Status', 'PINV_Fout']),
+                "pinv_state": lookup(p, ['PINV_Status', 'PINV_State']),
+                "pinv_grid_state": lookup(p, ['PINV_Status', 'PINV_GridState']),
+                "p_out": lookup(pinv, ['PINV_Status', 'PINV_Pout']),
+                "v_out": lookup(pinv, ['PINV_Status', 'PINV_Vout']),
+            })
+            i = i + 1
+        return block
+
     # End of TEDAPI Class

pypowerwall/tedapi/__main__.py

@@ -7,9 +7,6 @@
 """
 
 def run_tedapi_test(auto=False, debug=False):
-    # Print header
-    print("pyPowerwall - Powerwall Gateway TEDAPI Reader")
-
     # Imports
     from pypowerwall.tedapi import TEDAPI, GW_IP
     from pypowerwall import __version__
@@ -19,6 +16,9 @@ def run_tedapi_test(auto=False, debug=False):
     import requests
     import logging
 
+    # Print header
+    print(f"pyPowerwall - Powerwall Gateway TEDAPI Reader [v{__version__}]")
+
     # Setup Logging
     log = logging.getLogger(__name__)
 

pypowerwall/tedapi/decorators.py

@@ -0,0 +1,19 @@
+import functools
+import logging
+
+log = logging.getLogger('pypowerwall.tedapi.pypowerwall_tedapi')
+WARNED_ONCE = {}
+
+
+def not_implemented_mock_data(func):
+    @functools.wraps(func)
+    def wrapper(*args, **kwargs):
+        if not WARNED_ONCE.get(func.__name__):
+            log.warning(f"This API [{func.__name__}] is using mock data in tedapi mode. This message will be "
+                        "printed only once at the warning level.")
+            WARNED_ONCE[func.__name__] = 1
+        else:
+            log.debug(f"This API [{func.__name__}] is using mock data in tedapi mode.")
+        return func(*args, **kwargs)
+
+    return wrapper

pypowerwall/tedapi/exceptions.py

@@ -0,0 +1,14 @@
+class PyPowerwallTEDAPINoTeslaAuthFile(Exception):
+    pass
+
+
+class PyPowerwallTEDAPITeslaNotConnected(Exception):
+    pass
+
+
+class PyPowerwallTEDAPINotImplemented(Exception):
+    pass
+
+
+class PyPowerwallTEDAPIInvalidPayload(Exception):
+    pass