115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
"""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))
|