89 lines
2.3 KiB
Python
89 lines
2.3 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()
|
|
|
|
|
|
DISPOSITIVOS = {
|
|
"termo": "TAPO_TERMO_IP",
|
|
"pantalla": "TAPO_PANTALLA_IP",
|
|
}
|
|
|
|
|
|
async def _device(nombre="termo"):
|
|
var_ip = DISPOSITIVOS.get(nombre, "TAPO_TERMO_IP")
|
|
host = _env(var_ip)
|
|
username = _env("TAPO_USERNAME")
|
|
password = _env("TAPO_PASSWORD")
|
|
if not host or not username or not password:
|
|
raise RuntimeError(
|
|
f"Falta configurar {var_ip}, TAPO_USERNAME y TAPO_PASSWORD en .env"
|
|
)
|
|
return await Discover.discover_single(
|
|
host, username=username, password=password, timeout=8
|
|
)
|
|
|
|
|
|
async def _run(nombre, action):
|
|
device = await _device(nombre)
|
|
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"{nombre}: {state}"
|
|
|
|
|
|
def controlar(nombre, action):
|
|
"""Ejecuta una accion Tapo y devuelve un mensaje seguro para Telegram."""
|
|
try:
|
|
return asyncio.run(_run(nombre, 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 {nombre}: {message[:180]}"
|
|
|
|
|
|
def controlar_termo(action):
|
|
"""Compatibilidad: el termo es el dispositivo por defecto."""
|
|
return controlar("termo", action)
|
|
|
|
|
|
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())
|