"""Independent direct-selector audit plus known-answer fixtures.""" from collections import defaultdict from datetime import date, datetime, timedelta, timezone import copy import gzip import json import math from pathlib import Path import analyze ROOT = Path(__file__).resolve().parents[1] def fixtures(): passed = [] constant = [10] * 48 assert analyze.window_sums(constant) == [60] * 43 passed.append("constant window costs") isolated = [100] * 48 isolated[10:16] = [0] * 6 assert min(range(43), key=lambda x: analyze.window_sums(isolated)[x]) == 10 passed.append("known minimum at 05:00") boundary = [100] * 42 + [0] * 6 assert min(range(43), key=lambda x: analyze.window_sums(boundary)[x]) == 42 passed.append("last candidate ends exactly at midnight") training = {date(2023, 1, 1) + timedelta(days=i): constant[:] for i in range(365)} selected = analyze.learn(training) assert all(s["start_slot"] == 0 for s in selected.values()) passed.append("exact ties choose earliest") training[date(2025, 5, 1)] = isolated[:] assert analyze.learn(training) == selected training[date(2025, 5, 1)] = [100000] * 48 assert analyze.learn(training) == selected passed.append("test values cannot alter learned schedules") first = datetime(2024, 2, 29, tzinfo=timezone.utc) raw = [{"from": (first + timedelta(minutes=30 * i)).isoformat(), "to": (first + timedelta(minutes=30 * (i + 1))).isoformat(), "intensity": {"actual": i}} for i in range(48)] days, _, audit = analyze.ingest(raw + [copy.deepcopy(raw[0])]) assert days[date(2024, 2, 29)] == list(range(48)) and audit["identical_duplicates"] == 1 passed.append("leap day and identical boundary duplicate") missing = raw[:-1] days, exclusions, _ = analyze.ingest(missing) assert not days assert next(x for x in exclusions if x["date"] == "2024-02-29")["missing_slots"] == [47] passed.append("incomplete day excluded without interpolation") invalid = copy.deepcopy(raw) invalid[9]["intensity"]["actual"] = None days, exclusions, _ = analyze.ingest(invalid) assert not days assert next(x for x in exclusions if x["date"] == "2024-02-29")["invalid_slots"] == [9] passed.append("missing actual never replaced by forecast") for bad in (-1, float("nan"), True): invalid[9]["intensity"]["actual"] = bad assert not analyze.ingest(invalid)[0] passed.append("negative nonfinite and boolean values excluded") conflict = copy.deepcopy(raw[0]) conflict["intensity"]["actual"] = 123 try: analyze.ingest(raw + [conflict]) except ValueError: passed.append("contradictory duplicate rejected") else: raise AssertionError("Conflicting duplicates were accepted") bad_time = copy.deepcopy(raw[0]) bad_time["from"] = "2024-02-29T00:01Z" bad_time["to"] = "2024-02-29T00:31Z" try: analyze.ingest([bad_time]) except ValueError: passed.append("misaligned half-hour rejected") else: raise AssertionError("Misaligned interval accepted") try: analyze.learn({date(2023, 1, 1): constant}) except ValueError: passed.append("insufficient training coverage rejected") else: raise AssertionError("Insufficient training allowed") assert all(analyze.season(date(2023, month, 1)) == expected for month, expected in enumerate(["DJF", "DJF", "MAM", "MAM", "MAM", "JJA", "JJA", "JJA", "SON", "SON", "SON", "DJF"], 1)) passed.append("calendar season mapping") return passed def main(): tests = fixtures() declared = json.loads((ROOT / "results/R1.json").read_text()) # Independent parser and complete-day construction, avoiding analyze.load/ingest. flat = {} for source in json.loads((ROOT / "data/SOURCES.json").read_text()): for row in json.loads(gzip.decompress((ROOT / source["path"]).read_bytes()))["data"]: stamp = datetime.fromisoformat(row["from"].replace("Z", "+00:00")).astimezone(timezone.utc) if stamp.year in (2023, 2024, 2025): value = row["intensity"]["actual"] if stamp in flat: assert flat[stamp] == value flat[stamp] = value grouped = defaultdict(list) for stamp, value in flat.items(): grouped[stamp.date()].append((stamp.hour * 60 + stamp.minute, value)) valid = {day: sorted(v) for day, v in grouped.items() if len(v) == 48 and all(type(x) in (int, float) and math.isfinite(x) and x >= 0 for _, x in v)} assert all([minute for minute, _ in v] == list(range(0, 1440, 30)) for v in valid.values()) month_to_season = {1: "DJF", 2: "DJF", 3: "MAM", 4: "MAM", 5: "MAM", 6: "JJA", 7: "JJA", 8: "JJA", 9: "SON", 10: "SON", 11: "SON", 12: "DJF"} selection_checks = 0 for name, expected in declared["selections"].items(): train_days = [day for day in valid if day.year == 2023 and (name == "global" or month_to_season[day.month] == name)] candidate = [] for start in range(0, 1261, 30): values = [value for day in train_days for minute, value in valid[day] if start <= minute < start + 180] assert len(values) == 6 * len(train_days) candidate.append((math.fsum(values) / len(values), start)) minimum, start = min(candidate) assert start == 30 * expected["start_slot"] assert math.isclose(minimum, expected["training_intensity_gco2_per_kwh"], rel_tol=1e-12, abs_tol=1e-12) selection_checks += 1 direct = {} max_error = 0.0 for day, intervals in valid.items(): if day.year == 2023: continue starts = {"midnight": 0, "evening": 1080, "global": declared["selections"]["global"]["start_slot"] * 30, "seasonal": declared["selections"][month_to_season[day.month]]["start_slot"] * 30} costs = {name: math.fsum(value for minute, value in intervals if start <= minute < start + 180) / 6 for name, start in starts.items()} costs["hindsight"] = min(math.fsum(value for minute, value in intervals if start <= minute < start + 180) / 6 for start in range(0, 1261, 30)) assert all(costs["hindsight"] <= cost for cost in costs.values()) direct[day] = costs for year in (2023, 2024, 2025): assert sum(day.year == year for day in valid) == declared["coverage"][str(year)]["complete_days"] aggregate_checks = 0 for year in (2024, 2025): for month in (None,) + tuple(range(1, 13)): costs = [c for day, c in direct.items() if day.year == year and (month is None or day.month == month)] expected = declared["years"][str(year)] if month is not None: expected = expected["monthly"][str(month)] assert len(costs) == expected["days"] if not costs: continue for name in ("midnight", "evening", "global", "seasonal", "hindsight"): value = math.fsum(c[name] for c in costs) / len(costs) error = abs(value - expected["mean_intensity_gco2_per_kwh"][name]) max_error = max(max_error, error) assert error < 1e-10 aggregate_checks += 1 a = math.fsum(c["midnight"] for c in costs) b = math.fsum(c["seasonal"] for c in costs) assert math.isclose(100 * (a - b) / a, expected["seasonal_vs_midnight"]["percent_reduction"], abs_tol=1e-10) counts = {"better": sum(c["seasonal"] < c["midnight"] for c in costs), "equal": sum(c["seasonal"] == c["midnight"] for c in costs), "worse": sum(c["seasonal"] > c["midnight"] for c in costs)} assert counts == expected["day_counts"] result = {"all_passed": True, "synthetic_checks": tests, "synthetic_count": len(tests), "independent_schedule_checks": selection_checks, "independent_aggregate_checks": aggregate_checks, "independent_daily_checks": len(direct), "max_absolute_aggregate_error": max_error, "note": "Independent direct time-range selection from source response records agrees with the prefix-sum analysis; fixtures exercise missingness, ties, boundaries and training separation."} (ROOT / "results/validation.json").write_text(json.dumps(result, indent=2) + "\n") print(json.dumps(result, indent=2)) if __name__ == "__main__": main()