05-sep: [tapo] controlar termo desde Telegram

vigilancia/tapo_control.py

@@ -0,0 +1,76 @@
+"""Control directo de dispositivos Tapo desde la Raspberry Pi."""
+
+import asyncio
+import os
+
+from kasa import Discover
+
+
+def _load_env():
+    values = {}
+    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
+    try:
+        with open(path) as stream:
+            for line in stream:
+                line = line.strip()
+                if line and not line.startswith("#") and "=" in line:
+                    key, value = line.split("=", 1)
+                    values[key.strip()] = value.strip()
+    except OSError:
+        pass
+    return values
+
+
+FILE_ENV = _load_env()
+
+
+def _env(name):
+    return os.getenv(name, FILE_ENV.get(name, "")).strip()
+
+
+async def _device():
+    host = _env("TAPO_TERMO_IP")
+    username = _env("TAPO_USERNAME")
+    password = _env("TAPO_PASSWORD")
+    if not host or not username or not password:
+        raise RuntimeError(
+            "Falta configurar TAPO_TERMO_IP, TAPO_USERNAME y TAPO_PASSWORD en .env"
+        )
+    return await Discover.discover_single(
+        host, username=username, password=password, timeout=8
+    )
+
+
+async def _run(action):
+    device = await _device()
+    await device.update()
+    if action == "on":
+        await device.turn_on()
+    elif action == "off":
+        await device.turn_off()
+    elif action != "status":
+        raise ValueError("Accion no soportada")
+    await device.update()
+    state = "encendido" if device.is_on else "apagado"
+    return f"termo: {state}"
+
+
+def controlar_termo(action):
+    """Ejecuta una accion Tapo y devuelve un mensaje seguro para Telegram."""
+    try:
+        return asyncio.run(_run(action))
+    except Exception as exc:
+        message = str(exc)
+        if "TAPO_" in message:
+            return f"Error de configuracion Tapo: {message}"
+        return f"No se pudo controlar el termo: {message[:180]}"
+
+
+async def _discover():
+    devices = await Discover.discover(timeout=5)
+    for device in devices.values():
+        print(f"{type(device).__name__} host={device.host!r} mac={device.mac}")
+
+
+if __name__ == "__main__":
+    asyncio.run(_discover())