dace7ab46c
- petkit_food_check.py: watchdog для state.food (алярм мало/пусто, сброс при доливе) - petkit_raw.py, petkit_roster.py: прямые запросы к API для отладки - .gitignore: исключить .food_state (runtime состояние)
95 lines
3.6 KiB
Python
95 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
PetKit food watchdog — проверяет уровень корма в фидере.
|
||
|
||
Поведение:
|
||
- Читает state.food текущего фидера.
|
||
- Хранит последний известный уровень в state-файле.
|
||
- Печатает в stdout уведомление ТОЛЬКО при переходе в "мало"(1) или "пусто"(0),
|
||
либо когда корм снова долили (возврат к полному) — чтобы сбросить тревогу.
|
||
- При первом запуске просто запоминает состояние, ничего не печатает.
|
||
|
||
Предназначен для крона в watchdog-режиме:
|
||
- непустой stdout -> уведомление доставляется пользователю
|
||
- пустой stdout -> тихий прогон, никому ничего не шлём
|
||
"""
|
||
import asyncio, os, sys
|
||
from aiohttp import ClientSession
|
||
from petkitaio import PetKitClient
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
STATE_FILE = os.path.join(HERE, ".food_state")
|
||
|
||
# Уровни корма у D4H: 2 = достаточно, 1 = мало, 0 = пусто
|
||
# (порог можно скорректировать после наблюдения за реальными значениями)
|
||
ALERT_AT = 1 # алертим когда food <= 1
|
||
|
||
|
||
def load_env():
|
||
p = os.path.join(HERE, ".env")
|
||
if os.path.exists(p):
|
||
for line in open(p):
|
||
line = line.strip()
|
||
if line and not line.startswith("#") and "=" in line:
|
||
k, _, v = line.partition("=")
|
||
os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))
|
||
|
||
|
||
def read_state():
|
||
try:
|
||
with open(STATE_FILE) as f:
|
||
return int(f.read().strip())
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def write_state(v):
|
||
with open(STATE_FILE, "w") as f:
|
||
f.write(str(v))
|
||
|
||
|
||
async def main():
|
||
load_env()
|
||
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")
|
||
|
||
s = ClientSession()
|
||
c = PetKitClient(email, password, s, region)
|
||
try:
|
||
await c.login()
|
||
devs = await c.get_petkit_data()
|
||
feeders = devs.feeders or {}
|
||
if not feeders:
|
||
sys.exit(0) # фидер не виден — молчим
|
||
fid = list(feeders.keys())[0]
|
||
f = feeders[fid]
|
||
food = f.data.get("state", {}).get("food")
|
||
name = f.data.get("name", fid)
|
||
if food is None:
|
||
sys.exit(0)
|
||
|
||
prev = read_state()
|
||
write_state(food)
|
||
|
||
# Тревога "мало/пусто" — только при переходе именно в это состояние
|
||
if prev is not None and prev > ALERT_AT and food <= ALERT_AT:
|
||
state_txt = "ПОЧТИ ПУСТО" if food == 0 else "НА ИСХОДЕ"
|
||
print(
|
||
f"\U0001f9d4 Фидер «{name}»: КОРМ {state_txt} (уровень {food}). "
|
||
f"Проверь и долей корм."
|
||
)
|
||
# Сброс тревоги — корм снова долили
|
||
elif prev is not None and prev <= ALERT_AT and food > ALERT_AT:
|
||
print(f"🟢 Фидер «{name}»: корм долили (уровень {food}). Тревога сброшена.")
|
||
|
||
# Остальное — тихо
|
||
finally:
|
||
await s.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|