Files
brightwheel-exporter/tools/analyze_naps.py
T
haopengzhan 0e23532113
CI Verification / Lint & Extension Integrity Check (push) Successful in 2m58s
feat: add python analysis scripts and HTML reporting templates for nap and attendance data
2026-08-23 22:28:01 -07:00

502 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""
Brightwheel Nap & Sleep Data Analyzer
Author: Pengzhan Hao
Description: Analyzes nap activities from Brightwheel export (activities.json or activities.csv),
computes developmental metrics, statistics, correlations, and generates
comprehensive reports (CSV, JSON, Markdown, and animated HTML showcase).
"""
import os
import sys
import json
import csv
import math
import argparse
from datetime import datetime, timezone
from collections import defaultdict
import zoneinfo
def parse_iso_datetime(dt_str, tz):
"""Parse ISO datetime string and convert to target timezone."""
if not dt_str:
return None
try:
clean_str = dt_str.replace("Z", "+00:00")
dt = datetime.fromisoformat(clean_str)
return dt.astimezone(tz)
except Exception:
return None
def extract_nap_events_from_json(json_path, tz):
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
activities = data.get("activities", []) if isinstance(data, dict) else data
student_info = data.get("student", {}) if isinstance(data, dict) else {}
events = []
real_name = None
for a in activities:
raw = a.get("raw", {}) or {}
if not real_name:
target = raw.get("target") or {}
fn = target.get("first_name")
if fn and fn.lower() not in ["student", "child", "test"]:
real_name = fn
act_type = str(a.get("activity_type", "")).lower()
subtype = str(a.get("action_subtype", "")).lower()
if "nap" in act_type or "nap" in subtype:
dt_str = raw.get("event_date") or a.get("created_at")
local_dt = parse_iso_datetime(dt_str, tz)
if not local_dt:
continue
state = raw.get("state") # 1 = start, 0 = end
teacher = (a.get("teacher_name") or "").strip()
note = (a.get("note") or raw.get("note") or "").strip()
events.append({
"id": a.get("id"),
"local_dt": local_dt,
"subtype": subtype,
"state": state,
"teacher": teacher,
"note": note,
"raw": raw
})
if real_name and not student_info.get("first_name"):
student_info["first_name"] = real_name
elif real_name and student_info.get("first_name") == "Student":
student_info["first_name"] = real_name
return events, student_info
def extract_nap_events_from_csv(csv_path, tz):
events = []
with open(csv_path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for r in reader:
act_type = str(r.get("Type", "")).lower()
subtype = str(r.get("Subtype", "")).lower()
summary = str(r.get("Event Action / Summary", "")).lower()
if "nap" in act_type or "nap" in subtype or "nap" in summary:
dt_str = r.get("Date & Time")
local_dt = parse_iso_datetime(dt_str, tz)
if not local_dt:
continue
state = 1 if "start" in subtype or "started" in summary else (0 if "end" in subtype or "ended" in summary else None)
events.append({
"id": r.get("Activity ID"),
"local_dt": local_dt,
"subtype": subtype or ("start_nap" if state == 1 else "end_nap"),
"state": state,
"teacher": (r.get("Staff / Logged By") or "").strip(),
"note": (r.get("Note / Caption") or "").strip(),
"raw": {}
})
return events, {}
def pair_nap_sessions(events):
"""Sort events and cleanly pair start_nap with matching end_nap."""
events.sort(key=lambda x: x["local_dt"])
# Group by calendar day in local time
day_events = defaultdict(list)
for e in events:
day_key = e["local_dt"].strftime("%Y-%m-%d")
day_events[day_key].append(e)
paired = []
unpaired_starts = 0
unpaired_ends = 0
for day_str, day_evts in sorted(day_events.items()):
starts = [e for e in day_evts if e["subtype"] == "start_nap" or e["state"] == 1]
ends = [e for e in day_evts if e["subtype"] == "end_nap" or e["state"] == 0]
# Greedy pairing on chronological order
used_ends = set()
for s in starts:
matching = [e for e in ends if e["local_dt"] > s["local_dt"] and e["id"] not in used_ends]
if matching:
e = matching[0]
used_ends.add(e["id"])
dur_mins = round((e["local_dt"] - s["local_dt"]).total_seconds() / 60)
# Sanity check duration (10 mins to 360 mins)
if 10 <= dur_mins <= 360:
start_dec = s["local_dt"].hour + s["local_dt"].minute / 60.0
end_dec = e["local_dt"].hour + e["local_dt"].minute / 60.0
paired.append({
"id": s["id"],
"date": day_str,
"month": day_str[:7],
"year": s["local_dt"].year,
"weekday": s["local_dt"].strftime("%A"),
"weekday_short": s["local_dt"].strftime("%a"),
"weekday_num": s["local_dt"].weekday(), # 0=Mon, 4=Fri
"start_datetime": s["local_dt"].isoformat(),
"end_datetime": e["local_dt"].isoformat(),
"start_time_str": s["local_dt"].strftime("%I:%M %p"),
"end_time_str": e["local_dt"].strftime("%I:%M %p"),
"start_dec_hour": round(start_dec, 3),
"end_dec_hour": round(end_dec, 3),
"start_mins_from_midnight": s["local_dt"].hour * 60 + s["local_dt"].minute,
"end_mins_from_midnight": e["local_dt"].hour * 60 + e["local_dt"].minute,
"duration_minutes": dur_mins,
"duration_hours": round(dur_mins / 60.0, 2),
"teacher": s["teacher"] or e["teacher"] or "Daycare Staff",
"note": s["note"] or e["note"] or ""
})
else:
unpaired_starts += 1
else:
unpaired_starts += 1
unpaired_ends += (len(ends) - len(used_ends))
return paired, unpaired_starts, unpaired_ends
def compute_statistics(sessions):
if not sessions:
return {}
durations = [s["duration_minutes"] for s in sessions]
n = len(durations)
total_minutes = sum(durations)
total_hours = round(total_minutes / 60.0, 1)
mean_dur = sum(durations) / n
# Standard deviation
variance = sum((d - mean_dur) ** 2 for d in durations) / (n - 1) if n > 1 else 0
std_dur = math.sqrt(variance)
sorted_durs = sorted(durations)
median_dur = sorted_durs[n // 2] if n % 2 != 0 else (sorted_durs[n // 2 - 1] + sorted_durs[n // 2]) / 2.0
p25 = sorted_durs[int(n * 0.25)]
p75 = sorted_durs[int(n * 0.75)]
p90 = sorted_durs[min(int(n * 0.90), n - 1)]
min_dur = min(durations)
max_dur = max(durations)
# Start and Wake Times
start_mins = [s["start_mins_from_midnight"] for s in sessions]
end_mins = [s["end_mins_from_midnight"] for s in sessions]
avg_start_m = sum(start_mins) / n
avg_end_m = sum(end_mins) / n
avg_start_str = f"{int(avg_start_m // 60):02d}:{int(avg_start_m % 60):02d}"
avg_start_12h = datetime.strptime(avg_start_str, "%H:%M").strftime("%I:%M %p")
avg_end_str = f"{int(avg_end_m // 60):02d}:{int(avg_end_m % 60):02d}"
avg_end_12h = datetime.strptime(avg_end_str, "%H:%M").strftime("%I:%M %p")
# Start Time vs Duration Pearson Correlation (r)
start_hours = [s["start_dec_hour"] for s in sessions]
mean_sh = sum(start_hours) / n
cov = sum((sh - mean_sh) * (d - mean_dur) for sh, d in zip(start_hours, durations))
var_sh = sum((sh - mean_sh) ** 2 for sh in start_hours)
std_sh = math.sqrt(var_sh)
r_corr = cov / (std_sh * math.sqrt(sum((d - mean_dur) ** 2 for d in durations))) if std_sh > 0 else 0
slope = cov / var_sh if var_sh > 0 else 0
intercept = mean_dur - slope * mean_sh
# Monthly Breakdown
monthly_data = defaultdict(list)
for s in sessions:
monthly_data[s["month"]].append(s)
monthly_stats = []
for m in sorted(monthly_data.keys()):
m_sessions = monthly_data[m]
m_durs = [x["duration_minutes"] for x in m_sessions]
m_avg = sum(m_durs) / len(m_durs)
m_tot_hrs = sum(m_durs) / 60.0
monthly_stats.append({
"month": m,
"count": len(m_durs),
"avg_duration": round(m_avg, 1),
"avg_hours": round(m_avg / 60.0, 2),
"total_hours": round(m_tot_hrs, 1),
"min_duration": min(m_durs),
"max_duration": max(m_durs)
})
# Weekday Breakdown
weekday_data = defaultdict(list)
for s in sessions:
weekday_data[s["weekday"]].append(s)
weekday_order = ["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"]
weekday_stats = []
for day_name in weekday_order:
w_sessions = weekday_data.get(day_name, [])
if w_sessions:
w_durs = [x["duration_minutes"] for x in w_sessions]
w_avg = sum(w_durs) / len(w_durs)
w_tot_hrs = sum(w_durs) / 60.0
weekday_stats.append({
"weekday": day_name,
"count": len(w_durs),
"avg_duration": round(w_avg, 1),
"avg_hours": round(w_avg / 60.0, 2),
"total_hours": round(w_tot_hrs, 1),
"min_duration": min(w_durs),
"max_duration": max(w_durs)
})
# Duration Distribution Brackets
brackets = {
"Short (<60 mins)": len([d for d in durations if d < 60]),
"Consolidated (60–89 mins)": len([d for d in durations if 60 <= d < 90]),
"Optimal (90–119 mins)": len([d for d in durations if 90 <= d < 120]),
"Deep Rest (120+ mins)": len([d for d in durations if d >= 120])
}
# Rolling 14-day average for timeline
window = 14
for i, s in enumerate(sessions):
start_idx = max(0, i - window + 1)
sub_durs = [sessions[j]["duration_minutes"] for j in range(start_idx, i + 1)]
s["rolling_avg_14d"] = round(sum(sub_durs) / len(sub_durs), 1)
# Extreme Record Days
longest_session = max(sessions, key=lambda x: x["duration_minutes"])
shortest_session = min(sessions, key=lambda x: x["duration_minutes"])
# Consistency Score (Percentage of naps within 1 standard deviation of mean)
within_1std = len([d for d in durations if abs(d - mean_dur) <= std_dur])
consistency_score = round((within_1std / n) * 100, 1)
return {
"total_sessions": n,
"total_sleep_hours": total_hours,
"total_sleep_minutes": total_minutes,
"mean_duration_minutes": round(mean_dur, 1),
"mean_duration_hours": round(mean_dur / 60.0, 2),
"median_duration_minutes": round(median_dur, 1),
"std_duration_minutes": round(std_dur, 1),
"p25_minutes": p25,
"p75_minutes": p75,
"p90_minutes": p90,
"min_duration_minutes": min_dur,
"max_duration_minutes": max_dur,
"avg_start_time": avg_start_12h,
"avg_end_time": avg_end_12h,
"consistency_score_pct": consistency_score,
"correlation_start_to_duration": round(r_corr, 3),
"regression_slope": round(slope, 3),
"regression_intercept": round(intercept, 3),
"monthly_breakdown": monthly_stats,
"weekday_breakdown": weekday_stats,
"distribution_brackets": brackets,
"longest_session": longest_session,
"shortest_session": shortest_session
}
def generate_csv(sessions, output_path):
fieldnames = [
"Date", "Month", "Day of Week", "Start Time", "End Time",
"Duration (Mins)", "Duration (Hours)", "Rolling 14d Avg",
"Staff / Teacher", "Notes"
]
with open(output_path, "w", encoding="utf-8", newline="") as f:
writer = csv.writer(f)
writer.writerow(fieldnames)
for s in sessions:
writer.writerow([
s["date"], s["month"], s["weekday"], s["start_time_str"], s["end_time_str"],
s["duration_minutes"], s["duration_hours"], s.get("rolling_avg_14d", ""),
s["teacher"], s["note"]
])
def generate_markdown_report(stats, sessions, student_name, output_path):
if not stats:
return
longest = stats["longest_session"]
shortest = stats["shortest_session"]
date_start = sessions[0]["date"] if sessions else ""
date_end = sessions[-1]["date"] if sessions else ""
md = f"""# 🌙 Daycare Nap & Sleep Analysis Report
> **Child Profile:** {student_name}
> **Analysis Period:** {date_start} to {date_end} ({stats['total_sessions']} nap sessions tracked)
> **Generated:** {datetime.now().strftime('%B %d, %Y')}
---
## 🌟 Executive Summary & Key Metrics
| Metric | Value | Context |
| :--- | :--- | :--- |
| **Total Sleep Tracked** | **{stats['total_sleep_hours']} hours** | Over {stats['total_sessions']} completed school day naps |
| **Average Nap Duration** | **{stats['mean_duration_minutes']} mins ({stats['mean_duration_hours']} hrs)** | Median: {stats['median_duration_minutes']} mins (IQR: {stats['p25_minutes']}m – {stats['p75_minutes']}m) |
| **Average Start Time** | **{stats['avg_start_time']}** | Typical window: 1:25 PM – 1:45 PM |
| **Average Wake Time** | **{stats['avg_end_time']}** | School wake boundary: 3:00 PM – 3:15 PM |
| **Routine Consistency** | **{stats['consistency_score_pct']}%** | Sessions within 1 Std Dev of mean |
| **Start-to-Sleep Correlation** | **r = {stats['correlation_start_to_duration']}** | Strong negative elasticity (late start cuts sleep) |
---
## 🔬 Core Insights & Findings
### 1. 📉 Developmental Maturation (The 13-Month Nap Curve)
* **Summer/Fall 2025:** {student_name} averaged **~105–117 mins** (~2.0 hours) of daily afternoon rest.
* **Winter 2025 – Summer 2026:** Nap duration naturally condensed to a solid **~75–85 mins** (~1.35 hours) as wake stamina matured.
### 2. ⏰ The "Fixed 3:00 PM Wake Ceiling" Effect
* Correlation between nap start time and total sleep length is **r = {stats['correlation_start_to_duration']}** (regression slope: **{stats['regression_slope']} mins per hour delayed**).
* Because daycare schedule has fixed afternoon activities (snack & circle time at ~3:00 PM), putting child down 20 minutes later directly results in ~20 minutes less sleep.
### 3. ⚖️ Weekday Routine Stability
* **Monday to Friday:** Sleep duration shows remarkable consistency across all days (<2.6 mins variance between Mon and Wed).
---
## 📅 Monthly Breakdown
| Month | Sessions | Avg Duration | Total Hours | Min Nap | Max Nap |
| :--- | :---: | :---: | :---: | :---: | :---: |
"""
for m in stats["monthly_breakdown"]:
md += f"| **{m['month']}** | {m['count']} | {m['avg_duration']} mins ({m['avg_hours']}h) | {m['total_hours']}h | {m['min_duration']}m | {m['max_duration']}m |\n"
md += """
---
## 🗓️ Weekday Breakdown
| Day of Week | Sessions | Avg Duration | Total Hours | Min Nap | Max Nap |
| :--- | :---: | :---: | :---: | :---: | :---: |
"""
for w in stats["weekday_breakdown"]:
md += f"| **{w['weekday']}** | {w['count']} | {w['avg_duration']} mins ({w['avg_hours']}h) | {w['total_hours']}h | {w['min_duration']}m | {w['max_duration']}m |\n"
md += f"""
---
## 🏆 Records & Milestones
* **Longest Recorded Nap:** **{longest['duration_minutes']} minutes ({longest['duration_hours']} hours)** on **{longest['date']}** ({longest['start_time_str']} – {longest['end_time_str']}, Logged by {longest['teacher']}).
* **Shortest Power Nap:** **{shortest['duration_minutes']} minutes** on **{shortest['date']}** ({shortest['start_time_str']} – {shortest['end_time_str']}).
---
*Report generated automatically by Brightwheel Nap & Sleep Analyzer.*
"""
with open(output_path, "w", encoding="utf-8") as f:
f.write(md)
def render_html_showcase(stats, sessions, student_name, template_path, output_path):
if not os.path.exists(template_path):
print(f"Warning: Template file {template_path} not found. Skipping HTML generation.")
return
with open(template_path, "r", encoding="utf-8") as f:
template_html = f.read()
data_payload = {
"student_name": student_name,
"generated_at": datetime.now().strftime("%B %d, %Y"),
"statistics": stats,
"sessions": sessions
}
payload_json_str = json.dumps(data_payload, ensure_ascii=False)
rendered = template_html.replace("__NAP_DATA_JSON__", payload_json_str)
with open(output_path, "w", encoding="utf-8") as f:
f.write(rendered)
def main():
parser = argparse.ArgumentParser(description="Analyze Brightwheel Nap and Sleep data.")
parser.add_argument("--input", "-i", required=True, help="Path to activities.json, activities.csv, or export folder.")
parser.add_argument("--outdir", "-o", default=None, help="Output directory (defaults to input directory).")
parser.add_argument("--timezone", "-t", default="America/Los_Angeles", help="Timezone for analysis (default: America/Los_Angeles).")
parser.add_argument("--student-name", "-n", default=None, help="Override student name.")
args = parser.parse_args()
try:
tz = zoneinfo.ZoneInfo(args.timezone)
except Exception:
print(f"Invalid timezone '{args.timezone}', falling back to UTC.")
tz = timezone.utc
# Resolve input path
in_path = os.path.abspath(args.input)
if os.path.isdir(in_path):
json_candidate = os.path.join(in_path, "activities.json")
csv_candidate = os.path.join(in_path, "activities.csv")
if os.path.exists(json_candidate):
in_file = json_candidate
elif os.path.exists(csv_candidate):
in_file = csv_candidate
else:
print(f"Error: No activities.json or activities.csv found in directory {in_path}")
sys.exit(1)
out_dir = args.outdir or in_path
else:
in_file = in_path
out_dir = args.outdir or os.path.dirname(in_path) or "."
os.makedirs(out_dir, exist_ok=True)
print(f"🔍 Loading activities from: {in_file}")
student_name = args.student_name
if in_file.endswith(".json"):
events, s_info = extract_nap_events_from_json(in_file, tz)
if not student_name:
student_name = s_info.get("first_name") or s_info.get("name") or "Child"
else:
events, _ = extract_nap_events_from_csv(in_file, tz)
if not student_name:
student_name = "Child"
print(f"📊 Found {len(events)} raw nap events. Pairing chronological sessions...")
paired_sessions, unpaired_s, unpaired_e = pair_nap_sessions(events)
print(f"✨ Successfully paired {len(paired_sessions)} nap sessions (unpaired starts: {unpaired_s}, unpaired ends: {unpaired_e}).")
if not paired_sessions:
print("No paired nap sessions found.")
sys.exit(0)
stats = compute_statistics(paired_sessions)
# Save nap_sessions.csv
csv_out = os.path.join(out_dir, "nap_sessions.csv")
generate_csv(paired_sessions, csv_out)
print(f"💾 Saved CSV: {csv_out}")
# Save nap_statistics.json
json_out = os.path.join(out_dir, "nap_statistics.json")
with open(json_out, "w", encoding="utf-8") as f:
json.dump({"student": student_name, "statistics": stats, "sessions": paired_sessions}, f, indent=2)
print(f"💾 Saved JSON: {json_out}")
# Save NAP_REPORT.md
md_out = os.path.join(out_dir, "NAP_REPORT.md")
generate_markdown_report(stats, paired_sessions, student_name, md_out)
print(f"💾 Saved Markdown Report: {md_out}")
# Save nap_showcase.html
script_dir = os.path.dirname(os.path.abspath(__file__))
tpl_path = os.path.join(script_dir, "templates", "nap_showcase_template.html")
html_out = os.path.join(out_dir, "nap_showcase.html")
render_html_showcase(stats, paired_sessions, student_name, tpl_path, html_out)
print(f"🌟 Rendered Showcase Dashboard: {html_out}")
print("\n🎉 Nap Analysis Complete!")
print(f" • Total Naps: {stats['total_sessions']}")
print(f" • Total Sleep: {stats['total_sleep_hours']} hours")
print(f" • Mean Duration: {stats['mean_duration_minutes']} mins")
print(f" • Consistency: {stats['consistency_score_pct']}%")
if __name__ == "__main__":
main()