"""Prepare Telescope submissions offline using a downloaded producer definition. Python 3.9+, standard library only. This helper never sends a request. """ import argparse import csv import json import math from collections import defaultdict from datetime import date, datetime, time, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo UTC = timezone.utc def timestamp(value): instant = datetime.fromisoformat(value.replace("Z", "+00:00")) if instant.tzinfo is None: raise ValueError("Timestamps need Z or an explicit UTC offset") return instant.astimezone(UTC) def iso(instant): return instant.astimezone(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") def grid(contract, now, delivery_day=None): """Return one eligible full day or the minimum rolling interval range.""" definition, offering, policy = (contract.get(key) for key in ("definition", "offering", "policy")) if not definition or not offering or not policy: raise ValueError("The definition needs a live offering and policy") zone = ZoneInfo(definition["timezone"]) resolution = definition["resolution"] minutes = {"PT15M": 15, "PT1H": 60, "PT4H": 240}[resolution] today = now.astimezone(zone).date() first_day = today + timedelta(days=offering["firstDeliveryDayOffset"]) last_day = today + timedelta(days=offering["maxHorizonDays"]) def midnight(day): return datetime.combine(day, time(), zone).astimezone(UTC) def day_grid(day): if resolution == "PT4H": return [datetime.combine(day, time(hour), zone).astimezone(UTC) for hour in range(0, 24, 4)] result = [] instant, end = midnight(day), midnight(day + timedelta(days=1)) while instant < end: result.append(instant) instant += timedelta(minutes=minutes) return result update = policy["updatePolicy"] if update["mode"] == "rolling": if delivery_day: raise ValueError("Rolling definitions choose their start from the current time; omit --day") # Keep the current interval, providing transit tolerance without future gaps. candidates = [instant for offset in range(-1, (last_day - today).days + 2) for instant in day_grid(today + timedelta(days=offset))] start = max(instant for instant in candidates if instant <= now) if now - start > timedelta(hours=1): raise ValueError("Current grid start is outside the rolling one-hour tolerance; try near the next boundary") end = start + timedelta(hours=48) if end > midnight(last_day + timedelta(days=1)): raise ValueError("The offering horizon cannot cover the required 48 hours") return [instant for instant in candidates if start <= instant < end] def deadline(day): gate_day = day - timedelta(days=update["gateDaysBeforeDelivery"]) gate = datetime.combine(gate_day, time.fromisoformat(update["gateLocalTime"]), zone) return gate.astimezone(UTC) - timedelta(minutes=update["submissionLeadMinutes"]) if delivery_day: selected = date.fromisoformat(delivery_day) if not first_day <= selected <= last_day or now >= deadline(selected): raise ValueError("Requested day is outside the horizon or its submission deadline has passed") else: selected = next((first_day + timedelta(days=i) for i in range((last_day - first_day).days + 1) if now < deadline(first_day + timedelta(days=i))), None) if selected is None: raise ValueError("No delivery day still has an open submission deadline") return day_grid(selected) def dimensions(contract): return [series["key"] for series in contract["series"]] or ["value"] def write_template(contract, instants, path): with path.open("x", newline="") as handle: writer = csv.writer(handle) writer.writerow(["validTime", "seriesKey", "outputKey", "value"]) for instant in instants: for series in dimensions(contract): for output in contract["outputs"]: writer.writerow([iso(instant), series, output["key"], ""]) def build_payload(contract, rows, now, generated_at): if generated_at > now + timedelta(minutes=5): raise ValueError("Model generation time is more than five minutes in the future") series_keys = dimensions(contract) output_keys = {output["key"] for output in contract["outputs"]} if not output_keys: raise ValueError("The definition has no configured outputs") values = {} for row in rows: key = (timestamp(row["validTime"]), row["seriesKey"], row["outputKey"]) if key in values: raise ValueError(f"Duplicate prediction: {key}") if key[1] not in series_keys or key[2] not in output_keys: raise ValueError(f"Unknown series or output: {key}") value = float(row["value"]) output = next(item for item in contract["outputs"] if item["key"] == key[2]) if not math.isfinite(value) or value < output.get("min", -math.inf) or value > output.get("max", math.inf): raise ValueError(f"Value must be finite and within output bounds: {key}") values[key] = value instants = sorted({key[0] for key in values}) if not instants or len(instants) > 10_000: raise ValueError("Supply between 1 and 10,000 interval points") zone = ZoneInfo(contract["definition"]["timezone"]) delivery_day = instants[0].astimezone(zone).date().isoformat() rolling = contract["policy"]["updatePolicy"]["mode"] == "rolling" if rolling: # Validate against the grid at the CSV's start, then check current tolerance. expected = grid(contract, instants[0]) if not now - timedelta(hours=1) <= instants[0] <= now: raise ValueError("The rolling CSV is stale or starts in the future; generate a fresh grid") grid(contract, now) # Check the live horizon too. else: expected = grid(contract, now, delivery_day) if instants != expected: raise ValueError("CSV must match the helper's complete grid (one gate-closure day or exactly 48 rolling hours)") windows = defaultdict(list) for instant in instants: series = {} for key in series_keys: outputs = {output: values[(instant, key, output)] for output in output_keys} standard_quantiles = {"p10": 0.1, "p50": 0.5, "p90": 0.9} quantiles = sorted((output.get("quantileLevel", standard_quantiles.get(output["key"])), outputs[output["key"]]) for output in contract["outputs"] if "quantileLevel" in output or output["key"] in standard_quantiles) if any(left[1] > right[1] for left, right in zip(quantiles, quantiles[1:])): raise ValueError(f"Quantiles must be ordered for {key} at {iso(instant)}") series[key] = outputs value = {"validTime": iso(instant)} if contract["series"]: value["series"] = series else: value["outputs"] = series["value"] windows[instant.astimezone(zone).date().isoformat()].append(value) return {"definition": contract["definition"]["key"], "generatedAt": iso(generated_at), "deliveryWindows": [{"deliveryDay": day, "values": points} for day, points in sorted(windows.items())]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--definitions", type=Path, required=True) parser.add_argument("--definition", required=True) parser.add_argument("--generated-at", help="Actual model generation time (RFC 3339); required with --values") parser.add_argument("--day", help="Optional local delivery day for a gate-closure template") mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument("--template", type=Path, help="Write a CSV grid with empty prediction cells") mode.add_argument("--values", type=Path, help="Read a CSV containing your model's predictions") parser.add_argument("--output", type=Path, default=Path("submission.json")) args = parser.parse_args() try: contract = next((item for item in json.loads(args.definitions.read_text())["definitions"] if item["definition"]["key"] == args.definition), None) if contract is None: raise ValueError("Definition is not present in the downloaded catalog") now = datetime.now(UTC) if args.template: write_template(contract, grid(contract, now, args.day), args.template) else: if not args.generated_at: raise ValueError("Supply --generated-at with the actual model generation time") with args.values.open(newline="") as handle: payload = build_payload(contract, list(csv.DictReader(handle)), now, timestamp(args.generated_at)) with args.output.open("x") as handle: json.dump(payload, handle, indent=2, allow_nan=False) handle.write("\n") except (ValueError, KeyError, OSError) as error: parser.exit(1, f"Cannot prepare submission: {error}\n") if __name__ == "__main__": main()