Support PowerStream Messages with Multiple Payloads (#95)

* Supper PowerStream Messages with Multiple Payloads

When the mobile app is launched a request is sent out to request all the data. This change allows Home Assistant to read extra messages that ProtoBuf doesn't read naturally.

* Add Beta flag to PowerStream in ReadMe

To help keep requests and miss understandings to a minimum I have added notes to the PowerStream section of the readme with the limitations of the current implementation.

README.md

@@ -482,9 +482,21 @@ Once installed, use Add Integration -> Ecoflow Cloud.
 
 </p></details>
 
-<details><summary> POWERSTREAM <i>(sensors: 25, switches: 0, sliders: 0, selects: 0)</i> </summary>
+<details><summary> POWERSTREAM <sup>beta</sup> <i>(sensors: 55, switches: 0, sliders: 0, selects: 0)</i> </summary>
 <p>
 
+Due to limitations with the way that Ecoflow have implemented the communication
+between the device and app only a couple of fields at a time get updated. It may
+take several hours for all fields to receive data. However, launching the app
+and viewing the PowerStream should force all the fields to be updated at once.
+
+PowerStream support is still in development.
+
+Wishlist
+
+- Ability to change settings
+- Add Sensors to allow data to be displyed on the Energy Dashboard
+
 *Sensors*
 - Solar 1 Watts
 - Solar 1 Input Potential

custom_components/ecoflow_cloud/devices/powerstream.py

@@ -35,7 +35,7 @@ def sensors(self, client: EcoflowMQTTClient) -> list[BaseSensorEntity]:
             MiscSensorEntity(client, "pv2_warning_code", "Solar 2 Warning Code", False),
             MiscSensorEntity(client, "pv2_status", "Solar 2 Status", False),
 
-            MiscSensorEntity(client, "bpType", "Battery Type", False),
+            MiscSensorEntity(client, "bp_type", "Battery Type", False),
             LevelSensorEntity(client, "bat_soc", "Battery Charge"),
             DeciwattsSensorEntity(client, "bat_input_watts", "Battery Input Watts"),
             DecivoltSensorEntity(client, "bat_input_volt", "Battery Input Potential"),

custom_components/ecoflow_cloud/mqtt/ecoflow_mqtt.py

@@ -259,33 +259,43 @@ def on_json_message(self, client, userdata, message):
 
     def on_bytes_message(self, client, userdata, message):
         try:
-            packet = ecopacket.SendHeaderMsg()
-            packet.ParseFromString(message.payload)
+            payload = message.payload
 
-            _LOGGER.debug("cmd id %u payload \"%s\"", packet.msg.cmd_id, message.payload.hex())
+            while True:
+                packet = ecopacket.SendHeaderMsg()
+                packet.ParseFromString(payload)
 
-            if packet.msg.cmd_id != 1:
-                _LOGGER.info("Unsupported EcoPacket cmd id %u", packet.msg.cmd_id)
-                return
+                _LOGGER.debug("cmd id %u payload \"%s\"", packet.msg.cmd_id, payload.hex())
 
-            if message.topic != self._data_topic:
-                _LOGGER.info("PowerStream not listening to %s MQTT topic", message.topic)
-                return
+                if packet.msg.cmd_id != 1:
+                    _LOGGER.info("Unsupported EcoPacket cmd id %u", packet.msg.cmd_id)
 
-            heartbeat = powerstream.InverterHeartbeat()
-            heartbeat.ParseFromString(packet.msg.pdata)
+                else:
+                    heartbeat = powerstream.InverterHeartbeat()
+                    heartbeat.ParseFromString(packet.msg.pdata)
 
-            raw = {"params": {}}
+                    raw = {"params": {}}
 
-            for descriptor in heartbeat.DESCRIPTOR.fields:
-                if not heartbeat.HasField(descriptor.name):
-                    continue
+                    for descriptor in heartbeat.DESCRIPTOR.fields:
+                        if not heartbeat.HasField(descriptor.name):
+                            continue
 
-                raw["params"][descriptor.name] = getattr(heartbeat, descriptor.name)
+                        raw["params"][descriptor.name] = getattr(heartbeat, descriptor.name)
 
-            raw["timestamp"] = utcnow()
+                    _LOGGER.info("Found %u fields", len(raw["params"]))
+
+                    raw["timestamp"] = utcnow()
+
+                    self.data.update_data(raw)
+
+                if packet.ByteSize() >= len(payload):
+                    break
+
+                _LOGGER.info("Found another frame in payload")
+
+                packetLength = len(payload) - packet.ByteSize()
+                payload = payload[:packetLength]
 
-            self.data.update_data(raw)
         except Exception as error:
             _LOGGER.error(error)
             _LOGGER.info(message.payload.hex())