v0.10.0-pre: QuotasStatusSensor for DELTA_2, RIVER_2_MAX

custom_components/ecoflow_cloud/__init__.py

@@ -19,6 +19,9 @@
     Platform.SWITCH,
 }
 
+ATTR_STATUS_SN = "sn"
+ATTR_STATUS_UPDATES = "updates"
+
 
 async def async_migrate_entry(hass, config_entry: ConfigEntry):
     """Migrate old entry."""

custom_components/ecoflow_cloud/devices/__init__.py

@@ -1,5 +1,10 @@
 from abc import ABC, abstractmethod
 
+from homeassistant.components.number import NumberEntity
+from homeassistant.components.select import SelectEntity
+from homeassistant.components.sensor import SensorEntity
+from homeassistant.components.switch import SwitchEntity
+
 from ..mqtt.ecoflow_mqtt import EcoflowMQTTClient
 from ..entities import BaseNumberEntity, BaseSelectEntity
 from ..entities import BaseSensorEntity
@@ -12,32 +17,32 @@ def charging_power_step(self) -> int:
         return 100
 
     @abstractmethod
-    def sensors(self, client: EcoflowMQTTClient) -> list[BaseSensorEntity]:
+    def sensors(self, client: EcoflowMQTTClient) -> list[SensorEntity]:
         pass
 
     @abstractmethod
-    def numbers(self, client: EcoflowMQTTClient) -> list[BaseNumberEntity]:
+    def numbers(self, client: EcoflowMQTTClient) -> list[NumberEntity]:
         pass
 
     @abstractmethod
-    def switches(self, client: EcoflowMQTTClient) -> list[BaseSwitchEntity]:
+    def switches(self, client: EcoflowMQTTClient) -> list[SwitchEntity]:
         pass
 
     @abstractmethod
-    def selects(self, client: EcoflowMQTTClient) -> list[BaseSelectEntity]:
+    def selects(self, client: EcoflowMQTTClient) -> list[SelectEntity]:
         pass
 
 
 class DiagnosticDevice(BaseDevice):
 
-    def sensors(self, client: EcoflowMQTTClient) -> list[BaseSensorEntity]:
+    def sensors(self, client: EcoflowMQTTClient) -> list[SensorEntity]:
         return []
 
-    def numbers(self, client: EcoflowMQTTClient) -> list[BaseNumberEntity]:
+    def numbers(self, client: EcoflowMQTTClient) -> list[NumberEntity]:
         return []
 
-    def switches(self, client: EcoflowMQTTClient) -> list[BaseSwitchEntity]:
+    def switches(self, client: EcoflowMQTTClient) -> list[SwitchEntity]:
         return []
 
-    def selects(self, client: EcoflowMQTTClient) -> list[BaseSelectEntity]:
+    def selects(self, client: EcoflowMQTTClient) -> list[SelectEntity]:
         return []

custom_components/ecoflow_cloud/devices/delta2.py

@@ -5,7 +5,7 @@
     MaxGenStopLevelEntity, MinGenStartLevelEntity
 from ..select import DictSelectEntity, TimeoutDictSelectEntity
 from ..sensor import LevelSensorEntity, RemainSensorEntity, TempSensorEntity, CyclesSensorEntity, \
-    InWattsSensorEntity, OutWattsSensorEntity, VoltSensorEntity
+    InWattsSensorEntity, OutWattsSensorEntity, VoltSensorEntity, QuotasStatusSensorEntity
 from ..switch import BeeperEntity, EnabledEntity
 
 
@@ -61,7 +61,8 @@ def sensors(self, client: EcoflowMQTTClient) -> list[BaseSensorEntity]:
 
             CyclesSensorEntity(client, "bms_slave.cycles", const.SLAVE_CYCLES, False, True),
             InWattsSensorEntity(client, "bms_slave.inputWatts", const.SLAVE_IN_POWER, False, True),
-            OutWattsSensorEntity(client, "bms_slave.outputWatts", const.SLAVE_OUT_POWER, False, True)
+            OutWattsSensorEntity(client, "bms_slave.outputWatts", const.SLAVE_OUT_POWER, False, True),
+            QuotasStatusSensorEntity(client),
 
         ]
 

custom_components/ecoflow_cloud/devices/river2_max.py

@@ -5,7 +5,7 @@
 from ..select import DictSelectEntity, TimeoutDictSelectEntity
 from ..sensor import LevelSensorEntity, RemainSensorEntity, TempSensorEntity, \
     CyclesSensorEntity, InWattsSensorEntity, OutWattsSensorEntity, VoltSensorEntity, InAmpSensorEntity, \
