77 lines
2.0 KiB
Python
77 lines
2.0 KiB
Python
"""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())
|