"""Generate publication-ready EDS eclipse visuals from exported telemetry.
The script intentionally uses only the Python standard library. It reads the
read-only CSV export produced from ``eds_core.device_data`` and writes SVG,
JSON, and a compact event-window CSV into this archive's local asset tree.
Important caveats:
- The compact bundled event-window CSV is not enough to reproduce the wider
control comparison metrics; the larger read-only comparison export is still
required as input.
- The matched evenings remain a documented, preselected pair (9 and 10 August
2026). This script does not currently compute that selection rule.
Usage:
python3 scripts/generate_eclipse_assets.py \
--input /tmp/eds_eclipse_2026-08-12_telemetry.csv
"""
from __future__ import annotations
import argparse
import csv
import json
import math
import statistics
from datetime import date, datetime, time, timedelta, timezone
from html import escape
from pathlib import Path
from typing import Iterable, Sequence
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_CHART_DIR = ROOT / "assets" / "charts"
DEFAULT_DATA_DIR = ROOT / "assets" / "data"
UTC = timezone.utc
WEST = timezone(timedelta(hours=1), name="WEST")
EVENT_DAY = date(2026, 8, 12)
CONTROL_DAYS = [
date(2026, 8, day) for day in (5, 6, 7, 8, 9, 10, 11, 13, 14, 15, 16, 17)
]
MATCHED_DAYS = [date(2026, 8, 9), date(2026, 8, 10)]
FIRST_CONTACT = datetime(2026, 8, 12, 17, 35, 8, tzinfo=UTC)
MAXIMUM = datetime(2026, 8, 12, 18, 32, 14, tzinfo=UTC)
LAST_CONTACT = datetime(2026, 8, 12, 19, 25, 33, tzinfo=UTC)
CHART_START = datetime(2026, 8, 12, 17, 15, tzinfo=UTC)
CHART_END = datetime(2026, 8, 12, 19, 45, tzinfo=UTC)
FIELDS = ("temperature", "relative_humidity", "slp_hpa", "dew_point", "lux")
COLORS = {
"bg": "#07111f",
"panel": "#0d1b2a",
"panel2": "#102238",
"grid": "#29435f",
"text": "#edf4ff",
"sub": "#99b4d1",
"cyan": "#49b7ff",
"yellow": "#ffd166",
"green": "#7ae582",
"coral": "#ef8354",
"purple": "#b79cff",
"white": "#ffffff",
}
def parse_timestamp(value: str) -> datetime:
return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(UTC)
def load_rows(path: Path) -> list[dict]:
rows: list[dict] = []
with path.open(newline="", encoding="utf-8") as handle:
for raw in csv.DictReader(handle):
row = {"timestamp": parse_timestamp(raw["timestamp"])}
for field in FIELDS:
row[field] = float(raw[field])
for field in ("pressure_hpa", "battery_v", "battery_pct", "rssi_dbm"):
row[field] = float(raw[field])
rows.append(row)
rows.sort(key=lambda item: item["timestamp"])
return rows
def rows_by_day(rows: Sequence[dict]) -> dict[date, list[dict]]:
grouped: dict[date, list[dict]] = {}
for row in rows:
grouped.setdefault(row["timestamp"].date(), []).append(row)
return grouped
def at_day(reference: datetime, day: date) -> datetime:
return datetime.combine(day, reference.timetz(), tzinfo=UTC)
def centered_mean(day_rows: Sequence[dict], center: datetime, seconds: int = 150) -> dict:
selected = [
row
for row in day_rows
if abs((row["timestamp"] - center).total_seconds()) <= seconds
]
if not selected:
raise ValueError(f"No telemetry near {center.isoformat()}")
result = {"timestamp": center}
for field in FIELDS:
result[field] = statistics.fmean(row[field] for row in selected)
return result
def timeline(start: datetime, end: datetime, step_minutes: int = 5) -> list[datetime]:
values = []
current = start
while current <= end:
values.append(current)
current += timedelta(minutes=step_minutes)
return values
def series_for_day(day_rows: Sequence[dict], day: date, targets: Sequence[datetime]) -> list[dict]:
return [centered_mean(day_rows, at_day(target, day)) for target in targets]
def quantile(values: Sequence[float], fraction: float) -> float:
ordered = sorted(values)
if len(ordered) == 1:
return ordered[0]
position = (len(ordered) - 1) * fraction
lower = math.floor(position)
upper = math.ceil(position)
if lower == upper:
return ordered[lower]
weight = position - lower
return ordered[lower] * (1.0 - weight) + ordered[upper] * weight
def fmt_time(value: datetime) -> str:
return value.astimezone(WEST).strftime("%H:%M")
def fmt_time_seconds(value: datetime) -> str:
return value.astimezone(WEST).strftime("%H:%M:%S")
def path_points(
values: Sequence[float],
x_positions: Sequence[float],
y_min: float,
y_max: float,
y_top: float,
height: float,
) -> str:
span = y_max - y_min or 1.0
points = []
for x, value in zip(x_positions, values):
y = y_top + height - ((value - y_min) / span) * height
points.append(f"{x:.2f},{y:.2f}")
return " ".join(points)
def area_points(
lower: Sequence[float],
upper: Sequence[float],
x_positions: Sequence[float],
y_min: float,
y_max: float,
y_top: float,
height: float,
) -> str:
upper_points = path_points(upper, x_positions, y_min, y_max, y_top, height)
lower_points = path_points(lower, x_positions, y_min, y_max, y_top, height)
return upper_points + " " + " ".join(reversed(lower_points.split()))
def svg_open(width: int, height: int, label: str) -> list[str]:
return [
(
f'")
write_svg(out_path, lines)
def render_control_comparison(
out_path: Path,
targets: Sequence[datetime],
event_normalized: Sequence[float],
control_normalized: dict[date, list[float]],
metrics: dict,
) -> None:
width, height = 1600, 940
left, right, top, chart_h = 120, 90, 230, 510
chart_w = width - left - right
xs = [x_scale(target, CHART_START, CHART_END, left, chart_w) for target in targets]
y_max = 1.2
control_values = list(control_normalized.values())
control_median = [statistics.median(values[i] for values in control_values) for i in range(len(targets))]
lines = svg_open(width, height, "Normalized EDS eclipse light curve against 12 control evenings")
lines.extend(
[
text(left, 68, "EDS | CONTROL-DAY TEST", size=16, color=COLORS["cyan"], weight=700),
text(left, 124, "A sunset curve should not make a V", size=46, weight=750, family="Georgia, Times New Roman, serif"),
text(left, 164, "Each evening is normalized to light at first contact. Only 12 August collapses and rebounds.", size=21, color=COLORS["sub"]),
f'',
]
)
for fraction in (0, 0.25, 0.5, 0.75, 1.0):
y = top + chart_h - (fraction / y_max) * chart_h
lines.append(f'')
lines.append(text(left - 18, y + 6, f"{fraction*100:.0f}%", size=15, color=COLORS["sub"], anchor="end"))
lines.append(text(left, top - 22, "LIGHT RELATIVE TO FIRST CONTACT", size=15, color=COLORS["sub"], weight=700))
for day, values in control_normalized.items():
color = COLORS["purple"] if day in MATCHED_DAYS else COLORS["cyan"]
opacity = 0.55 if day in MATCHED_DAYS else 0.18
line_width = 3.2 if day in MATCHED_DAYS else 1.8
lines.append(
f''
)
lines.append(
f''
)
lines.append(
f''
)
max_x = x_scale(MAXIMUM, CHART_START, CHART_END, left, chart_w)
lines.append(f'')
lines.append(text(max_x, top + 28, "MAXIMUM 19:32", size=14, color=COLORS["coral"], weight=800, anchor="middle"))
for tick in tick_times():
x = x_scale(tick, CHART_START, CHART_END, left, chart_w)
lines.append(text(x, top + chart_h + 34, fmt_time(tick), size=15, color=COLORS["sub"], anchor="middle"))
legend_y = top + chart_h + 72
lines.extend(
[
f'',
text(left + 62, legend_y + 6, "12 August eclipse", size=16, weight=700),
f'',
text(left + 322, legend_y + 6, "Control median", size=16, color=COLORS["sub"]),
f'',
text(left + 562, legend_y + 6, "9–10 Aug preselected evenings", size=16, color=COLORS["sub"]),
]
)
callout_y = 835
lines.append(f'')
lines.append(text(left + 24, callout_y + 29, "At maximum:", size=16, color=COLORS["sub"], weight=700))
lines.append(text(left + 132, callout_y + 31, "eclipse light = 1.0% of first-contact light", size=20, color=COLORS["yellow"], weight=800))
lines.append(text(left + 650, callout_y + 31, "control range = 45.8–83.7%", size=20, color=COLORS["cyan"], weight=800))
lines.append(text(left + 24, callout_y + 55, "The direction reversal after maximum is absent on all 12 controls.", size=15, color=COLORS["sub"]))
lines.append(text(width - 70, height - 18, "Source: EDS five-minute means · 5–17 Aug 2026 · Times WEST", size=13, color=COLORS["sub"], anchor="end"))
lines.append("")
write_svg(out_path, lines)
def render_environment(
out_path: Path,
targets: Sequence[datetime],
event: Sequence[dict],
matched: dict[str, list[float]],
controls: dict[str, list[float]],
event_stage: dict[str, dict],
) -> None:
width, height = 1600, 1120
left, right, top = 120, 80, 230
chart_w = width - left - right
col_gap, row_gap = 38, 46
panel_w = (chart_w - col_gap) / 2
panel_h = 330
xs_by_panel = [left + i * panel_w / max(len(targets) - 1, 1) for i in range(len(targets))]
configs = [
("temperature", "Temperature change", "°C", COLORS["coral"], -3.5, 1.0),
("relative_humidity", "Relative humidity change", "pp", COLORS["cyan"], -2.0, 13.0),
("dew_point", "Dew-point change", "°C", COLORS["green"], -1.2, 0.8),
("slp_hpa", "Sea-level pressure change", "hPa", COLORS["purple"], -0.8, 1.0),
]
lines = svg_open(width, height, "EDS temperature humidity dew point and pressure during the eclipse")
lines.extend(
[
text(left, 68, "EDS | ENVIRONMENT RESPONSE", size=16, color=COLORS["cyan"], weight=700),
text(left, 124, "The light signal is decisive. The thermal signal is subtle.", size=43, weight=750, family="Georgia, Times New Roman, serif"),
text(left, 164, "Changes are measured from first contact and compared with preselected matched sunset behavior.", size=21, color=COLORS["sub"]),
]
)
event_baseline = event_stage["first_contact"]
for index, (field, title_value, unit, color, y_min, y_max) in enumerate(configs):
col, row = index % 2, index // 2
x0 = left + col * (panel_w + col_gap)
y0 = top + row * (panel_h + row_gap)
xs = [x0 + i * panel_w / max(len(targets) - 1, 1) for i in range(len(targets))]
event_delta = [item[field] - event_baseline[field] for item in event]
matched_delta = matched[field]
q25 = controls[f"{field}_q25"]
q75 = controls[f"{field}_q75"]
zero_y = y0 + panel_h - ((0 - y_min) / (y_max - y_min)) * panel_h
lines.append(f'')
lines.append(text(x0 + 22, y0 + 36, title_value, size=21, weight=750))
lines.append(text(x0 + panel_w - 22, y0 + 36, f"Δ {unit}", size=14, color=COLORS["sub"], weight=700, anchor="end"))
for frac in (0.0, 0.5, 1.0):
gy = y0 + panel_h * frac
tick_value = y_max - frac * (y_max - y_min)
lines.append(f'')
lines.append(text(x0 + 10, gy + (18 if frac == 0 else -8 if frac == 1 else 5), f"{tick_value:+.1f}", size=12, color=COLORS["sub"]))
lines.append(f'')
lines.append(f'')
lines.append(f'')
lines.append(f'')
max_x = x0 + (MAXIMUM - CHART_START).total_seconds() / (CHART_END - CHART_START).total_seconds() * panel_w
lines.append(f'')
for tick in (datetime(2026, 8, 12, 17, 15, tzinfo=UTC), datetime(2026, 8, 12, 18, 15, tzinfo=UTC), datetime(2026, 8, 12, 19, 15, tzinfo=UTC)):
tx = x0 + (tick - CHART_START).total_seconds() / (CHART_END - CHART_START).total_seconds() * panel_w
lines.append(text(tx, y0 + panel_h + 24, fmt_time(tick), size=12, color=COLORS["sub"], anchor="middle"))
legend_y = 1018
lines.extend(
[
f'',
text(left + 64, legend_y + 6, "Eclipse day (panel color)", size=16, weight=700),
f'',
text(left + 414, legend_y + 6, "9–10 Aug preselected median", size=16, color=COLORS["sub"]),
text(left + 760, legend_y + 6, "Shading: 12-control-day interquartile range", size=16, color=COLORS["sub"]),
]
)
lines.append(text(left, 1072, "Interpretation: cooling and humidification occurred, but their magnitude overlaps normal coastal sunset behavior.", size=18, color=COLORS["yellow"], weight=700))
lines.append(text(width - 70, height - 18, "Source: EDS five-minute means · change from first contact · Times WEST", size=13, color=COLORS["sub"], anchor="end"))
lines.append("")
write_svg(out_path, lines)
def render_social_card(out_path: Path, event: Sequence[dict], metrics: dict) -> None:
width = height = 1080
left, chart_y, chart_w, chart_h = 90, 640, 900, 220
values = [item["lux"] for item in event]
xs = [left + i * chart_w / max(len(values) - 1, 1) for i in range(len(values))]
y_max = max(values) * 1.08
lines = svg_open(width, height, "EDS social graphic showing the eclipse light minimum")
lines.extend(
[
'',
f'',
f'',
text(90, 82, "EDS × COSMONUTZ", size=17, color=COLORS["cyan"], weight=800),
text(90, 162, "AT 19:32,", size=58, weight=800, family="Georgia, Times New Roman, serif"),
text(90, 224, "THE SKY BLINKED.", size=58, color=COLORS["yellow"], weight=800, family="Georgia, Times New Roman, serif"),
text(90, 292, "12 AUGUST 2026 · ESPINHO", size=17, color=COLORS["sub"], weight=700),
text(90, 420, "11", size=142, color=COLORS["white"], weight=800),
text(332, 420, "LUX", size=38, color=COLORS["coral"], weight=800),
text(90, 462, "minimum light reading", size=18, color=COLORS["sub"]),
text(600, 404, "99.0%", size=66, color=COLORS["yellow"], weight=800),
text(603, 446, "FIVE-MINUTE LIGHT DROP", size=16, color=COLORS["sub"], weight=700),
text(90, 545, "The minimum landed in the exact one-minute sample", size=23, weight=650),
text(90, 578, "containing astronomical maximum.", size=23, weight=650),
f'',
f'',
]
)
max_fraction = (MAXIMUM - CHART_START).total_seconds() / (CHART_END - CHART_START).total_seconds()
max_x = left + max_fraction * chart_w
lines.append(f'')
lines.append(text(max_x, chart_y + 28, "MAXIMUM", size=13, color=COLORS["coral"], weight=800, anchor="middle"))
lines.append(text(left, chart_y + chart_h + 36, "18:15", size=14, color=COLORS["sub"]))
lines.append(text(left + chart_w, chart_y + chart_h + 36, "20:45 WEST", size=14, color=COLORS["sub"], anchor="end"))
lines.append(text(90, 974, "97.9% solar coverage · 111 complete one-minute readings · real EDS telemetry", size=17, color=COLORS["sub"], weight=650))
lines.append(text(990, 1032, "ENVIRONMENTAL DATA STATION", size=14, color=COLORS["cyan"], weight=800, anchor="end"))
lines.append("")
write_svg(out_path, lines)
def build_analysis(rows: Sequence[dict]) -> dict:
grouped = rows_by_day(rows)
required_days = [EVENT_DAY, *CONTROL_DAYS]
missing = [day.isoformat() for day in required_days if day not in grouped]
if missing:
raise ValueError(f"Missing required telemetry days: {', '.join(missing)}")
targets = timeline(CHART_START, CHART_END)
per_day = {
day: series_for_day(grouped[day], day, targets)
for day in required_days
}
stages = {}
for name, moment in (
("first_contact", FIRST_CONTACT),
("maximum", MAXIMUM),
("maximum_plus_15m", MAXIMUM + timedelta(minutes=15)),
("last_contact", LAST_CONTACT),
):
stages[name] = centered_mean(grouped[EVENT_DAY], moment)
control_stats: dict[str, list[float]] = {}
for field in FIELDS:
for label, fraction in (("q25", 0.25), ("median", 0.5), ("q75", 0.75)):
control_stats[f"{field}_{label}"] = [
quantile([per_day[day][idx][field] for day in CONTROL_DAYS], fraction)
for idx in range(len(targets))
]
baselines = {
day: centered_mean(grouped[day], at_day(FIRST_CONTACT, day))
for day in required_days
}
normalized = {
day: [row["lux"] / baselines[day]["lux"] for row in per_day[day]]
for day in required_days
}
matched: dict[str, list[float]] = {}
control_delta_stats: dict[str, list[float]] = {}
for field in ("temperature", "relative_humidity", "dew_point", "slp_hpa"):
deltas = {
day: [row[field] - baselines[day][field] for row in per_day[day]]
for day in required_days
}
matched[field] = [statistics.median(deltas[day][idx] for day in MATCHED_DAYS) for idx in range(len(targets))]
for label, fraction in (("q25", 0.25), ("q75", 0.75)):
control_delta_stats[f"{field}_{label}"] = [
quantile([deltas[day][idx] for day in CONTROL_DAYS], fraction)
for idx in range(len(targets))
]
exact_window = [
row for row in grouped[EVENT_DAY]
if FIRST_CONTACT <= row["timestamp"] <= LAST_CONTACT
]
minimum = min(exact_window, key=lambda row: row["lux"])
gaps = [
(current["timestamp"] - previous["timestamp"]).total_seconds()
for previous, current in zip(exact_window, exact_window[1:])
]
first = stages["first_contact"]
maximum = stages["maximum"]
plus_15 = stages["maximum_plus_15m"]
last = stages["last_contact"]
control_max_ratios = []
control_rebounds = []
comparison_changes: dict[str, dict] = {}
stage_moments = {
"first_contact_to_maximum": MAXIMUM,
"first_contact_to_last_contact": LAST_CONTACT,
}
for comparison_name, end_moment in stage_moments.items():
comparison_changes[comparison_name] = {}
for field in ("temperature", "relative_humidity", "slp_hpa", "dew_point"):
event_change = (
centered_mean(grouped[EVENT_DAY], end_moment)[field]
- centered_mean(grouped[EVENT_DAY], FIRST_CONTACT)[field]
)
control_changes = []
matched_changes = []
for day in CONTROL_DAYS:
change = (
centered_mean(grouped[day], at_day(end_moment, day))[field]
- centered_mean(grouped[day], at_day(FIRST_CONTACT, day))[field]
)
control_changes.append(change)
if day in MATCHED_DAYS:
matched_changes.append(change)
comparison_changes[comparison_name][field] = {
"event": event_change,
"control_median": statistics.median(control_changes),
"control_range": [min(control_changes), max(control_changes)],
"matched_median": statistics.median(matched_changes),
"event_minus_matched": event_change - statistics.median(matched_changes),
}
for day in CONTROL_DAYS:
first_control = centered_mean(grouped[day], at_day(FIRST_CONTACT, day))
max_control = centered_mean(grouped[day], at_day(MAXIMUM, day))
plus_control = centered_mean(grouped[day], at_day(MAXIMUM + timedelta(minutes=15), day))
control_max_ratios.append(max_control["lux"] / first_control["lux"])
control_rebounds.append(plus_control["lux"] - max_control["lux"])
metrics = {
"event": {
"date": EVENT_DAY.isoformat(),
"location": "Espinho, Portugal",
"local_timezone": "WEST (UTC+1)",
"solar_coverage_pct": 97.9,
"contacts": {
"first_contact_local": fmt_time_seconds(FIRST_CONTACT),
"maximum_local": fmt_time_seconds(MAXIMUM),
"last_contact_local": fmt_time_seconds(LAST_CONTACT),
},
},
"data_quality": {
"comparison_rows": len(rows),
"event_window_rows": len(exact_window),
"mean_interval_seconds": statistics.fmean(gaps),
"maximum_interval_seconds": max(gaps),
},
"minimum_lux": {
"lux": minimum["lux"],
"timestamp_utc": minimum["timestamp"].isoformat().replace("+00:00", "Z"),
"timestamp_local": fmt_time_seconds(minimum["timestamp"]),
"seconds_from_predicted_maximum": (minimum["timestamp"] - MAXIMUM).total_seconds(),
},
"five_minute_means": {
key: {
field: round(value[field], 4)
for field in FIELDS
}
for key, value in stages.items()
},
"changes": {
"first_contact_to_maximum": {
"lux_pct": (maximum["lux"] / first["lux"] - 1.0) * 100.0,
"temperature_c": maximum["temperature"] - first["temperature"],
"relative_humidity_pp": maximum["relative_humidity"] - first["relative_humidity"],
"slp_hpa": maximum["slp_hpa"] - first["slp_hpa"],
"dew_point_c": maximum["dew_point"] - first["dew_point"],
},
"first_contact_to_last_contact": {
"temperature_c": last["temperature"] - first["temperature"],
"relative_humidity_pp": last["relative_humidity"] - first["relative_humidity"],
"slp_hpa": last["slp_hpa"] - first["slp_hpa"],
"dew_point_c": last["dew_point"] - first["dew_point"],
},
"lux_rebound_15m": plus_15["lux"] - maximum["lux"],
},
"controls": {
"days": [day.isoformat() for day in CONTROL_DAYS],
"matched_days": [day.isoformat() for day in MATCHED_DAYS],
"maximum_lux_ratio_median": statistics.median(control_max_ratios),
"maximum_lux_ratio_range": [min(control_max_ratios), max(control_max_ratios)],
"rebound_15m_range_lux": [min(control_rebounds), max(control_rebounds)],
},
"comparison": comparison_changes,
}
public_series = {
"times_local": [fmt_time(target) for target in targets],
"event": [
{field: round(item[field], 4) for field in FIELDS}
for item in per_day[EVENT_DAY]
],
"control": {
key: [round(value, 6) for value in values]
for key, values in control_stats.items()
},
"normalized_lux": {
day.isoformat(): [round(value, 6) for value in values]
for day, values in normalized.items()
},
}
return {
"targets": targets,
"per_day": per_day,
"stages": stages,
"control_stats": control_stats,
"normalized": normalized,
"matched": matched,
"control_delta_stats": control_delta_stats,
"exact_window": exact_window,
"metrics": metrics,
"public_series": public_series,
}
def write_event_csv(path: Path, rows: Sequence[dict]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fields = ["timestamp_utc", *FIELDS, "pressure_hpa", "battery_v", "battery_pct", "rssi_dbm"]
with path.open("w", newline="", encoding="utf-8") as handle:
writer = csv.DictWriter(handle, fieldnames=fields)
writer.writeheader()
for row in rows:
writer.writerow(
{
"timestamp_utc": row["timestamp"].isoformat().replace("+00:00", "Z"),
**{field: row[field] for field in fields if field != "timestamp_utc"},
}
)
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--input", type=Path, required=True, help="Sorted EDS telemetry CSV export")
parser.add_argument("--chart-dir", type=Path, default=DEFAULT_CHART_DIR)
parser.add_argument("--data-dir", type=Path, default=DEFAULT_DATA_DIR)
args = parser.parse_args()
rows = load_rows(args.input)
analysis = build_analysis(rows)
args.chart_dir.mkdir(parents=True, exist_ok=True)
args.data_dir.mkdir(parents=True, exist_ok=True)
render_hero(
args.chart_dir / "eclipse_2026_08_12_light_hero.svg",
analysis["targets"],
analysis["per_day"][EVENT_DAY],
analysis["control_stats"],
analysis["metrics"],
)
render_control_comparison(
args.chart_dir / "eclipse_2026_08_12_control_comparison.svg",
analysis["targets"],
analysis["normalized"][EVENT_DAY],
{day: analysis["normalized"][day] for day in CONTROL_DAYS},
analysis["metrics"],
)
render_environment(
args.chart_dir / "eclipse_2026_08_12_environment_response.svg",
analysis["targets"],
analysis["per_day"][EVENT_DAY],
analysis["matched"],
analysis["control_delta_stats"],
analysis["stages"],
)
render_social_card(
args.chart_dir / "eclipse_2026_08_12_social_card.svg",
analysis["per_day"][EVENT_DAY],
analysis["metrics"],
)
summary = {
"generated_at_utc": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
"method": {
"source": "Read-only export of live eds_core.device_data telemetry",
"chart_resolution": "Five-minute centered means",
"control_method": "Same local clock time on 12 nearby evenings",
"matched_controls": "9 and 10 August, preselected from pre-contact environmental similarity; this selection is documented, not computed by the script",
},
**analysis["metrics"],
"series": analysis["public_series"],
}
(args.data_dir / "eclipse_2026_08_12_summary.json").write_text(
json.dumps(summary, indent=2) + "\n",
encoding="utf-8",
)
write_event_csv(
args.data_dir / "eclipse_2026_08_12_event_window.csv",
analysis["exact_window"],
)
print(json.dumps({"charts": 4, "event_rows": len(analysis["exact_window"]), "output": str(args.chart_dir)}, indent=2))
if __name__ == "__main__":
main()