133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
"""Экспорт данных по теплице из Home Assistant за указанный период (stdlib only)."""
|
||
|
||
import json
|
||
import sys
|
||
import urllib.request
|
||
import urllib.parse
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
# ============ НАСТРОЙКИ (подставлены из .env и parameters.yaml) ============
|
||
HA_URL = "https://ha.homenet542.keenetic.pro"
|
||
TOKEN = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJmOTkyNzMxNmNlZTI0MjYzOWU4NjRhMGZlOGI2OTExZSIsImlhdCI6MTc3NTIzOTM1OCwiZXhwIjoyMDkwNTk5MzU4fQ.eumM2Vhk68uZZTvA4uIjKDqzlwBPKhBV6JeVRmSAJos"
|
||
DAYS = 7
|
||
ENTITIES = [
|
||
"sensor.ulitsa_teplitsa_klimat_temperature",
|
||
"sensor.ulitsa_teplitsa_klimat_humidity",
|
||
"switch.ulitsa_teplitsa_vytiazhka",
|
||
"switch.ulitsa_teplitsa_obogrev",
|
||
"sun.sun",
|
||
]
|
||
OUTPUT_FILE = "/home/node/.openclaw/workspace/teplitsa_export.json"
|
||
# ====================================
|
||
|
||
|
||
def api_get(path, params=None):
|
||
headers = {
|
||
"Authorization": f"Bearer {TOKEN}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
url = f"{HA_URL}{path}"
|
||
if params:
|
||
url = f"{url}?{urllib.parse.urlencode(params)}"
|
||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||
return json.loads(resp.read().decode("utf-8"))
|
||
|
||
|
||
def main():
|
||
start = datetime.now(timezone.utc) - timedelta(days=DAYS)
|
||
start_iso = start.isoformat()
|
||
|
||
print(f"📊 Сбор данных за {DAYS} дней с {start_iso[:10]}...")
|
||
|
||
try:
|
||
# История сенсоров и переключателей
|
||
params = {
|
||
"filter_entity_id": ",".join(ENTITIES),
|
||
"minimal_response": "true",
|
||
}
|
||
history_raw = api_get(f"/api/history/period/{start_iso}", params)
|
||
# Фильтруем unavailable/unknown
|
||
history = []
|
||
for series in history_raw:
|
||
filtered = [s for s in series if s.get("state") not in ("unavailable", "unknown", None, "")]
|
||
history.append(filtered)
|
||
total = sum(len(s) for s in history)
|
||
print(f"✅ История: {total} записей по {len(history)} сущностям")
|
||
|
||
# Логбук
|
||
logbook_raw = api_get(f"/api/logbook/{start_iso}")
|
||
teplitsa_log = [
|
||
e for e in logbook_raw
|
||
if e.get("entity_id") in ENTITIES
|
||
or "теплиц" in (e.get("name", "") or "").lower()
|
||
]
|
||
print(f"✅ Логбук: {len(teplitsa_log)} релевантных событий")
|
||
|
||
# Текущие состояния
|
||
states = []
|
||
for eid in ENTITIES:
|
||
try:
|
||
states.append(api_get(f"/api/states/{eid}"))
|
||
except Exception as e:
|
||
print(f"⚠️ Не удалось получить {eid}: {e}")
|
||
|
||
except urllib.error.HTTPError as e:
|
||
print(f"❌ HTTP ошибка {e.code}: {e.reason}")
|
||
sys.exit(1)
|
||
except urllib.error.URLError as e:
|
||
print(f"❌ Ошибка соединения: {e.reason}")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
print(f"❌ Ошибка: {e}")
|
||
sys.exit(1)
|
||
|
||
output = {
|
||
"exported_at": datetime.now().isoformat(),
|
||
"period_days": DAYS,
|
||
"entities": ENTITIES,
|
||
"current_states": states,
|
||
"history": history,
|
||
"logbook": teplitsa_log,
|
||
}
|
||
|
||
with open(OUTPUT_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(output, f, ensure_ascii=False, indent=2, default=str)
|
||
|
||
print(f"\n💾 Сохранено: {OUTPUT_FILE}")
|
||
|
||
# Краткая сводка
|
||
print("\n" + "=" * 50)
|
||
print("📈 КРАТКАЯ СВОДКА")
|
||
print("=" * 50)
|
||
|
||
for series in history:
|
||
if not series:
|
||
continue
|
||
eid = series[0].get("entity_id", "?")
|
||
if eid.startswith("sensor."):
|
||
try:
|
||
values = [float(s["state"]) for s in series]
|
||
if values:
|
||
print(f"\n{eid}")
|
||
print(f" Мин/Макс/Среднее: {min(values):.1f} / {max(values):.1f} / {sum(values)/len(values):.1f}")
|
||
except (ValueError, KeyError):
|
||
pass
|
||
elif eid.startswith("switch."):
|
||
transitions = sum(
|
||
1 for i in range(1, len(series))
|
||
if series[i]["state"] != series[i - 1]["state"]
|
||
)
|
||
on_count = sum(1 for s in series if s["state"] == "on")
|
||
off_count = sum(1 for s in series if s["state"] == "off")
|
||
print(f"\n{eid}")
|
||
print(f" Состояний 'on': {on_count}, 'off': {off_count}")
|
||
print(f" Переключений: {transitions}")
|
||
|
||
print("\n📤 Загрузи файл teplitsa_export.json в чат для анализа.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|