06-sep: [domotica] Mesillas Meross (comando + submenu)

vigilancia/meross_control.py

@@ -0,0 +1,114 @@
+"""Enchufes Meross (mesillas) via cuenta Meross en la nube.
+
+Lee MEROSS_EMAIL/PASSWORD del .env. Solo lectura salvo on/off explicito.
+Uso: python meross_control.py estado|on|off [nombre]
+"""
+import asyncio
+import os
+import ssl
+import sys
+
+BASE = os.path.dirname(os.path.abspath(__file__))
+if BASE not in sys.path:
+    sys.path.insert(0, BASE)
+
+
+def _env(nombre):
+    valor = os.getenv(nombre, "").strip()
+    if not valor:
+        try:
+            with open(os.path.join(BASE, ".env")) as f:
+                for linea in f:
+                    linea = linea.strip()
+                    if linea.startswith(nombre + "="):
+                        valor = linea.split("=", 1)[1].strip()
+                        break
+        except OSError:
+            pass
+    return valor
+
+
+async def _dispositivos():
+    from meross_iot.http_api import MerossHttpClient
+    from meross_iot.manager import MerossManager
+    http = await MerossHttpClient.async_from_user_password(
+        api_base_url="https://iotx-eu.meross.com",
+        email=_env("MEROSS_EMAIL"), password=_env("MEROSS_PASSWORD"),
+        ssl_context=ssl.create_default_context())
+    mgr = MerossManager(http_client=http)
+    await mgr.async_init()
+    await mgr.async_device_discovery()
+    return mgr, http
+
+
+def _estado_dev(d):
+    try:
+        v = getattr(d, "is_on", None)
+        if callable(v):
+            v = v()
+        if isinstance(v, bool):
+            return v
+    except Exception:
+        pass
+    return None
+
+
+async def _actuar(nombre, accion):
+    mgr, http = await _dispositivos()
+    try:
+        objetivo = (nombre or "").lower()
+        elegidos = [d for d in mgr.find_devices()
+                    if not objetivo or objetivo in (d.name or "").lower()]
+        if not elegidos:
+            return [f"❌ No encuentro '{nombre}' (solo mesillas)."]
+        lineas = []
+        for d in elegidos:
+            try:
+                await d.async_update()
+            except Exception:
+                pass
+        for d in elegidos:
+            if accion == "on":
+                try:
+                    await d.async_turn_on()
+                except Exception as e:
+                    lineas.append(f"❌ {d.name}: {str(e)[:100]}")
+                    continue
+            elif accion == "off":
+                try:
+                    await d.async_turn_off()
+                except Exception as e:
+                    lineas.append(f"❌ {d.name}: {str(e)[:100]}")
+                    continue
+            try:
+                await d.async_update()
+            except Exception:
+                pass
+            on = _estado_dev(d)
+            txt = "encendido" if on else ("apagado" if on is False else "?")
+            lineas.append(f"🛏️ {d.name}: <b>{txt}</b>")
+        return lineas
+    finally:
+        mgr.close()
+        await http.async_logout()
+
+
+def controlar(nombre, accion):
+    """Devuelve lista de lineas HTML con el resultado."""
+    try:
+        return asyncio.run(_actuar(nombre or "", accion or "estado"))
+    except Exception as e:
+        return [f"❌ Error Meross: {str(e)[:150]}"]
+
+
+def main(argv):
+    if len(argv) < 2 or argv[1] not in ("estado", "on", "off"):
+        print("Uso: meross_control.py estado|on|off [nombre]")
+        return 1
+    for linea in controlar(argv[2] if len(argv) > 2 else "", argv[1]):
+        print(linea.replace("<b>", "").replace("</b>", ""))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))