Add PetKit CLI wrapper (status/feed/fountain/dump)

2026-08-04 05:55:35 +00:00
commit b5c497530d
2 changed files with 123 additions and 0 deletions
--- a/petkit_cli.py
+++ b/petkit_cli.py
@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""
+PetKit CLI wrapper — статус фидеров/фонтанов + дамп структуры.
+
+Режимы:
+  status     — показать понятный статус всех устройств
+  dump       — вывести сырую JSON-структуру фидеров (для отладки полей)
+  feed ГР     — покормить в граммах
+  fountain РЕЖИМ — переключить фонтан (smart|normal|pause ...)
+
+Креды читаются из .env рядом со скриптом:
+  PETKIT_EMAIL, PETKIT_PASSWORD
+Регион задаётся через PETKIT_REGION (по умолчанию "Russian Federation").
+"""
+import asyncio, json, os, sys
+
+from aiohttp import ClientSession
+from petkitaio import PetKitClient
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+
+
+def load_env():
+    env_path = os.path.join(HERE, ".env")
+    if os.path.exists(env_path):
+        with open(env_path) as f:
+            for line in f:
+                line = line.strip()
+                if not line or line.startswith("#") or "=" not in line:
+                    continue
+                k, _, v = line.partition("=")
+                os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
+
+
+async def get_client() -> PetKitClient:
+    email = os.environ.get("PETKIT_EMAIL")
+    password = os.environ.get("PETKIT_PASSWORD")
+    region = os.environ.get("PETKIT_REGION", "Russian Federation")
+    if not email or not password:
+        sys.exit("ОШИБКА: задайте PETKIT_EMAIL и PETKIT_PASSWORD в .env (PetKit не менял пользователя).")
+    session = ClientSession()
+    client = PetKitClient(email, password, session, region)
+    await client.login()
+    return client, session
+
+
+async def cmd_status(client: PetKitClient):
+    devices = await client.get_petkit_data()
+    lines = []
+    fids = {}
+
+    feeders = devices.feeders or {}
+    for fid, feeder in feeders.items():
+        fids[fid] = feeder
+        d = feeder.data
+        st = d.get("state", {})
+        lines.append(f"[Feeder] {d.get('name', fid)} (id={fid}, type={feeder.type})")
+        lines.append(f"  food sensor: {st.get('food')}  feeding: {st.get('feeding')}")
+        lines.append(f"  battery: {st.get('batteryPower')} ({st.get('batteryStatus')})")
+        lines.append(f"  foodWarn(food low alarm): {d.get('foodWarn', st.get('foodWarn', '?'))}")
+
+    for wid, wf in (devices.water_fountains or {}).items():
+        d = wf.data
+        lines.append(f"[Fountain] {d.get('name', wid)} (id={wid})")
+
+    if not lines:
+        lines.append("Нет устройств (или не удалось получить данные).")
+    print("\n".join(lines))
+    return fids
+
+
+async def cmd_dump(client: PetKitClient):
+    devices = await client.get_petkit_data()
+    out = {"feeders": {}, "fountains": {}}
+    for fid, feeder in (devices.feeders or {}).items():
+        out["feeders"][str(fid)] = feeder.data
+    for wid, wf in (devices.water_fountains or {}).items():
+        out["fountains"][str(wid)] = wf.data
+    print(json.dumps(out, ensure_ascii=False, indent=2, default=str))
+
+
+async def main():
+    load_env()
+    mode = sys.argv[1] if len(sys.argv) > 1 else "status"
+    client, session = await get_client()
+    try:
+        if mode == "dump":
+            await cmd_dump(client)
+        elif mode == "feed":
+            amount = int(sys.argv[2]) if len(sys.argv) > 2 else 10
+            devices = await client.get_petkit_data()
+            fid = list((devices.feeders or {}).keys())[0]
+            await client.manual_feeding(feeder=devices.feeders[fid], amount=amount)
+            print(f"Покормил {amount}г (feeder {fid})")
+        elif mode == "fountain":
+            command = sys.argv[2] if len(sys.argv) > 2 else "smart"
+            from petkitaio.constants import W5Command
+            devices = await client.get_petkit_data()
+            wid = list((devices.water_fountains or {}).keys())[0]
+            cmd_map = {"smart": W5Command.SMART, "normal": W5Command.NORMAL}
+            wcmd = cmd_map.get(command)
+            if not wcmd:
+                sys.exit(f"Неизвестный режим фонтана: {command}")
+            await client.control_water_fountain(water_fountain=devices.water_fountains[wid], command=wcmd)
+            print(f"Фонтан {wid} -> {command}")
+        else:
+            await cmd_status(client)
+    finally:
+        await session.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())