- Sync and async HTTP clients for Ignis Core WiZ server - 23 endpoints: auth, devices, groups, control, schedules, stats, API keys - Pydantic models with client-side validation - 108 unit tests - README with role table and usage examples
159 lines
5.3 KiB
Python
159 lines
5.3 KiB
Python
"""
|
||
Пример использования Ignis Client.
|
||
|
||
Скопируй каталог `ignis_client/` на целевую машину и положи рядом с ботом.
|
||
Единственные внешние зависимости — httpx и pydantic.
|
||
|
||
import sys
|
||
sys.path.insert(0, "/opt/my-bot")
|
||
from ignis_client import IgnisClient, CommandRequest
|
||
|
||
ignis = IgnisClient("http://192.168.1.50:8000", api_key="change-me")
|
||
"""
|
||
|
||
from ignis_client import (
|
||
IgnisClient,
|
||
AsyncIgnisClient,
|
||
CommandRequest,
|
||
ScheduleOnceRequest,
|
||
ScheduleCronRequest,
|
||
)
|
||
|
||
SERVER_URL = "http://127.0.0.1:8000"
|
||
API_KEY = "change-me"
|
||
|
||
|
||
def sync_example():
|
||
"""Синхронное использование — для простых скриптов и cron-задач."""
|
||
|
||
with IgnisClient(SERVER_URL, API_KEY) as ignis:
|
||
me = ignis.auth_me()
|
||
print(f"Авторизован как: {me['name']} (admin={me['is_admin']})")
|
||
|
||
devices = ignis.list_devices()
|
||
print(f"Устройств в сети: {len(devices)}")
|
||
|
||
groups = ignis.list_groups()
|
||
for gid, g in groups.items():
|
||
print(f" Группа {g['name']}: {g['device_ids']}")
|
||
|
||
if devices:
|
||
mac = next(iter(devices))
|
||
status = ignis.device_status(mac)
|
||
print(
|
||
f" Статус {mac}: on={status.status.get('state')}, "
|
||
f"brightness={status.status.get('dimming')}"
|
||
)
|
||
|
||
for gid in groups:
|
||
ignis.control_group(gid, CommandRequest(state=True, brightness=80))
|
||
print(f" Включил группу {gid}")
|
||
|
||
if me["is_admin"]:
|
||
tasks = ignis.list_schedules()
|
||
print(f"Активных расписаний: {len(tasks.tasks)}")
|
||
summary = ignis.stats_summary(days=7)
|
||
print(f"Статистика за {summary['period_days']} дн:")
|
||
for g in summary.get("groups", []):
|
||
print(f" {g['target_id']}: {g['total_commands']} команд")
|
||
else:
|
||
print("Гостевой ключ — расписания и статистика недоступны.")
|
||
|
||
|
||
def cron_example():
|
||
"""Создание cron-расписания для группы."""
|
||
|
||
with IgnisClient(SERVER_URL, API_KEY) as ignis:
|
||
req = ScheduleCronRequest(
|
||
target_id="bedroom",
|
||
hour="7",
|
||
minute="30",
|
||
day_of_week="mon-fri",
|
||
state=True,
|
||
brightness=60,
|
||
scene="wake_up",
|
||
)
|
||
result = ignis.create_cron_schedule(req)
|
||
print(f"Расписание создано: job_id={result.job_id}")
|
||
|
||
|
||
def scene_example():
|
||
"""Включение сцены и смена температуры."""
|
||
|
||
with IgnisClient(SERVER_URL, API_KEY) as ignis:
|
||
ignis.control_group(
|
||
"livingroom", CommandRequest(scene="fireplace", brightness=50)
|
||
)
|
||
|
||
ignis.control_device("aa:bb:cc:dd:ee:ff", CommandRequest(temp=3200))
|
||
|
||
ignis.control_group("bedroom", CommandRequest(r=255, g=120, b=60))
|
||
|
||
|
||
def blink_example():
|
||
"""Мигание лампой (поиск устройства в комнате)."""
|
||
|
||
with IgnisClient(SERVER_URL, API_KEY) as ignis:
|
||
result = ignis.blink_device("aa:bb:cc:dd:ee:ff")
|
||
print(f"Исходное состояние: {'вкл' if result.original else 'выкл'}")
|
||
|
||
|
||
def schedule_example():
|
||
"""Одноразовое расписание."""
|
||
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
with IgnisClient(SERVER_URL, API_KEY) as ignis:
|
||
run_at = datetime.now(timezone.utc) + timedelta(hours=1)
|
||
req = ScheduleOnceRequest(
|
||
target_id="livingroom",
|
||
run_at=run_at,
|
||
is_group=True,
|
||
state=False,
|
||
)
|
||
result = ignis.create_once_schedule(req)
|
||
print(f"Отложенное выключение: {result.job_id} в {result.run_at}")
|
||
|
||
|
||
def rescan_example():
|
||
"""Ручной рескан сети."""
|
||
|
||
with IgnisClient(SERVER_URL, API_KEY) as ignis:
|
||
result = ignis.rescan()
|
||
print(
|
||
f"Найдено: {result.found}, новых: {result.added}, онлайн: {result.online}"
|
||
)
|
||
|
||
|
||
async def async_example():
|
||
"""Асинхронное использование — для ботов на asyncio."""
|
||
|
||
async with AsyncIgnisClient(SERVER_URL, API_KEY) as ignis:
|
||
me = await ignis.auth_me()
|
||
print(f"Авторизован как: {me['name']}")
|
||
|
||
devices = await ignis.list_devices()
|
||
if devices:
|
||
mac = next(iter(devices))
|
||
await ignis.control_device(mac, CommandRequest(state=True, brightness=100))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
|
||
mode = sys.argv[1] if len(sys.argv) > 1 else "sync"
|
||
|
||
if mode == "sync":
|
||
sync_example()
|
||
elif mode == "cron":
|
||
cron_example()
|
||
elif mode == "async":
|
||
import asyncio
|
||
|
||
asyncio.run(async_example())
|
||
else:
|
||
print(f"Использование: python example.py [sync|cron|async]")
|
||
print(f" sync — показать устройства и включить группы")
|
||
print(f" cron — создать утреннее расписание")
|
||
print(f" async — асинхронный пример")
|