improve msg100 emulation

custom_components/meross_lan/emulator/__init__.py

@@ -30,8 +30,11 @@ def build_emulator(tracefile, uuid, key) -> MerossEmulator:
  mixin_classes = []
 
  if mc.KEY_THERMOSTAT in descriptor.digest:
- from .mixins.thermostat import ThermostatMixin # pylint: disable=import-outside-toplevel
+ from .mixins.thermostat import ThermostatMixin
  mixin_classes.append(ThermostatMixin)
+ if mc.KEY_GARAGEDOOR in descriptor.digest:
+ from .mixins.garagedoor import GarageDoorMixin
+ mixin_classes.append(GarageDoorMixin)
 
  mixin_classes.append(MerossEmulator)
  # build a label to cache the set

custom_components/meross_lan/emulator/emulator.py

@@ -111,11 +111,18 @@ def _handler_default(self, method: str, namespace: str, payload: dict):
  If the state is not stored in all->digest we'll search our namespace(s) list for
  state carried through our GETACK messages in the trace
  """
- # quick lookup since state is saved in our namespaces
- if (method == mc.METHOD_GET) and (namespace in self.descriptor.namespaces):
- return mc.METHOD_GETACK, self.descriptor.namespaces[namespace]
+ try:
+ key, p_state = self._get_key_state(namespace)
+ except Exception as error:
+ # when the 'looking for state' euristic fails
+ # we might fallback to a static reply should it fit...
+ if (method == mc.METHOD_GET) and (namespace in self.descriptor.namespaces):
+ return mc.METHOD_GETACK, self.descriptor.namespaces[namespace]
+ raise error
+
+ if method == mc.METHOD_GET:
+ return mc.METHOD_GETACK, { key: p_state }
 
- key, p_state = self._get_key_state(namespace)
  if method != mc.METHOD_SET:
  # TODO.....
  raise Exception(f"{method} not supported in emulator")

custom_components/meross_lan/emulator/mixins/garagedoor.py

@@ -0,0 +1,52 @@
+""""""
+from __future__ import annotations
+import typing
+import asyncio
+from ..emulator import MerossEmulator # pylint: disable=relative-beyond-top-level
+from ...merossclient import const as mc, get_element_by_key # pylint: disable=relative-beyond-top-level
+
+
+class GarageDoorMixin(MerossEmulator if typing.TYPE_CHECKING else object):
+
+ def _SET_Appliance_GarageDoor_Config(self, header, payload):
+ p_config = self.descriptor.namespaces[mc.NS_APPLIANCE_GARAGEDOOR_CONFIG][mc.KEY_CONFIG]
+ p_request = payload[mc.KEY_CONFIG]
+ for _key, _value in p_request.items():
+ if _key in p_config:
+ p_config[_key] = _value
+ return mc.METHOD_SETACK, { }
+
+ def _GET_Appliance_GarageDoor_State(self, header, payload):
+ # return everything...at the moment we always query all
+ p_garageDoor: list = self.descriptor.digest[mc.KEY_GARAGEDOOR]
+ if len(p_garageDoor) == 1:
+ # un-pack the list since real traces show no list
+ # in this response payloads (we only have msg100 so far..)
+ return mc.METHOD_GETACK, { mc.KEY_STATE: p_garageDoor[0] }
+ else:
+ return mc.METHOD_GETACK, { mc.KEY_STATE: p_garageDoor }
+
+ def _SET_Appliance_GarageDoor_State(self, header, payload):
+ p_request = payload[mc.KEY_STATE]
+ request_channel = p_request[mc.KEY_CHANNEL]
+ request_open = p_request[mc.KEY_OPEN]
+ p_digest = self.descriptor.digest
+
+ p_state = get_element_by_key(
+ p_digest[mc.KEY_GARAGEDOOR],
+ mc.KEY_CHANNEL,
+ request_channel
+ )
+
+ p_response = dict(p_state)
+ if request_open != p_state[mc.KEY_OPEN]:
+
+ def _state_update_callback():
+ p_state[mc.KEY_OPEN] = request_open
+
+ loop = asyncio.get_event_loop()
+ loop.call_later(2 if request_open else 10, _state_update_callback)
+
+ p_response[mc.KEY_EXECUTE] = 1
+ return mc.METHOD_SETACK, { mc.KEY_STATE: p_response }
+

custom_components/meross_lan/emulator/mixins/thermostat.py

@@ -2,40 +2,39 @@
 from __future__ import annotations
 import typing
 from ..emulator import MerossEmulator # pylint: disable=relative-beyond-top-level
-from ...merossclient import const as mc # pylint: disable=relative-beyond-top-level
+from ...merossclient import const as mc, get_element_by_key # pylint: disable=relative-beyond-top-level
 
 
 class ThermostatMixin(MerossEmulator if typing.TYPE_CHECKING else object):
 
-
  def _SET_Appliance_Control_Thermostat_Mode(self, header, payload):
  p_digest = self.descriptor.digest
  p_digest_mode_list = p_digest[mc.KEY_THERMOSTAT][mc.KEY_MODE]
  p_digest_windowopened_list = {}
  p_mode_list = payload[mc.KEY_MODE]
  for p_mode in p_mode_list:
  channel = p_mode[mc.KEY_CHANNEL]
- for p_digest_mode in p_digest_mode_list:
- if p_digest_mode[mc.KEY_CHANNEL] == channel:
-  p_digest_mode.update(p_mode)
-  mode = p_digest_mode[mc.KEY_MODE]
-  MODE_KEY_MAP = {
-  mc.MTS200_MODE_HEAT: mc.KEY_HEATTEMP,
-  mc.MTS200_MODE_COOL: mc.KEY_COOLTEMP,
-  mc.MTS200_MODE_ECO: mc.KEY_ECOTEMP,
-  mc.MTS200_MODE_CUSTOM: mc.KEY_MANUALTEMP
-  }
-  if mode in MODE_KEY_MAP:
-  p_digest_mode[mc.KEY_TARGETTEMP] = p_digest_mode[MODE_KEY_MAP[mode]]
-  else:# we use this to trigger a windowOpened later in code
-  p_digest_windowopened_list = p_digest[mc.KEY_THERMOSTAT][mc.KEY_WINDOWOPENED]
-  if p_digest_mode[mc.KEY_ONOFF]:
-  p_digest_mode[mc.KEY_STATE] = 1 if p_digest_mode[mc.KEY_TARGETTEMP] > p_digest_mode[mc.KEY_CURRENTTEMP] else 0
-  else:
-  p_digest_mode[mc.KEY_STATE] = 0
-  break
+ p_digest_mode = get_element_by_key(
+ p_digest_mode_list,
+ mc.KEY_CHANNEL,
+ channel
+ )
+ p_digest_mode.update(p_mode)
+ mode = p_digest_mode[mc.KEY_MODE]
+ MODE_KEY_MAP = {
+ mc.MTS200_MODE_HEAT: mc.KEY_HEATTEMP,
+ mc.MTS200_MODE_COOL: mc.KEY_COOLTEMP,
+ mc.MTS200_MODE_ECO: mc.KEY_ECOTEMP,
+ mc.MTS200_MODE_CUSTOM: mc.KEY_MANUALTEMP
+ }
+ if mode in MODE_KEY_MAP:
+ p_digest_mode[mc.KEY_TARGETTEMP] = p_digest_mode[MODE_KEY_MAP[mode]]
+ else:# we use this to trigger a windowOpened later in code
+ p_digest_windowopened_list = p_digest[mc.KEY_THERMOSTAT][mc.KEY_WINDOWOPENED]
+ if p_digest_mode[mc.KEY_ONOFF]:
+ p_digest_mode[mc.KEY_STATE] = 1 if p_digest_mode[mc.KEY_TARGETTEMP] > p_digest_mode[mc.KEY_CURRENTTEMP] else 0
  else:
- raise Exception(f"{channel} not present in digest.thermostat")
+ p_digest_mode[mc.KEY_STATE] = 0
 
  # randomly switch the window
  for p_digest_windowopened in p_digest_windowopened_list:
@@ -44,17 +43,3 @@ def _SET_Appliance_Control_Thermostat_Mode(self, header, payload):
  break
 
  return mc.METHOD_SETACK, {}
-
-
- """
- def _GET_Appliance_Control_Thermostat_Sensor(self, header, payload):
- pass
-
-
- def _SET_Appliance_Control_Thermostat_Sensor(self, header, payload):
- if mc.NS_APPLIANCE_CONTROL_THERMOSTAT_SENSOR not in self.descriptor.namespaces:
- raise Exception(f"{mc.NS_APPLIANCE_CONTROL_THERMOSTAT_SENSOR} not supported in namespaces")
- mp3 = self.descriptor.namespaces[mc.NS_APPLIANCE_CONTROL_MP3]
- mp3[mc.KEY_MP3].update(payload[mc.KEY_MP3])
- return mc.METHOD_SETACK, {}
- """

custom_components/meross_lan/merossclient/__init__.py

@@ -120,6 +120,17 @@ def get_replykey(header: dict, key:KeyType = None) -> KeyType:
 
  return header
 
+def get_element_by_key(payload: list, key: str, value: object) -> dict:
+ """
+ scans the payload(list) looking for the first item matching
+ the key value. Usually looking for the matching channel payload
+ inside list paylaods
+ """
+ for p in payload:
+ if p.get(key) == value:
+ return p
+ raise KeyError(f"No match for key '{key}' on value:'{value}'")
+
 def get_productname(producttype: str) -> str:
  for _type, _name in mc.TYPE_NAME_MAP.items():
  if producttype.startswith(_type):