Add PW3 Vitals

.gitignore

@@ -158,3 +158,7 @@ j
 config.json
 status.json
 tools/tedapi/firmware.raw
+tools/tedapi/components.json
+tools/tedapi/manypw3.json
+tools/tedapi/pw3.json
+tools/tedapi/test.py

RELEASE.md

@@ -1,5 +1,16 @@
 # RELEASE NOTES
 
+## v0.11.0 - Add PW3 Vitals
+
+* Add polling of Powerwall 3 Devices to pull in PW3 specific string data, capacity, voltages, frequencies, and alerts. 
+* This creates mock TEPOD, PVAC and PVS compatible payloads available in vitals().
+
+Proxy URLs updated for PW3:
+* http://localhost:8675/vitals 
+* http://localhost:8675/help (verify pw3 shows True) 
+* http://localhost:8675/tedapi/components
+* http://localhost:8675/tedapi/battery  
+
 ## v0.10.10 - Add Grid Control
 
 * Add a function and command line options to allow user to get and set grid charging and exporting modes (see https://github.com/jasonacox/pypowerwall/issues/108).

proxy/server.py

@@ -54,7 +54,7 @@
 import pypowerwall
 from pypowerwall import parse_version
 
-BUILD = "t63"
+BUILD = "t64"
 ALLOWLIST = [
     '/api/status', '/api/site_info/site_name', '/api/meters/site',
     '/api/meters/solar', '/api/sitemaster', '/api/powerwalls',
@@ -113,6 +113,7 @@
     'cloudmode': False,
     'fleetapi': False,
     'tedapi': False,
+    'pw3': False,
     'tedapi_mode': "off",
     'siteid': None,
     'counter': 0
@@ -207,6 +208,7 @@ def get_value(a, key):
     if pw.tedapi:
         proxystats['tedapi'] = True
         proxystats['tedapi_mode'] = pw.tedapi_mode
+        proxystats['pw3'] = pw.tedapi.pw3
         log.info(f"TEDAPI Mode Enabled for Device Vitals ({pw.tedapi_mode})")
 
 pw_control = None
@@ -551,11 +553,15 @@ def do_GET(self):
         elif self.path.startswith('/tedapi'):
             # TEDAPI Specific Calls
             if pw.tedapi:
-                message = '{"error": "Use /tedapi/config, /tedapi/status"}'
+                message = '{"error": "Use /tedapi/config, /tedapi/status, /tedapi/components, /tedapi/battery"}'
                 if self.path == '/tedapi/config':
                     message = json.dumps(pw.tedapi.get_config())
                 if self.path == '/tedapi/status':
                     message = json.dumps(pw.tedapi.get_status())
+                if self.path == '/tedapi/components':
+                    message = json.dumps(pw.tedapi.get_components())
+                if self.path == '/tedapi/battery':
+                    message = json.dumps(pw.tedapi.get_battery_blocks())
             else:
                 message = '{"error": "TEDAPI not enabled"}'
         elif self.path.startswith('/cloud'):

pypowerwall/__init__.py

@@ -88,7 +88,7 @@
 from typing import Union, Optional
 import time
 
-version_tuple = (0, 10, 10)
+version_tuple = (0, 11, 0)
 version = __version__ = '%d.%d.%d' % version_tuple
 __author__ = 'jasonacox'
 

pypowerwall/tedapi/__init__.py

@@ -27,8 +27,10 @@
     battery_level() - Get the battery level as a percentage
     vitals() - Use tedapi data to create a vitals dictionary
     get_firmware_version() - Get the Powerwall Firmware Version
+    get_battery_blocks() - Get list of Powerwall Battery Blocks
     get_components() - Get the Powerwall 3 Device Information
-    get_battery_block() - Get the Powerwall 3 Battery Blocks
+    get_battery_block(din) - Get the Powerwall 3 Battery Block Information
+    get_pw3_vitals() - Get the Powerwall 3 Vitals Information
 
  Note:
     This module requires access to the Powerwall Gateway. You can add a route to
@@ -534,6 +536,184 @@ def get_components(self, force=False):
         return components
 
 
+    def get_pw3_vitals(self, force=False):
+        """
+        Get Powerwall 3 Battery Vitals Data
+
+        Returns:
+        {
+            "PVAC--{part}--{sn}" {
+                "PVAC_PvState_A": "PV_Active",
+                "PVAC_PVCurrent_A": 0.0,
+                ...
+                "PVAC_PVMeasuredVoltage_A": 0.0,
+                ...
+                "PVAC_PVMeasuredPower_A": 0.0,
+                ...
+                "PVAC_Fout": 60.0,
+                "PVAC_Pout": 0.0,
+                "PVAC_State": X,
+                "PVAC_VL1Ground": lookup(p, ['PVAC_Logging', 'PVAC_VL1Ground']),
+                "PVAC_VL2Ground": lookup(p, ['PVAC_Logging', 'PVAC_VL2Ground']),
+                "PVAC_Vout": lookup(p, ['PVAC_Status', 'PVAC_Vout']),
+                "manufacturer": "TESLA",
+                "partNumber": packagePartNumber,
+                "serialNumber": packageSerialNumber,
+            }.
+            "PVS--{part}--{sn}" {
+                "PVS_StringA_Connected": true,
+                ...
+            },
+            "TEPOD--{part}--{sn}" {
+                "alerts": [],
+                "POD_nom_energy_remaining": 0.0,
+                "POD_nom_full_pack_energy": 0.0,
+                "POD_nom_energy_to_be_charged": 0.0,
+            }
+        }   
+        """
+        # Check Connection
+        if not self.din:
+            if not self.connect():
+                log.error("Not Connected - Unable to get configuration")
+                return None
+        # Check Cache
+        if not force and "pw3_vitals" in self.pwcachetime:
+            if time.time() - self.pwcachetime["pw3_vitals"] < self.pwconfigexpire:
+                log.debug("Using Cached Components")
+                return self.pwcache["pw3_vitals"]
+        if not force and self.pwcooldown > time.perf_counter():
+            # Rate limited - return None
+            log.debug('Rate limit cooldown period - Pausing API calls')
+            return None
+        components = self.get_components(force)
+        din = self.din
+        if not components:
+            log.error("Unable to get Powerwall 3 Components")
+            return None
+
+        response = {}
+        config = self.get_config(force)
+        battery_blocks = config['battery_blocks']
+
+        # Loop through all the battery blocks (Powerwalls)
+        for battery in battery_blocks:
+            pw_din = battery['vin'] # 1707000-11-J--TG12xxxxxx3A8Z
+            pw_part, pw_serial = pw_din.split('--')
+            battery_type = battery['type']
+            if "Powerwall3" not in battery_type:
+                continue
+            # Fetch Device ComponentsQuery from each Powerwall
+            pb = tedapi_pb2.Message()
+            pb.message.deliveryChannel = 1
+            pb.message.sender.local = 1
+            pb.message.sender.din = din  # DIN of Primary Powerwall 3 / System
+            pb.message.recipient.din = pw_din  # DIN of Powerwall of Interest
+            pb.message.payload.send.num = 2
+            pb.message.payload.send.payload.value = 1
+            pb.message.payload.send.payload.text = " query ComponentsQuery (\n  $pchComponentsFilter: ComponentFilter,\n  $pchSignalNames: [String!],\n  $pwsComponentsFilter: ComponentFilter,\n  $pwsSignalNames: [String!],\n  $bmsComponentsFilter: ComponentFilter,\n  $bmsSignalNames: [String!],\n  $hvpComponentsFilter: ComponentFilter,\n  $hvpSignalNames: [String!],\n  $baggrComponentsFilter: ComponentFilter,\n  $baggrSignalNames: [String!],\n  ) {\n  # TODO STST-57686: Introduce GraphQL fragments to shorten\n  pw3Can {\n    firmwareUpdate {\n      isUpdating\n      progress {\n         updating\n         numSteps\n         currentStep\n         currentStepProgress\n         progress\n      }\n    }\n  }\n  components {\n    pws: components(filter: $pwsComponentsFilter) {\n      signals(names: $pwsSignalNames) {\n        name\n        value\n        textValue\n        boolValue\n        timestamp\n      }\n      activeAlerts {\n        name\n      }\n    }\n    pch: components(filter: $pchComponentsFilter) {\n      signals(names: $pchSignalNames) {\n        name\n        value\n        textValue\n        boolValue\n        timestamp\n      }\n      activeAlerts {\n        name\n      }\n    }\n    bms: components(filter: $bmsComponentsFilter) {\n      signals(names: $bmsSignalNames) {\n        name\n        value\n        textValue\n        boolValue\n        timestamp\n      }\n      activeAlerts {\n        name\n      }\n    }\n    hvp: components(filter: $hvpComponentsFilter) {\n      partNumber\n      serialNumber\n      signals(names: $hvpSignalNames) {\n        name\n        value\n        textValue\n        boolValue\n        timestamp\n      }\n      activeAlerts {\n        name\n      }\n    }\n    baggr: components(filter: $baggrComponentsFilter) {\n      signals(names: $baggrSignalNames) {\n        name\n        value\n        textValue\n        boolValue\n        timestamp\n      }\n      activeAlerts {\n        name\n      }\n    }\n  }\n}\n"
+            pb.message.payload.send.code = b'0\201\210\002B\000\270q\354>\243m\325p\371S\253\231\346~:\032\216~\242\263\207\017L\273O\203u\241\270\333w\233\354\276\246h\262\243\255\261\007\202D\277\353x\023O\022\303\216\264\010-\'i6\360>B\237\236\304\244m\002B\001\023Pk\033)\277\236\342R\264\247g\260u\036\023\3662\354\242\353\035\221\234\027\245\321J\342\345\037q\262O\3446-\353\315m1\237zai0\341\207C4\307\300Z\177@h\335\327\0239\252f\n\206W'
+            pb.message.payload.send.b.value = "{\"pwsComponentsFilter\":{\"types\":[\"PW3SAF\"]},\"pwsSignalNames\":[\"PWS_SelfTest\",\"PWS_PeImpTestState\",\"PWS_PvIsoTestState\",\"PWS_RelaySelfTest_State\",\"PWS_MciTestState\",\"PWS_appGitHash\",\"PWS_ProdSwitch_State\"],\"pchComponentsFilter\":{\"types\":[\"PCH\"]},\"pchSignalNames\":[\"PCH_State\",\"PCH_PvState_A\",\"PCH_PvState_B\",\"PCH_PvState_C\",\"PCH_PvState_D\",\"PCH_PvState_E\",\"PCH_PvState_F\",\"PCH_AcFrequency\",\"PCH_AcVoltageAB\",\"PCH_AcVoltageAN\",\"PCH_AcVoltageBN\",\"PCH_packagePartNumber_1_7\",\"PCH_packagePartNumber_8_14\",\"PCH_packagePartNumber_15_20\",\"PCH_packageSerialNumber_1_7\",\"PCH_packageSerialNumber_8_14\",\"PCH_PvVoltageA\",\"PCH_PvVoltageB\",\"PCH_PvVoltageC\",\"PCH_PvVoltageD\",\"PCH_PvVoltageE\",\"PCH_PvVoltageF\",\"PCH_PvCurrentA\",\"PCH_PvCurrentB\",\"PCH_PvCurrentC\",\"PCH_PvCurrentD\",\"PCH_PvCurrentE\",\"PCH_PvCurrentF\",\"PCH_BatteryPower\",\"PCH_AcRealPowerAB\",\"PCH_SlowPvPowerSum\",\"PCH_AcMode\",\"PCH_AcFrequency\",\"PCH_DcdcState_A\",\"PCH_DcdcState_B\",\"PCH_appGitHash\"],\"bmsComponentsFilter\":{\"types\":[\"PW3BMS\"]},\"bmsSignalNames\":[\"BMS_nominalEnergyRemaining\",\"BMS_nominalFullPackEnergy\",\"BMS_appGitHash\"],\"hvpComponentsFilter\":{\"types\":[\"PW3HVP\"]},\"hvpSignalNames\":[\"HVP_State\",\"HVP_appGitHash\"],\"baggrComponentsFilter\":{\"types\":[\"BAGGR\"]},\"baggrSignalNames\":[\"BAGGR_State\",\"BAGGR_OperationRequest\",\"BAGGR_NumBatteriesConnected\",\"BAGGR_NumBatteriesPresent\",\"BAGGR_NumBatteriesExpected\",\"BAGGR_LOG_BattConnectionStatus0\",\"BAGGR_LOG_BattConnectionStatus1\",\"BAGGR_LOG_BattConnectionStatus2\",\"BAGGR_LOG_BattConnectionStatus3\"]}"
+            pb.tail.value = 2
+            url = f'https://{GW_IP}/tedapi/device/{pw_din}/v1'
+            r = requests.post(url, auth=('Tesla_Energy_Device', self.gw_pwd), verify=False,
+                            headers={'Content-type': 'application/octet-string'},
+                            data=pb.SerializeToString(), timeout=self.timeout)
+            if r.status_code == 200:
+                # Decode response
+                tedapi = tedapi_pb2.Message()
+                tedapi.ParseFromString(r.content)
+                payload = tedapi.message.payload.recv.text
+                if payload:
+                    data = json.loads(payload)
+                    # TEDPOD
+                    alerts = []
+                    components = data['components']
+                    for component in components:
+                        for alert in components[component][0]['activeAlerts']:
+                            if alert['name'] not in alerts:
+                                alerts.append(alert['name'])
+                    bms_component = data['components']['bms'][0] # TODO: Process all BMS components
+                    signals = bms_component['signals']
+                    nom_energy_remaining = 0
+                    nom_full_pack_energy = 0
+                    for signal in signals:
+                        if "BMS_nominalEnergyRemaining" == signal['name']:
+                            nom_energy_remaining = int(signal['value'] * 1000) # Convert to Wh
+                        elif "BMS_nominalFullPackEnergy" == signal['name']:
+                            nom_full_pack_energy = int(signal['value'] * 1000) # Convert to Wh
+                    response[f"TEPOD--{pw_din}"] = {
+                        "alerts": alerts,
+                        "POD_nom_energy_remaining": nom_energy_remaining,
+                        "POD_nom_energy_to_be_charged": nom_full_pack_energy - nom_energy_remaining,
+                        "POD_nom_full_pack_energy": nom_full_pack_energy,
+                    }
+                    # PVAC, PVS and TEPINV
+                    response[f"PVAC--{pw_din}"] = {}
+                    response[f"PVS--{pw_din}"] = {}
+                    response[f"TEPINV--{pw_din}"] = {}
+                    pch_components = data['components']['pch']
+                    # pch_components contain:
+                    #   PCH_PvState_A through F - textValue in [Pv_Active, Pv_Active_Parallel, Pv_Standby]
+                    #   PCH_PvVoltageA through F - value
+                    #   PCH_PvCurrentA through F - value
+                    # Loop through and find all the strings - PW3 has 6 strings A-F
+                    for n in ["A", "B", "C", "D", "E", "F"]:
+                        pv_state = "Unknown"
+                        pv_voltage = 0
+                        pv_current = 0
+                        for component in pch_components: # TODO: Probably better way to do this
+                            signals = component['signals']
+                            for signal in signals:
+                                if f'PCH_PvState_{n}' == signal['name']:
+                                    pv_state = signal['textValue']
+                                elif f'PCH_PvVoltage{n}' == signal['name']:
+                                    pv_voltage = signal['value'] if signal['value'] > 0 else 0
+                                elif f'PCH_PvCurrent{n}' == signal['name']:
+                                    pv_current = signal['value'] if signal['value'] > 0 else 0
+                                elif 'PCH_AcFrequency' == signal['name']:
+                                    response[f"PVAC--{pw_din}"]["PVAC_Fout"] = signal['value']
+                                    response[f"TEPINV--{pw_din}"]["PINV_Fout"] = signal['value']
+                                elif 'PCH_AcVoltageAN' == signal['name']:
+                                    response[f"PVAC--{pw_din}"]["PVAC_VL1Ground"] = signal['value']
+                                    response[f"TEPINV--{pw_din}"]["PINV_VSplit1"] = signal['value']
+                                elif 'PCH_AcVoltageBN' == signal['name']:
+                                    response[f"PVAC--{pw_din}"]["PVAC_VL2Ground"] = signal['value']
+                                    response[f"TEPINV--{pw_din}"]["PINV_VSplit2"] = signal['value']
+                                elif 'PCH_AcVoltageAB' == signal['name']:
+                                    response[f"PVAC--{pw_din}"]["PVAC_Vout"] = signal['value']
+                                    response[f"TEPINV--{pw_din}"]["PINV_Vout"] = signal['value']
+                                elif 'PCH_AcRealPowerAB' == signal['name']:
+                                    response[f"PVAC--{pw_din}"]["PVAC_Pout"] = signal['value']
+                                    response[f"TEPINV--{pw_din}"]["PINV_Pout"] = (signal['value'] or 0) / 1000
+                                elif 'PCH_AcMode' == signal['name']:
+                                    response[f"PVAC--{pw_din}"]["PVAC_State"] = signal['textValue']
+                                    response[f"TEPINV--{pw_din}"]["PINV_State"] = signal['textValue']
+                        pv_power = pv_voltage * pv_current # Calculate power
+                        response[f"PVAC--{pw_din}"][f"PVAC_PvState_{n}"] = pv_state
+                        response[f"PVAC--{pw_din}"][f"PVAC_PVMeasuredVoltage_{n}"] = pv_voltage
+                        response[f"PVAC--{pw_din}"][f"PVAC_PVCurrent_{n}"] = pv_current
+                        response[f"PVAC--{pw_din}"][f"PVAC_PVMeasuredPower_{n}"] = pv_power
+                        response[f"PVAC--{pw_din}"]["manufacturer"] = "TESLA"
+                        response[f"PVAC--{pw_din}"]["partNumber"] = pw_part
+                        response[f"PVAC--{pw_din}"]["serialNumber"] = pw_serial
+                        response[f"PVS--{pw_din}"][f"PVS_String{n}_Connected"] = ("Pv_Active" in pv_state)
+                else:
+                    log.debug(f"No payload for {pw_din}")
+            else:
+                log.debug(f"Error fetching components: {r.status_code}")
+        return response
+
+
+    def get_battery_blocks(self, force=False):
+        """
+        Return Powerwall Battery Blocks
+        """
+        config = self.get_config(force)
+        battery_blocks = config.get('battery_blocks') or []
+        return battery_blocks
+
+
     def get_battery_block(self, din=None, force=False):
         """
         Get the Powerwall 3 Battery Block Information
@@ -546,8 +726,6 @@ def get_battery_block(self, din=None, force=False):
         """
         data = None
         # Make sure we have a DIN
-        if not din:
-            din = self.get_din()
         if not din:
             log.error("No DIN specified - Unable to get battery block")
             return None
@@ -1108,6 +1286,11 @@ def calculate_dc_power(V, I):
             **tesync,
             **tethc,
         }
+        # Merge in the Powerwall 3 data if available
+        if self.pw3:
+            pw3_data = self.get_pw3_vitals(force) or {}
+            vitals.update(pw3_data)
+
         return vitals