feat(ha): cushion rain check via Gemini vision
New pyscript automation that snaps reolink_5_main when rain is incoming, asks Gemini (ai_task.google_ai_task) whether loose cushions/cushions are still on the terrace sofa, and alerts on mobile_app_is17 + Telegram sendPhoto when has_loose_cushions=True with confidence >= 70%. Triggers (all respect 3h cooldown): - period 30min: forecast precipitation > 0.5mm in next 2h - state: sensor.home_precipitation_intensity >= 0.1mm/h - state: weather.forecast_home transitions into rainy/pouring/snowy/hail - service: pyscript.cushion_check_now (manual) Adds TELEGRAM_BOT_TOKEN/TELEGRAM_CHAT_ID to HA container environment (HA scrubs env_file vars but honors hardcoded environment entries). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
9b069a15c4
commit
6328a7c275
|
|
@ -343,6 +343,8 @@ services:
|
|||
environment:
|
||||
- TZ=Europe/Prague
|
||||
- GRPC_VERBOSITY=NONE
|
||||
- TELEGRAM_BOT_TOKEN=${TELEGRAM_BOT_TOKEN}
|
||||
- TELEGRAM_CHAT_ID=${TELEGRAM_CHAT_ID}
|
||||
restart: unless-stopped
|
||||
network_mode: host
|
||||
security_opt:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,376 @@
|
|||
"""
|
||||
Cushion rain check - wykrywa poduszki na sofie po prawej stronie tarasu
|
||||
i alertuje zanim spadnie deszcz.
|
||||
|
||||
Triggery:
|
||||
- Cyklicznie co 30 min (sprawdza forecast 2h)
|
||||
- Natychmiastowo gdy zaczyna padać (sensor.home_precipitation_intensity >= 0.1)
|
||||
- Ręcznie: service pyscript.cushion_check_now
|
||||
|
||||
Kanaly:
|
||||
- notify.mobile_app_is17 (z thumbnailem)
|
||||
- Telegram bot (z pelnym obrazem)
|
||||
|
||||
Cooldown:
|
||||
- 3h po wyslaniu alertu (zeby nie spamowac)
|
||||
"""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
import socket
|
||||
import subprocess
|
||||
import urllib.request
|
||||
from datetime import datetime
|
||||
|
||||
GO2RTC_URL = "http://localhost:1984/api/frame.jpeg?src=reolink_5_main"
|
||||
# Default HA media_source "local" mapuje na /media (w kontenerze, niepersistowane)
|
||||
SNAPSHOT_DIR = "/media"
|
||||
SNAPSHOT_NAME = "cushion_check_latest.jpg"
|
||||
SNAPSHOT_PATH = f"{SNAPSHOT_DIR}/{SNAPSHOT_NAME}"
|
||||
# kopia do /config/www/tmp/ - dla HA Companion image attachment (/local/...)
|
||||
SNAPSHOT_WWW_DIR = "/config/www/tmp"
|
||||
SNAPSHOT_WWW_PATH = f"{SNAPSHOT_WWW_DIR}/{SNAPSHOT_NAME}"
|
||||
PUBLIC_URL_PATH = f"/local/tmp/{SNAPSHOT_NAME}"
|
||||
# media_source dla ai_task attachments
|
||||
MEDIA_SOURCE_ID = f"media-source://media_source/local/{SNAPSHOT_NAME}"
|
||||
|
||||
AI_ENTITY = "ai_task.google_ai_task"
|
||||
WEATHER_ENTITY = "weather.forecast_home"
|
||||
RAIN_NOW_SENSOR = "sensor.home_precipitation_intensity"
|
||||
LAST_CHECK_ENTITY = "pyscript.cushion_last_check"
|
||||
|
||||
COOLDOWN_HOURS = 3
|
||||
FORECAST_PRECIP_THRESHOLD_MM = 0.5
|
||||
RAIN_NOW_THRESHOLD_MM = 0.1
|
||||
|
||||
INSTRUCTIONS = """Analizujesz obraz z kamery zewnetrznej (widok z lotu ptaka na taras o kamiennej posadzce).
|
||||
|
||||
KONTEKST SCENY:
|
||||
Po PRAWEJ stronie kadru znajduje sie taras z meblami ogrodowymi. Glowny mebel do obserwacji to SOFA OGRODOWA z metalowym/wiklinowym stelazem, ktora ma KOMPLET TEKSTYLIOW: dwa MATERACE SIEDZISKOWE (dolne, plaskie, ciemne) plus trzy PODUSZKI OPARCIOWE (gorne, kwadratowe, ciemne). W tej scenie tekstylia sa CIEMNOSZARE / CZARNE.
|
||||
|
||||
ZADANIE: Sprawdz czy te tekstylia (materace + poduszki) sa OBECNIE na sofie. Wszystkie sa luzne i moga sie zniszczyc od deszczu wiec trzeba je zabrac.
|
||||
|
||||
CO LICZYC jako tekstylia (rosnij count o 1 za kazdy widoczny element):
|
||||
- Materac siedziskowy luzno polozony na ramie sofy
|
||||
- Poduszka oparciowa lub dekoracyjna na sofie/oparciu
|
||||
- Dodatkowo: luzne koce, narzuty, mniejsze poduszki dorzucone
|
||||
|
||||
CZEGO NIE LICZYC (False positives - traktuj jako NIE):
|
||||
- Same metalowe / wiklinowe / drewniane elementy konstrukcji mebli BEZ tekstyliow
|
||||
- Plandeka zakrywajaca mebel (jesli widac plandeke - tekstylia juz zabezpieczone)
|
||||
- Doniczki, donice, kamienie, posadzka, dywany podlogowe
|
||||
- Cienie, plamy swiatla
|
||||
- Krzesla po przeciwnej (lewej) stronie tarasu
|
||||
|
||||
ZASADA: Jesli na ramie sofy po prawej WIDAC ciemne/kolorowe miekkie elementy - to sa tekstylia, has_loose_cushions=true. Jesli widac sama goly metalowy/wiklinowy stelaz bez tekstyliow - has_loose_cushions=false.
|
||||
|
||||
Odpowiedz w formacie strukturalnym (JSON), pola:
|
||||
- has_loose_cushions (bool): true gdy na sofie po prawej sa widoczne JAKIEKOLWIEK tekstylia (materace, poduszki, koce)
|
||||
- count (int): laczna liczba widocznych tekstyliow (materace + poduszki + dodatkowe). Sofa w komplecie = 5 (2 materace + 3 poduszki)
|
||||
- confidence (int 0-100): pewnosc oceny
|
||||
- description (string, po polsku, max 150 znakow): co konkretnie widzisz na sofie po prawej
|
||||
|
||||
PRZYKLAD KOMPLET: has_loose_cushions=true, count=5, confidence=90, description="Sofa po prawej w komplecie - widoczne 2 materace siedziskowe i 3 poduszki oparciowe"
|
||||
PRZYKLAD CZESCIOWY: has_loose_cushions=true, count=2, confidence=85, description="Na sofie po prawej widoczne tylko 2 materace, brak poduszek oparciowych"
|
||||
PRZYKLAD PUSTY: has_loose_cushions=false, count=0, confidence=90, description="Sofa po prawej pusta - widac sam metalowy stelaz, tekstylia zabrane"
|
||||
PRZYKLAD ZASLONIETY: has_loose_cushions=false, count=0, confidence=80, description="Sofa zakryta plandeka - tekstylia juz zabezpieczone"
|
||||
"""
|
||||
|
||||
AI_STRUCTURE = {
|
||||
"has_loose_cushions": {
|
||||
"selector": {"boolean": {}},
|
||||
"description": "Czy widac LUZNE poduszki dekoracyjne (nie wbudowane siedziska)",
|
||||
"required": True,
|
||||
},
|
||||
"count": {
|
||||
"selector": {"number": {"min": 0, "max": 20}},
|
||||
"description": "Liczba luznych poduszek dekoracyjnych",
|
||||
"required": True,
|
||||
},
|
||||
"confidence": {
|
||||
"selector": {"number": {"min": 0, "max": 100}},
|
||||
"description": "Pewnosc procentowa 0-100",
|
||||
"required": True,
|
||||
},
|
||||
"description": {
|
||||
"selector": {"text": {}},
|
||||
"description": "Co widzisz na prawej stronie tarasu",
|
||||
"required": True,
|
||||
},
|
||||
}
|
||||
|
||||
# Prog pewnosci - powiadomienie tylko gdy AI naprawde widzi poduszki
|
||||
MIN_CONFIDENCE_FOR_ALERT = 70
|
||||
|
||||
|
||||
# Snapshot i Telegram send sa wykonywane inline w _run_check / _notify
|
||||
# przez task.executor(stdlib_function, args...) - pyscript NIE pozwala
|
||||
# wywolac task.executor() na funkcjach zdefiniowanych w pyscript.
|
||||
|
||||
|
||||
def _get_last_check():
|
||||
try:
|
||||
return state.get(LAST_CHECK_ENTITY)
|
||||
except (NameError, Exception):
|
||||
return None
|
||||
|
||||
|
||||
def _on_cooldown():
|
||||
last = _get_last_check()
|
||||
if not last or last in ("never", "unknown", "unavailable"):
|
||||
return False
|
||||
try:
|
||||
last_ts = datetime.fromisoformat(last)
|
||||
delta_h = (datetime.now() - last_ts).total_seconds() / 3600.0
|
||||
if delta_h < COOLDOWN_HOURS:
|
||||
log.info(
|
||||
f"cushion_check: cooldown {delta_h:.2f}h/{COOLDOWN_HOURS}h - skip"
|
||||
)
|
||||
return True
|
||||
except Exception as e:
|
||||
log.warning(f"cushion_check: cooldown parse err: {e}")
|
||||
return False
|
||||
|
||||
|
||||
def _rain_now():
|
||||
try:
|
||||
return float(state.get(RAIN_NOW_SENSOR) or 0) >= RAIN_NOW_THRESHOLD_MM
|
||||
except (TypeError, ValueError):
|
||||
return False
|
||||
|
||||
|
||||
def _rain_in_forecast_2h():
|
||||
try:
|
||||
resp = service.call(
|
||||
"weather",
|
||||
"get_forecasts",
|
||||
entity_id=WEATHER_ENTITY,
|
||||
type="hourly",
|
||||
return_response=True,
|
||||
blocking=True,
|
||||
)
|
||||
except Exception as e:
|
||||
log.warning(f"cushion_check: get_forecasts err: {e}")
|
||||
return False
|
||||
|
||||
data = resp.get(WEATHER_ENTITY, {}) if isinstance(resp, dict) else {}
|
||||
forecasts = (data.get("forecast") or [])[:2]
|
||||
total = 0.0
|
||||
for h in forecasts:
|
||||
try:
|
||||
total += float(h.get("precipitation") or 0)
|
||||
except (TypeError, ValueError):
|
||||
pass
|
||||
return total >= FORECAST_PRECIP_THRESHOLD_MM
|
||||
|
||||
|
||||
def _notify(count, confidence, desc, reason):
|
||||
rain_now = state.get(RAIN_NOW_SENSOR) or "?"
|
||||
title = f"Poduszki na tarasie ({count} szt.)"
|
||||
body = f"AI wykrylo {count} luznych poduszek (pewnosc {confidence}%).\n"
|
||||
body += f"Powod: {reason} | Deszcz teraz: {rain_now} mm/h\n"
|
||||
if desc:
|
||||
body += f"{desc}"
|
||||
|
||||
try:
|
||||
notify.mobile_app_is17(
|
||||
title=title,
|
||||
message=body,
|
||||
data={
|
||||
"image": PUBLIC_URL_PATH,
|
||||
"tag": "cushion_rain",
|
||||
"channel": "weather",
|
||||
"priority": "high",
|
||||
"ttl": 3600,
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
log.error(f"cushion_check: notify is17 fail: {e}")
|
||||
|
||||
token = os.environ.get("TELEGRAM_BOT_TOKEN")
|
||||
chat = os.environ.get("TELEGRAM_CHAT_ID")
|
||||
if token and chat:
|
||||
caption = f"{title}\n{body}"
|
||||
url = f"https://api.telegram.org/bot{token}/sendPhoto"
|
||||
cmd = [
|
||||
"curl", "-s", "-S", "--max-time", "20",
|
||||
"-X", "POST",
|
||||
"-F", f"chat_id={chat}",
|
||||
"-F", f"caption={caption}",
|
||||
"-F", f"photo=@{SNAPSHOT_PATH}",
|
||||
url,
|
||||
]
|
||||
try:
|
||||
result = task.executor(
|
||||
subprocess.run, cmd, capture_output=True, text=True, timeout=25
|
||||
)
|
||||
ok = result.returncode == 0 and '"ok":true' in (result.stdout or "")
|
||||
if ok:
|
||||
log.info("cushion_check: telegram OK")
|
||||
else:
|
||||
log.error(
|
||||
f"cushion_check: telegram fail rc={result.returncode} "
|
||||
f"stdout={(result.stdout or '')[:200]} stderr={(result.stderr or '')[:200]}"
|
||||
)
|
||||
except Exception as e:
|
||||
log.error(f"cushion_check: telegram exception: {e}")
|
||||
else:
|
||||
log.warning("cushion_check: brak TELEGRAM env, pomijam telegram")
|
||||
|
||||
|
||||
def _run_check(reason):
|
||||
log.info(f"cushion_check START reason={reason}")
|
||||
try:
|
||||
if not os.path.isdir(SNAPSHOT_DIR):
|
||||
os.makedirs(SNAPSHOT_DIR, exist_ok=True)
|
||||
if not os.path.isdir(SNAPSHOT_WWW_DIR):
|
||||
os.makedirs(SNAPSHOT_WWW_DIR, exist_ok=True)
|
||||
socket.setdefaulttimeout(15)
|
||||
task.executor(urllib.request.urlretrieve, GO2RTC_URL, SNAPSHOT_PATH)
|
||||
sz = os.path.getsize(SNAPSHOT_PATH) if os.path.exists(SNAPSHOT_PATH) else 0
|
||||
if sz < 5000:
|
||||
raise RuntimeError(f"snapshot empty/small ({sz}B)")
|
||||
# kopia do www/tmp dla HA Companion image
|
||||
try:
|
||||
task.executor(shutil.copyfile, SNAPSHOT_PATH, SNAPSHOT_WWW_PATH)
|
||||
except Exception as e:
|
||||
log.warning(f"cushion_check: copy to www fail (companion image may not work): {e}")
|
||||
log.info(f"cushion_check: snapshot {sz}B -> {SNAPSHOT_PATH}")
|
||||
except Exception as e:
|
||||
log.error(f"cushion_check: snapshot fail: {e}")
|
||||
return
|
||||
|
||||
try:
|
||||
ai_resp = service.call(
|
||||
"ai_task",
|
||||
"generate_data",
|
||||
entity_id=AI_ENTITY,
|
||||
task_name="cushion_check",
|
||||
instructions=INSTRUCTIONS,
|
||||
structure=AI_STRUCTURE,
|
||||
attachments=[
|
||||
{
|
||||
"media_content_id": MEDIA_SOURCE_ID,
|
||||
"media_content_type": "image/jpeg",
|
||||
}
|
||||
],
|
||||
return_response=True,
|
||||
blocking=True,
|
||||
)
|
||||
except Exception as e:
|
||||
log.error(f"cushion_check: ai_task fail: {e}")
|
||||
return
|
||||
|
||||
log.info(f"cushion_check AI resp: {str(ai_resp)[:400]}")
|
||||
|
||||
data = {}
|
||||
if isinstance(ai_resp, dict):
|
||||
d = ai_resp.get("data")
|
||||
if isinstance(d, dict):
|
||||
data = d
|
||||
|
||||
has_cushions = bool(data.get("has_loose_cushions", False))
|
||||
try:
|
||||
count = int(data.get("count", 0) or 0)
|
||||
except (TypeError, ValueError):
|
||||
count = 0
|
||||
try:
|
||||
confidence = int(data.get("confidence", 0) or 0)
|
||||
except (TypeError, ValueError):
|
||||
confidence = 0
|
||||
desc = str(data.get("description", "") or "")[:300]
|
||||
|
||||
state.set(
|
||||
LAST_CHECK_ENTITY,
|
||||
datetime.now().isoformat(timespec="seconds"),
|
||||
new_attributes={
|
||||
"has_cushions": has_cushions,
|
||||
"count": count,
|
||||
"confidence": confidence,
|
||||
"desc": desc[:200],
|
||||
"reason": reason,
|
||||
},
|
||||
)
|
||||
|
||||
log.info(
|
||||
f"cushion_check RESULT has_cushions={has_cushions} count={count} "
|
||||
f"confidence={confidence}% desc={desc[:140]}"
|
||||
)
|
||||
|
||||
if has_cushions and count > 0 and confidence >= MIN_CONFIDENCE_FOR_ALERT:
|
||||
_notify(count, confidence, desc, reason)
|
||||
else:
|
||||
log.info(
|
||||
f"cushion_check: brak alertu (has_cushions={has_cushions}, "
|
||||
f"count={count}, confidence={confidence}%, prog={MIN_CONFIDENCE_FOR_ALERT}%)"
|
||||
)
|
||||
|
||||
|
||||
# === TRIGGERY ===
|
||||
|
||||
@time_trigger("startup")
|
||||
def cushion_init():
|
||||
try:
|
||||
state.persist(
|
||||
LAST_CHECK_ENTITY,
|
||||
default_value="never",
|
||||
default_attributes={
|
||||
"answer": "never",
|
||||
"count": 0,
|
||||
"desc": "",
|
||||
"reason": "init",
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
log.warning(f"cushion_init: state.persist err: {e}")
|
||||
try:
|
||||
state.set(
|
||||
LAST_CHECK_ENTITY,
|
||||
"never",
|
||||
new_attributes={"answer": "never", "count": 0, "desc": "", "reason": "init"},
|
||||
)
|
||||
except Exception as e2:
|
||||
log.error(f"cushion_init: state.set err: {e2}")
|
||||
log.info("cushion_rain_check: zainicjalizowane")
|
||||
|
||||
|
||||
@time_trigger("period(now, 30min)")
|
||||
def cushion_periodic():
|
||||
if _on_cooldown():
|
||||
return
|
||||
if _rain_now():
|
||||
_run_check("rain_now (periodic)")
|
||||
return
|
||||
if _rain_in_forecast_2h():
|
||||
_run_check("forecast_2h")
|
||||
|
||||
|
||||
@state_trigger(f"float({RAIN_NOW_SENSOR} or 0) >= {RAIN_NOW_THRESHOLD_MM}")
|
||||
def cushion_rain_started():
|
||||
if _on_cooldown():
|
||||
return
|
||||
_run_check("rain_started")
|
||||
|
||||
|
||||
WET_WEATHER_STATES = {
|
||||
"rainy", "pouring", "snowy", "snowy-rainy", "lightning-rainy", "hail"
|
||||
}
|
||||
|
||||
|
||||
@state_trigger(f"{WEATHER_ENTITY}")
|
||||
def cushion_weather_changed(value=None, old_value=None):
|
||||
"""Reaguj na zmiane stanu pogody na mokre warunki."""
|
||||
if value not in WET_WEATHER_STATES:
|
||||
return
|
||||
if old_value in WET_WEATHER_STATES:
|
||||
# juz wczesniej bylo mokro, to tylko refresh - nie reagujemy
|
||||
return
|
||||
if _on_cooldown():
|
||||
return
|
||||
_run_check(f"weather_change {old_value}->{value}")
|
||||
|
||||
|
||||
@service("pyscript.cushion_check_now")
|
||||
def cushion_check_now():
|
||||
"""Manualny test - ignoruje cooldown."""
|
||||
_run_check("manual")
|
||||
Loading…
Reference in New Issue