140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
"""Extract IoT rows for cycle window — scan only pages likely in date range."""
|
|
import json
|
|
import urllib.request
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
import sys
|
|
|
|
sys.path.insert(0, str(SCRIPT_DIR))
|
|
from experience_temperature import (
|
|
LANTAI_1_IOT_FLOCK_ID,
|
|
LANTAI_2_IOT_FLOCK_ID,
|
|
LANTAI_2_ADJUST,
|
|
clone_lantai2_panel_row,
|
|
panel_fields_from_payload,
|
|
)
|
|
|
|
BASE = "https://dashboard.cpsp.id/api/iot/flocks/"
|
|
FLOCK_ID = LANTAI_1_IOT_FLOCK_ID
|
|
CYCLE_START = datetime(2026, 5, 22).date()
|
|
CYCLE_END = datetime(2026, 7, 9).date()
|
|
TZ = timezone(timedelta(hours=7))
|
|
|
|
|
|
def parse_dt(value: str) -> datetime:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
|
|
|
|
def fetch_page(page: int) -> dict:
|
|
with urllib.request.urlopen(f"{BASE}?page={page}", timeout=60) as resp:
|
|
return json.loads(resp.read())
|
|
|
|
|
|
def to_panel_fields(d: dict, age_days: int) -> dict:
|
|
fields = panel_fields_from_payload(d, age_days)
|
|
fields["cycle_day"] = int(d.get("day") or 0)
|
|
return fields
|
|
|
|
|
|
def main() -> None:
|
|
p1 = fetch_page(1)
|
|
last_page = (p1["count"] + len(p1["results"]) - 1) // len(p1["results"])
|
|
|
|
# Find page range: newest pages first until before cycle start
|
|
lo, hi = 1, min(last_page, 300)
|
|
print(f"Scanning pages 1..{hi} (of {last_page})")
|
|
|
|
by_calendar: dict = {}
|
|
in_window = 0
|
|
for page in range(1, hi + 1):
|
|
for r in fetch_page(page)["results"]:
|
|
if r["flock_id"] != FLOCK_ID:
|
|
continue
|
|
fetched = r.get("fetched_at")
|
|
if not fetched:
|
|
continue
|
|
cal = parse_dt(fetched).astimezone(TZ).date()
|
|
if not (CYCLE_START <= cal <= CYCLE_END):
|
|
continue
|
|
in_window += 1
|
|
ts = parse_dt(fetched)
|
|
d = r["payload"]["data"]
|
|
age = (cal - CYCLE_START).days + 1
|
|
prev = by_calendar.get(cal)
|
|
if prev is None or ts > prev["ts"]:
|
|
by_calendar[cal] = {"ts": ts, "fields": to_panel_fields(d, age)}
|
|
|
|
missing = []
|
|
d = CYCLE_START
|
|
while d <= CYCLE_END:
|
|
if d not in by_calendar:
|
|
missing.append(d)
|
|
d += timedelta(days=1)
|
|
|
|
print(f"snapshots in window: {in_window}")
|
|
print(f"unique dates: {len(by_calendar)}/49, missing: {len(missing)}")
|
|
if missing:
|
|
print(f"missing: {[str(x) for x in missing]}")
|
|
|
|
print(
|
|
f"\n{'date':>10} {'age':>3} {'wind':>6} {'hum%':>6} {'avgT':>6} {'expT':>6} {'water':>8}"
|
|
)
|
|
d = CYCLE_START
|
|
while d <= CYCLE_END:
|
|
e = by_calendar.get(d)
|
|
age = (d - CYCLE_START).days + 1
|
|
if e:
|
|
f = e["fields"]
|
|
print(
|
|
f"{d} {age:3d} {f['wind_speed']:6.2f} {f['humidity']:6.1f} "
|
|
f"{f['average_temperature']:6.1f} {f['experience_temperature']:6.1f} "
|
|
f"{f['water_total']:8.0f}"
|
|
)
|
|
else:
|
|
print(f"{d} {age:3d} — missing —")
|
|
d += timedelta(days=1)
|
|
|
|
lantai1_rows = [
|
|
{"date": str(c), "age": (c - CYCLE_START).days + 1, **by_calendar[c]["fields"]}
|
|
for c in sorted(by_calendar)
|
|
]
|
|
lantai2_rows = [clone_lantai2_panel_row(row) for row in lantai1_rows]
|
|
|
|
out = Path(__file__).resolve().parent / "iot_panel_cycle_extract.json"
|
|
out.write_text(
|
|
json.dumps(
|
|
{
|
|
"cycle_start": str(CYCLE_START),
|
|
"cycle_end": str(CYCLE_END),
|
|
"experience_temperature": (
|
|
"avg_temp - (wind_speed * chill_factor - (humidity-70)/5); "
|
|
"wind_speed is sensor m/s not fan count"
|
|
),
|
|
"flocks": [
|
|
{
|
|
"name": "Lantai 1",
|
|
"iot_flock_id": LANTAI_1_IOT_FLOCK_ID,
|
|
"source": "api",
|
|
"rows": lantai1_rows,
|
|
},
|
|
{
|
|
"name": "Lantai 2",
|
|
"iot_flock_id": LANTAI_2_IOT_FLOCK_ID,
|
|
"source": "cloned_from_lantai_1",
|
|
"adjust": LANTAI_2_ADJUST,
|
|
"rows": lantai2_rows,
|
|
},
|
|
],
|
|
},
|
|
indent=2,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
print(f"\nWrote {out} ({len(lantai1_rows)} Lantai 1 + {len(lantai2_rows)} Lantai 2)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|