-    InVoltSensorEntity
+    InVoltSensorEntity, QuotasStatusSensorEntity
 from ..switch import EnabledEntity
 
 
@@ -48,6 +48,8 @@ def sensors(self, client: EcoflowMQTTClient) -> list[BaseSensorEntity]:
             VoltSensorEntity(client, "bms_bmsStatus.vol", const.BATTERY_VOLT, False),
             VoltSensorEntity(client, "bms_bmsStatus.minCellVol", const.MIN_CELL_VOLT, False),
             VoltSensorEntity(client, "bms_bmsStatus.maxCellVol", const.MAX_CELL_VOLT, False),
+
+            QuotasStatusSensorEntity(client),
             # FanSensorEntity(client, "bms_emsStatus.fanLevel", "Fan Level"),
 
         ]

custom_components/ecoflow_cloud/diagnostics.py

@@ -19,9 +19,11 @@ def _to_serializable(x):
 async def async_get_config_entry_diagnostics(hass: HomeAssistant, entry: ConfigEntry):
     client: EcoflowMQTTClient = hass.data[DOMAIN][entry.entry_id]
     values = {
-        'data': dict(sorted(client.data.params.items())),
-        'set_commands': {k: v.diagnostic_dict() for k, v in client.data.set_commands.items()},
-        'get_commands': {k: v.diagnostic_dict() for k, v in client.data.get_commands.items()},
+        'params':     dict(sorted(client.data.params.items())),
+        'set':       [dict(sorted(k.items())) for k in client.data.set],
+        'set_reply': [dict(sorted(k.items())) for k in client.data.set_reply],
+        'get':       [dict(sorted(k.items())) for k in client.data.get],
+        'get_reply': [dict(sorted(k.items())) for k in client.data.get_reply],
         'raw_data': client.data.raw_data,
     }
     return values

custom_components/ecoflow_cloud/entities/__init__.py

@@ -9,21 +9,31 @@
 from ..mqtt.ecoflow_mqtt import EcoflowMQTTClient
 
 
-class EcoFlowBaseEntity(Entity):
+class EcoFlowAbstractEntity(Entity):
     _attr_has_entity_name = True
     _attr_should_poll = False
 
+    def __init__(self, client: EcoflowMQTTClient, title: str, key: str) -> object:
+        self._client = client
+        self._attr_name = title
+        self._attr_device_info = client.device_info_main
+        self._attr_unique_id = 'ecoflow-' + client.device_sn + '-' + key.replace('.', '-').replace('_', '-')
+
+    def send_get_message(self, command: dict):
+        self._client.send_get_message(command)
+
+    def send_set_message(self, target_dict: dict[str, Any] | None, command: dict):
+        self._client.send_set_message(target_dict, command)
+
+
+class EcoFlowDictEntity(EcoFlowAbstractEntity):
+
     def __init__(self, client: EcoflowMQTTClient, mqtt_key: str, title: str, enabled: bool = True,
                  auto_enable: bool = False) -> object:
-        # self._attr_available = False
-        self._client = client
+        super().__init__(client, title, mqtt_key)
         self._mqtt_key = mqtt_key
         self._auto_enable = auto_enable
-
-        self._attr_name = title
         self._attr_entity_registry_enabled_default = enabled
-        self._attr_device_info = client.device_info_main
-        self._attr_unique_id = 'ecoflow-' + client.device_sn + '-' + mqtt_key.replace('.', '-').replace('_', '-')
 
     @property
     def mqtt_key(self):
@@ -33,6 +43,9 @@ def mqtt_key(self):
     def auto_enable(self):
         return self._auto_enable
 
+    def send_set_message(self, target_value: Any, command: dict):
+        super().send_set_message({self._mqtt_key: target_value}, command)
+
     @property
     def enabled_default(self):
         return self._attr_entity_registry_enabled_default
@@ -54,11 +67,8 @@ def __updated(self, data: dict[str, Any]):
     def _update_value(self, val: Any) -> bool:
         return False
 
-    def send_message(self, target_value: Any, command: dict):
-        self._client.send_message({self._mqtt_key: target_value}, command)
-
 
-class EcoFlowBaseCommandEntity(EcoFlowBaseEntity):
+class EcoFlowBaseCommandEntity(EcoFlowDictEntity):
     def __init__(self, client: EcoflowMQTTClient, mqtt_key: str, title: str,
                  command: Callable[[int], dict[str, any]] | None, enabled: bool = True, auto_enable: bool = False):
         super().__init__(client, mqtt_key, title, enabled, auto_enable)
