114 lines
4.4 KiB
Python
114 lines
4.4 KiB
Python
#!/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())
|