#!/usr/bin/env python3 """Fetch the four immutable R2 objects used by the V2.1 map presentation.""" from __future__ import annotations import hashlib import json import os import re from concurrent.futures import ThreadPoolExecutor from pathlib import Path import boto3 ENDPOINT = "https://45ba44570ad6c0653ba0dbb52589159e.r2.cloudflarestorage.com" BUCKET = "optical-ground-siting-derived" OUTPUT_ROOT = Path("/srv/data/optical-ground-siting/skyclimate-v21-presentation-source") OBJECTS = { "atlas-admission.json": ( "skyclimate/global-atlas/v2.1/admissions/global-atlas/7a9975fd723f921c93d6981a278c442ed41c53c42c03bf8612b53db8b600ff28.json", "7a9975fd723f921c93d6981a278c442ed41c53c42c03bf8612b53db8b600ff28", 2_931, ), "atlas-manifest.json": ( "skyclimate/global-atlas/v2.1/packs/scga21-4cd927cac60f0a63c7479ad447c69a62bd0f4bae0e8b25df34f6ddee02f57452/manifest.json", "a2de8837fa0aec5d180b77bd89a6840a8322ce687bf3f2ed97b7092b60e143b6", 638_291, ), "release-catalogue.json": ( "skyclimate/catalogue/v2.1/releases/scv21cat1-51b380ad10d1d46a31ad1eadc0732fcf16f91edb7bfcf1b437350ae8d3fd387e/catalogue.json", "6354f2d9f8a1ccbad663005d3c7e3b7590188b368d5d94a84c7fad0ae14ad7b2", 36_615, ), "map_summary.parquet": ( "skyclimate/global-atlas/v2.1/packs/scga21-4cd927cac60f0a63c7479ad447c69a62bd0f4bae0e8b25df34f6ddee02f57452/map_summary.parquet", "707648c5e62114deb6de6a3041965e95289b82075ab3ac170533b89aeaece29a", 62_683_385, ), } SKETCH_NORMALISATIONS = { "cloud": ( "skyclimate/cloud-sketch/v2.1/normalisation/sha256:fa92d86c1b259e57a60fd8ec8739ca9d508922399729b774ea4980f0a04e9624/scsp1-f80cd49925e72f6e309b67ebad4d0a7f1ac994775e94f2c96251c15a58801633", "d2f2782ff5aea071b1f57fb8f63808f8059d868a43f930771e2686d426331541", ), "tcwv": ( "skyclimate/water-vapour-sketch/v2.1/3adf56a90a96ffd0c3caf48a403009827ac6ac6ebd496c7c5345bd40b00963e4/normalisation/sha256:e5c8f40642ace9a782778ce032202dc2fb3182bec37a78ee330b5d5432468eb6/scsp1-f80cd49925e72f6e309b67ebad4d0a7f1ac994775e94f2c96251c15a58801633", "ee6bf17c14cf25c2da70ab32f3e4bc34d1faee34cb6d3214e1450df863225ced", ), } def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def valid(path: Path, expected_sha256: str, expected_size: int) -> bool: return ( path.is_file() and path.stat().st_size == expected_size and sha256(path) == expected_sha256 ) def fetch_runner_publication_metadata( client: object, *, bucket: str, receipt_key: str, receipt_sha256: str, kernel: str, identity_id: str, names: tuple[str, ...] = ("manifest.json",), ) -> dict[str, object]: """Read pinned runner metadata, not payloads or an implicit admission. A computed receipt proves a completed shard, not scientific calibration or publication permission. The release composer must still apply those rules. This reader deliberately never imports kernels or starts a missing shard. """ from skyclimate.r2_v21_catalogue_admission import ( MAXIMUM_JSON_BYTES, _read_bounded, _safe_key, ) def digest(value: object) -> str: if not isinstance(value, str) or not re.fullmatch( r"(?:sha256:)?[0-9a-f]{64}", value ): raise ValueError("runner publication SHA-256 differs") return value.removeprefix("sha256:") if not bucket or not kernel or not identity_id: raise ValueError("runner publication parent is required") digest(identity_id) _safe_key(receipt_key, prefix="skyclimate") if ( Path(receipt_key).name != "receipt.json" or not names or len(names) != len(set(names)) or any(not re.fullmatch(r"[A-Za-z0-9_-]+\.json", name) for name in names) ): raise ValueError("runner publication metadata selection differs") expected_receipt_sha256 = digest(receipt_sha256) payload = _read_bounded(client, bucket=bucket, key=receipt_key) if hashlib.sha256(payload).hexdigest() != expected_receipt_sha256: raise ValueError("runner publication receipt hash differs") receipt = json.loads(payload) if ( type(receipt) is not dict or receipt.get("kind") != "skyclimate-runner-shard-receipt-v1" or receipt.get("status") != "computed" or receipt.get("kernel") != kernel or receipt.get("identity_id") != identity_id or receipt.get("error") is not None or receipt.get("failure_class") is not None or type(receipt.get("shard_id")) is not str or Path(receipt_key).parent.name != receipt["shard_id"] or Path(receipt_key).parent.parent.name != identity_id ): raise ValueError("runner publication receipt is not the completed parent") outputs = receipt.get("outputs") if type(outputs) is not list or any(type(row) is not dict for row in outputs): raise ValueError("runner publication outputs differ") prefix = str(Path(receipt_key).parent) documents: dict[str, object] = {} references: dict[str, object] = {} for name in names: matches = [row for row in outputs if row.get("name") == name] if len(matches) != 1: raise ValueError("runner publication metadata output is absent or duplicated") row = matches[0] if ( row.get("key") != f"{prefix}/{name}" or type(row.get("size_bytes")) is not int or not 0 < row["size_bytes"] <= MAXIMUM_JSON_BYTES ): raise ValueError("runner publication metadata reference differs") expected_sha256 = digest(row.get("sha256")) body = _read_bounded(client, bucket=bucket, key=row["key"]) if ( len(body) != row["size_bytes"] or hashlib.sha256(body).hexdigest() != expected_sha256 ): raise ValueError("runner publication metadata read-back differs") document = json.loads(body) if type(document) is not dict: raise ValueError("runner publication metadata must be an object") documents[name] = document references[name] = { "bucket": bucket, "key": row["key"], "size_bytes": len(body), "sha256": "sha256:" + expected_sha256, } return { "receipt": receipt, "receipt_reference": { "bucket": bucket, "key": receipt_key, "size_bytes": len(payload), "sha256": "sha256:" + expected_receipt_sha256, }, "documents": documents, "document_references": references, "metadata_readback_verified": True, } def public_object_index(release_id: str, group: str, prefix: str, records: list[dict], proof: dict) -> dict: """Build a bounded public allowlist from explicit verified parent records.""" if not re.fullmatch(r"[a-z0-9][a-z0-9._-]{0,127}", release_id) or not re.fullmatch(r"[a-z0-9][a-z0-9-]{0,79}", group): raise ValueError("unsafe public release/group") if not prefix.startswith("skyclimate/") or any(part in {"", ".", ".."} for part in prefix.split("/")): raise ValueError("unsafe source prefix") objects = {} for row in records: key = row.get("key", "") if row.get("bucket", BUCKET) != BUCKET or not key.startswith(prefix + "/"): raise ValueError("public object escapes its derived parent") relative = key[len(prefix) + 1:] digest = str(row.get("sha256", "")).removeprefix("sha256:") size = row.get("size_bytes") if (not re.fullmatch(r"[A-Za-z0-9/._-]+", relative) or any(part in {"", ".", ".."} for part in relative.split("/")) or not re.fullmatch(r"[a-f0-9]{64}", digest) or type(size) is not int or size < 0 or relative in objects): raise ValueError("public object path, digest, size or uniqueness differs") media = {".json": "application/json", ".parquet": "application/vnd.apache.parquet", ".npy": "application/x-npy", ".nc": "application/x-netcdf", ".npz": "application/zip"}.get(Path(relative).suffix, "application/octet-stream") objects[relative] = {"bucket": BUCKET, "key": key, "sha256": digest, "size_bytes": size, "media_type": media} if not objects or proof.get("metadata_readback_verified") is not True: raise ValueError("public index requires verified parent metadata") result = {"$schema": "https://data.dynamikorbits.com/schemas/release-manifest-v1.json#/$defs/derivedObjectIndex", "kind": "skyclimate-public-derived-index-v1", "release_id": release_id, "group": group, "parent": proof, "objects": dict(sorted(objects.items()))} if len(json.dumps(result).encode()) > 4 * 1024 * 1024: raise ValueError("public index exceeds its gateway bound") return result def build_public_data_inventory(client, *, release_id: str, destination: Path, aerosol_release: dict) -> list[dict]: """Publish metadata aliases for existing packs; never invoke a data kernel.""" from skyclimate.r2_hourly_admission import ( fetch_r2_hourly_axis_admission, fetch_r2_hourly_month_report, merge_r2_hourly_axis_admissions, ) from skyclimate.v21_streaming_deployment import HOURLY_ADMISSION_KEYS cache = destination.parent / "publication-metadata-cache" cache.mkdir(parents=True, exist_ok=True) index_root = destination / "_data-index" index_root.mkdir(parents=True, exist_ok=True) def metadata(key, digest): digest = digest.removeprefix("sha256:") if not re.fullmatch(r"[a-f0-9]{64}", digest): raise ValueError("invalid metadata parent digest") path = cache / (digest + ".json") if path.exists(): payload = path.read_bytes() else: response = client.get_object(Bucket=BUCKET, Key=key) try: payload = response["Body"].read(32 * 1024 * 1024 + 1) finally: response["Body"].close() if len(payload) > 32 * 1024 * 1024 or hashlib.sha256(payload).hexdigest() != digest: raise ValueError("public parent metadata hash/bound differs") if not path.exists(): path.write_bytes(payload) return json.loads(payload), {"key": key, "sha256": digest, "size_bytes": len(payload)} def write_group(group, prefix, records, proof, **info): document = public_object_index(release_id, group, prefix, records, proof) path = index_root / f"{group}.json" payload = (json.dumps(document, sort_keys=True, separators=(",", ":")) + "\n").encode() if path.exists() and path.read_bytes() != payload: raise ValueError("existing immutable public index differs") if not path.exists(): path.write_bytes(payload) return {"group": group, "index": f"_data-index/{group}.json", "index_sha256": hashlib.sha256(payload).hexdigest(), "public_data_prefix": f"data/{group}/", "object_count": len(records), "existing_source_bytes": sum(row["size_bytes"] for row in records), **info} def manifest_group(group, prefix, digest, **info): manifest, manifest_record = metadata(prefix + "/manifest.json", digest) artifacts = manifest["artifacts"] records = [manifest_record] for name, row in artifacts.items(): records.append({"key": prefix + "/" + name, "sha256": row["sha256"], "size_bytes": row["size_bytes"]}) proof = {"manifest": manifest_record, "metadata_readback_verified": True, "artifact_hashes_bound_to_parent": True} if manifest.get("claim_class") is not None: proof["claim_class"] = manifest["claim_class"] return write_group(group, prefix, records, proof, **info), manifest def batches(function, items): results = [] with ThreadPoolExecutor(max_workers=6) as pool: for start in range(0, len(items), 6): results.extend(pool.map(function, items[start:start + 6])) return results # The existing admission checks enforce all 96 months and their native axis. admissions = [fetch_r2_hourly_axis_admission(client, bucket=BUCKET, key=key) for key in HOURLY_ADMISSION_KEYS] axis = merge_r2_hourly_axis_admissions(admissions, start_month="2018-01", end_month_inclusive="2025-12") def hourly(row): report = fetch_r2_hourly_month_report(client, bucket=BUCKET, admission_row=row) if report.get("provider_readback_verified") is not True or any(record.get("provider_readback_verified") is not True for record in report["r2_objects"]): raise ValueError("hourly publication lacks original full read-back") # The source manifest remains untouched. Its old exact-site pilot is # explicitly excluded; public map/series cover every native grid cell. records = [record for record in report["r2_objects"] if not record["key"].endswith("/exact-sites.parquet")] return write_group("hourly-" + row["month"], report["r2_prefix"], records, {"report_key": row["r2_report_key"], "report_sha256": Path(row["r2_report_key"]).stem, "original_provider_readback_verified": True, "metadata_readback_verified": True, "claim_class": "source-grid-statistic", "pack_id": row["pack_id"]}, family="cloud-tcwv-hourly", month=row["month"], cadence="PT1H", excluded_artifacts={"exact-sites.parquet": "historical candidate pilot; use global cell-based series instead"}) groups = batches(hourly, axis["months"]) print(json.dumps({"indexed_hourly_months": len(groups), "new_data_computation": False}), flush=True) # All spatial reductions stay available, including the 325 conditioned views. atlas_prefix = OBJECTS["atlas-manifest.json"][0].removesuffix("/manifest.json") atlas_group, _ = manifest_group("atlas", atlas_prefix, OBJECTS["atlas-manifest.json"][1], family="cloud-tcwv-statistics", cadence="statistics-over-PT1H") groups.append(atlas_group) for family, (prefix, digest) in SKETCH_NORMALISATIONS.items(): group, normalisation = manifest_group(family + "-normalisation", prefix, digest, family=family + "-sketch", cadence="PT1H", claim_class="derived-retrieval-index") groups.append(group) identity = normalisation["inputs_identity"]["shard_kernel_identity_id"] shards = sorted(normalisation["inputs_identity"]["shard_manifest_sha256"].items()) if len(shards) != 178: raise ValueError("full-catalogue sketch shard inventory differs") shard_root = prefix.split("/normalisation/")[0] + "/shards/" + identity def one_sketch(item, family=family, shard_root=shard_root): shard, digest = item group, manifest = manifest_group(f"{family}-sketch-{shard.replace('.', '-')}", shard_root + "/" + shard, digest, family=family + "-sketch", cadence="PT1H", claim_class="derived-retrieval-index") if manifest.get("claim_class") != "derived-retrieval-index": raise ValueError("sketch source claim differs") group["candidate_support"] = manifest.get("candidates") return group groups.extend(batches(one_sketch, shards)) print(json.dumps({"indexed_sketch_family": family, "shards": len(shards)}), flush=True) if len(aerosol_release["tiles"]) != 120 or aerosol_release["period"].get("cadence") not in {None, "PT3H"}: raise ValueError("aerosol release inventory differs") def aerosol(tile): prefix = tile["manifest_key"].removesuffix("/manifest.json") group, manifest = manifest_group("aerosol-" + tile["shard_id"], prefix, tile["manifest_sha256"], family="aerosol-statistics-sketch", cadence="PT3H") if manifest.get("claim_class") not in {"derived-retrieval-index", "model-derived-screening"}: raise ValueError("aerosol source claim differs") return group groups.extend(batches(aerosol, aerosol_release["tiles"])) aerosol_root = ( "skyclimate/aerosol-sketch/v2.1/release/sha256:d90a4531594149e46869cd6d54f7e125a4a28c062c1dc764220b4c15b1252442/" "scasp21-9b05685159b13db0136baad07b668800e17568ed445248262216feb5e416c6b1" ) group, _ = manifest_group("aerosol-normalisation", aerosol_root, "0051547fb451a2fb4c4b537398375affa848ca6b99703a3e769cc74af7363878", family="aerosol-sketch", cadence="PT3H") groups.append(group) axis_prefix = "skyclimate/cams-eac4-spectral/v2.1/temporal-axis/sha256:4c5525debb3a329334efe647fd5774eba002fa5135d36e6854cc2ac4518070c5/2018-2025" aerosol_axis, axis_record = metadata(axis_prefix + "/axis.json", "df3533f5f3a4d564f93967eb47595cbbd93554b574ceb62f141d4c86dece691a") if (aerosol_axis.get("axis_id") != aerosol_release["parent_temporal_axis_id"] or len(aerosol_axis.get("months", [])) != 96 or aerosol_axis["period"]["ordered_sample_count"] != 23376): raise ValueError("published aerosol axis differs from its admitted release") groups.append(write_group("aerosol-axis", axis_prefix, [axis_record], {"metadata_readback_verified": True, "axis_id": aerosol_axis["axis_id"], "manifest": axis_record}, family="aerosol-temporal", cadence="PT3H")) def aerosol_month(row): prefix = row["receipt_key"].removesuffix("/receipt.json") group, manifest = manifest_group("aerosol-month-" + row["month"], prefix, row["manifest_sha256"], family="aerosol-temporal", month=row["month"], cadence="PT3H") if manifest.get("pack_id") != row["pack_id"]: raise ValueError("aerosol month identity differs") return group groups.extend(batches(aerosol_month, aerosol_axis["months"])) print(json.dumps({"indexed_aerosol_tiles": 120, "indexed_aerosol_months": 96, "groups": len(groups)}), flush=True) return sorted(groups, key=lambda row: row["group"]) def main() -> None: if not os.environ.get("AWS_ACCESS_KEY_ID") or not os.environ.get("AWS_SECRET_ACCESS_KEY"): raise RuntimeError("scoped R2 credentials are required in the environment") OUTPUT_ROOT.mkdir(parents=True, exist_ok=True) if not str(OUTPUT_ROOT.resolve()).startswith("/srv/data/optical-ground-siting/"): raise RuntimeError("presentation source cache escaped its bounded data root") client = boto3.client("s3", endpoint_url=ENDPOINT, region_name="auto") for filename, (key, expected_sha256, expected_size) in OBJECTS.items(): target = OUTPUT_ROOT / filename if target.exists(): if not valid(target, expected_sha256, expected_size): raise RuntimeError(f"existing immutable source differs: {target}") print(f"verified {filename}") continue partial = OUTPUT_ROOT / f".{filename}.{os.getpid()}.part" if partial.exists(): raise RuntimeError(f"private partial already exists: {partial}") try: client.download_file(BUCKET, key, str(partial)) if not valid(partial, expected_sha256, expected_size): raise RuntimeError(f"provider object differs after download: {key}") partial.replace(target) finally: if partial.exists(): partial.unlink() print(f"downloaded and verified {filename}") if __name__ == "__main__": main()