@@ -89,7 +99,7 @@ def _update_value(self, val: Any) -> bool:
             return False
 
 
-class BaseSensorEntity(SensorEntity, EcoFlowBaseEntity):
+class BaseSensorEntity(SensorEntity, EcoFlowDictEntity):
 
     def _update_value(self, val: Any) -> bool:
         if self._attr_native_value != val:

custom_components/ecoflow_cloud/manifest.json

@@ -12,5 +12,5 @@
     "paho-mqtt==1.6.1",
     "reactivex==4.0.4"
   ],
-  "version": "0.9.1"
+  "version": "0.10.0-pre"
 }

custom_components/ecoflow_cloud/mqtt/ecoflow_mqtt.py

@@ -16,7 +16,7 @@
 from homeassistant.util import utcnow
 from reactivex import Subject, Observable
 
-from .utils import LimitedSizeOrderedDict
+from .utils import BoundFifoList
 from ..config.const import CONF_DEVICE_TYPE, CONF_DEVICE_ID, OPTS_REFRESH_PERIOD_SEC
 
 _LOGGER = logging.getLogger(__name__)
@@ -95,89 +95,68 @@ def get_json_response(self, request):
         return response
 
 
-class EcoflowCommandInfo():
-
-    def __init__(self, target_state: dict[str, Any] | None, command: dict[str, Any]) -> None:
-        self.target_state = target_state
-        self.command = command
-        self.reply: dict[str, Any] = {}
-        self.time = utcnow().timestamp()
-
-    def diagnostic_dict(self) -> dict[str, Any]:
-        return {
-            "target_state": self.target_state,
-            "command": self.command,
-            "reply": self.reply
-        }
-
-
 class EcoflowDataHolder:
     def __init__(self, update_period_sec: int, collect_raw: bool = False):
         self.__update_period_sec = update_period_sec
         self.__collect_raw = collect_raw
-        self.set_commands = LimitedSizeOrderedDict[int, EcoflowCommandInfo]()
-        self.get_commands = LimitedSizeOrderedDict[int, EcoflowCommandInfo]()
-        self.raw_data = list[dict[str, Any]]()
+        self.set = BoundFifoList[dict[str, Any]]()
+        self.set_reply = BoundFifoList[dict[str, Any]]()
+        self.get = BoundFifoList[dict[str, Any]]()
+        self.get_reply = BoundFifoList[dict[str, Any]]()
         self.params = dict[str, Any]()
-        self.__broadcast_time: datetime = utcnow()
-        self.__observable = Subject[dict[str, Any]]()
+
+        self.raw_data = BoundFifoList[dict[str, Any]]()
+
+        self.__params_broadcast_time: datetime = utcnow()
+        self.__params_observable = Subject[dict[str, Any]]()
+
+        self.__set_reply_observable = Subject[list[dict[str, Any]]]()
+        self.__get_reply_observable = Subject[list[dict[str, Any]]]()
 
     def observable(self) -> Observable[dict[str, Any]]:
-        return self.__observable
-
-    def put_set_command(self, cmd_id: int, cmd: EcoflowCommandInfo) -> EcoflowCommandInfo:
-        self.set_commands.append(cmd_id, cmd)
-        return cmd
-
-    def add_set_command(self, target_state: dict[str, Any] | None, cmd: dict[str, Any]) -> EcoflowCommandInfo:
-        cmd_id = int(cmd["id"])
-        if cmd_id not in self.set_commands:
-            self.set_commands.append(cmd_id, EcoflowCommandInfo(target_state, cmd))
-            if target_state is not None:
-                self.__update_to_target_state(target_state)
-        return self.set_commands[cmd_id]
-
-    def add_set_command_reply(self, cmd: dict[str, Any]) -> EcoflowCommandInfo | None:
-        cmd_id = int(cmd["id"])
-        if cmd_id in self.set_commands:
-            cmd_info = self.set_commands[cmd_id]
-            cmd_info.reply = cmd
-            return cmd_info
-
-    def add_get_command(self, target_state: dict[str, Any] | None, cmd: dict[str, Any]):
-        cmd_id = int(cmd["id"])
-        if id not in self.get_commands:
-            self.get_commands.append(cmd_id, EcoflowCommandInfo(target_state, cmd))
-
-    def add_get_command_reply(self, cmd: dict[str, Any]):
-        cmd_id = int(cmd["id"])
-        if id in self.get_commands:
-            self.get_commands[cmd_id].reply = cmd
-
-    def __update_to_target_state(self, target_state: dict[str, Any]):
+        return self.__params_observable
+
+    def get_reply_observable(self) -> Observable[list[dict[str, Any]]]:
+        return self.__get_reply_observable
+
+    def set_reply_observable(self) -> Observable[list[dict[str, Any]]]:
+        return self.__set_reply_observable
+
+    def add_set_message(self, msg: dict[str, Any]):
+        self.set.append(msg)
+
+    def add_set_reply_message(self, msg: dict[str, Any]):
+        self.set_reply.append(msg)
+        self.__set_reply_observable.on_next(self.set_reply)
+
+    def add_get_message(self, msg: dict[str, Any]):
+        self.get.append(msg)
+
+    def add_get_reply_message(self, msg: dict[str, Any]):
+        self.get_reply.append(msg)
+        self.__get_reply_observable.on_next(self.get_reply)
+
+    def update_to_target_state(self, target_state: dict[str, Any]):
         self.params.update(target_state)
         self.__broadcast()
 
     def update_data(self, raw: dict[str, Any]):
         self.__add_raw_data(raw)
         self.params.update(raw['params'])
 
