Add PetKit food watchdog + D4H raw probes
- petkit_food_check.py: watchdog для state.food (алярм мало/пусто, сброс при доливе) - petkit_raw.py, petkit_roster.py: прямые запросы к API для отладки - .gitignore: исключить .food_state (runtime состояние)
This commit is contained in:
94
petkit_food_check.py
Normal file
94
petkit_food_check.py
Normal file
@@ -0,0 +1,94 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user