commit b5c497530d16d9becd783e8d84faa996cdd77ed4 Author: hermes-home Date: Tue Aug 4 05:55:35 2026 +0000 Add PetKit CLI wrapper (status/feed/fountain/dump) diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ee5ce70 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +# Секреты — не заливать в репо +.env +secrets/ +*.token + +# Python +__pycache__/ +*.pyc +.venv/ +venv/ diff --git a/petkit_cli.py b/petkit_cli.py new file mode 100644 index 0000000..b9fc724 --- /dev/null +++ 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())