-        if (utcnow() - self.__broadcast_time).total_seconds() > self.__update_period_sec:
+        if (utcnow() - self.__params_broadcast_time).total_seconds() > self.__update_period_sec:
             self.__broadcast()
 
     def __broadcast(self):
-        self.__broadcast_time = utcnow()
-        self.__observable.on_next(self.params)
-
-    def __trim_list(self, src: list, size: int):
-        while len(src) >= size:
-            src.pop(0)
+        self.__params_broadcast_time = utcnow()
+        self.__params_observable.on_next(self.params)
 
     def __add_raw_data(self, raw: dict[str, Any]):
         if self.__collect_raw:
-            self.__trim_list(self.raw_data, 20)
             self.raw_data.append(raw)
 
+    def last_params_broadcast_time(self) -> datetime:
+        return self.__params_broadcast_time
 
 class EcoflowMQTTClient:
 
@@ -253,37 +232,34 @@ def on_message(self, client, userdata, message):
             if message.topic == self._data_topic:
                 self.data.update_data(raw)
             elif message.topic == self._set_topic:
-                self.data.add_set_command(None, raw)
+                self.data.add_set_message(raw)
             elif message.topic == self._set_reply_topic:
-                self.data.add_set_command_reply(raw)
+                self.data.add_set_reply_message(raw)
             elif message.topic == self._get_topic:
-                self.data.add_get_command(None, raw)
+                self.data.add_get_message(raw)
             elif message.topic == self._get_reply_topic:
-                self.data.add_get_command_reply(raw)
+                self.data.add_get_reply_message(raw)
         except UnicodeDecodeError as error:
             _LOGGER.error(f"UnicodeDecodeError: {error}. Ignoring message and waiting for the next one.")
 
     message_id = 999900000 + random.randint(10000, 99999)
 
-    def resend_message(self, msg: EcoflowCommandInfo):
-        self.message_id += 1
-        payload = {}
-        payload.update(msg.command)
-        payload["id"] = f"{self.message_id}"
-
-        self.data.put_set_command(self.message_id, msg)
-
-        info = self.client.publish(self._set_topic, json.dumps(payload), 1)
-        _LOGGER.debug("ReSending " + json.dumps(payload) + " :" + str(info) + "(" + str(info.is_published()) + ")")
-
-    def send_message(self, mqtt_state: dict[str, Any], command: dict):
+    def __prepare_payload(self, command: dict):
         self.message_id += 1
         payload = {"from": "HomeAssistant",
                    "id": f"{self.message_id}",
                    "version": "1.0"}
         payload.update(command)
-        self.data.add_set_command(mqtt_state, payload)
+        return payload
+
+    def send_get_message(self, command: dict):
+        payload = self.__prepare_payload(command)
+        info = self.client.publish(self._get_topic, json.dumps(payload), 1)
+        _LOGGER.debug("Sending " + json.dumps(payload) + " :" + str(info) + "(" + str(info.is_published()) + ")")
 
+    def send_set_message(self, mqtt_state: dict[str, Any], command: dict):
+        self.data.update_to_target_state(mqtt_state)
+        payload = self.__prepare_payload(command)
         info = self.client.publish(self._set_topic, json.dumps(payload), 1)
         _LOGGER.debug("Sending " + json.dumps(payload) + " :" + str(info) + "(" + str(info.is_published()) + ")")