From ab74034f45039b4adfc7b971a89d4813c67df26b Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:19 +0000 Subject: [PATCH 1/3] Add generic Core Stack GEE ingestion script This script standardizes memory-bounded publication of large Core Stack GeoPackage and GeoJSON assets to Earth Engine through TSV, GCS, and manifest ingestion. It is kept separate from dataset-building code so upload/retry/verify workflows can be reviewed independently. --- .../scripts/gee/core_stack_gee_ingest.py | 1306 +++++++++++++++++ 1 file changed, 1306 insertions(+) create mode 100644 utilities/scripts/gee/core_stack_gee_ingest.py diff --git a/utilities/scripts/gee/core_stack_gee_ingest.py b/utilities/scripts/gee/core_stack_gee_ingest.py new file mode 100644 index 00000000..4df050a4 --- /dev/null +++ b/utilities/scripts/gee/core_stack_gee_ingest.py @@ -0,0 +1,1306 @@ +#!/usr/bin/env python3 +"""Build and upload config-driven Core Stack vector assets to Earth Engine. + +The ingestion path deliberately mirrors the reliable facilities upload flow: + +1. Export a bounded-memory tab-delimited table with one GeoJSON geometry column. +2. Stage the same bytes to GCS with a ``.csv`` object suffix. +3. Start an Earth Engine table ingestion task from a manifest. + +The script is generic for common GeoPackage/GeoJSON feature layers, with +configurable special exporters for assets that are relational locally but should +be flat and native in Earth Engine. +""" + +from __future__ import annotations + +import argparse +import csv +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone +from decimal import Decimal +import json +import logging +import math +from pathlib import Path +import resource +import sqlite3 +import time +from typing import Any, Iterable, Iterator, Sequence +import uuid + +import yaml + + +log = logging.getLogger("core_stack_gee_ingest") + +DEFAULT_CONFIG = Path(__file__).with_name("core_stack_gee_assets.yaml") +TERMINAL_TASK_STATES = {"SUCCEEDED", "COMPLETED", "FAILED", "CANCELLED"} +SUCCESS_TASK_STATES = {"SUCCEEDED", "COMPLETED"} +GEE_OAUTH_SCOPES = [ + "https://www.googleapis.com/auth/earthengine", + "https://www.googleapis.com/auth/devstorage.full_control", +] +DEFAULT_MAX_RSS_MB = 5_000 + + +def setup_logging(debug: bool = False, log_file: Path | None = None) -> None: + handlers: list[logging.Handler] = [logging.StreamHandler()] + if log_file is not None: + log_file.parent.mkdir(parents=True, exist_ok=True) + handlers.append(logging.FileHandler(log_file, encoding="utf-8")) + logging.basicConfig( + level=logging.DEBUG if debug else logging.INFO, + format="%(asctime)s [%(levelname)-7s] %(message)s", + datefmt="%H:%M:%S", + handlers=handlers, + ) + + +def find_repo_root(start: Path) -> Path: + for candidate in [start, *start.parents]: + if (candidate / ".git").exists() or (candidate / "manage.py").exists(): + return candidate + return start + + +def repo_path(repo_root: Path, value: str | Path) -> Path: + path = Path(value) + return path if path.is_absolute() else repo_root / path + + +def read_yaml(path: Path) -> dict[str, Any]: + with path.open("r", encoding="utf-8") as handle: + return yaml.safe_load(handle) or {} + + +def write_yaml(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + with path.open("w", encoding="utf-8") as handle: + yaml.safe_dump(payload, handle, sort_keys=False, allow_unicode=True) + + +def write_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2, ensure_ascii=False, sort_keys=True), encoding="utf-8") + + +def sql_ident(value: str) -> str: + return '"' + str(value).replace('"', '""') + '"' + + +def slug(value: Any) -> str: + text = str(value or "").strip().lower() + out: list[str] = [] + previous_underscore = False + for char in text: + if char.isalnum(): + out.append(char) + previous_underscore = False + elif not previous_underscore: + out.append("_") + previous_underscore = True + return "".join(out).strip("_") + + +def max_rss_mb() -> float: + # Linux ru_maxrss is KiB. + return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 + + +def enforce_memory(max_rss_mb_limit: int, label: str) -> None: + rss = max_rss_mb() + if rss > max_rss_mb_limit: + raise MemoryError(f"{label} exceeded max RSS {max_rss_mb_limit} MB; current RSS={rss:.1f} MB") + + +def json_safe(value: Any) -> Any: + if value is None: + return None + if isinstance(value, float) and (math.isnan(value) or math.isinf(value)): + return None + if isinstance(value, Decimal): + return int(value) if value == value.to_integral_value() else float(value) + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return value + + +def csv_safe(value: Any) -> Any: + value = json_safe(value) + if value is None: + return "" + if isinstance(value, (dict, list, tuple)): + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + if isinstance(value, str): + return value.replace("\r\n", " ").replace("\n", " ").replace("\r", " ").replace("\t", " ") + return value + + +def compact_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":")) + + +def with_suffix_token(path: Path, token: str | None) -> Path: + if not token: + return path + clean = token if token.startswith(".") else f".{token}" + return path.with_name(f"{path.stem}{clean}{path.suffix}") + + +class DelimitedTableWriter: + def __init__(self, path: Path, fieldnames: Sequence[str], delimiter: str, quotechar: str): + self.path = path + self.fieldnames = list(fieldnames) + self.delimiter = delimiter + self.quotechar = quotechar + self.handle = None + self.writer = None + self.rows_written = 0 + + def __enter__(self) -> "DelimitedTableWriter": + self.path.parent.mkdir(parents=True, exist_ok=True) + self.handle = self.path.open("w", encoding="utf-8", newline="") + self.writer = csv.DictWriter( + self.handle, + fieldnames=self.fieldnames, + delimiter=self.delimiter, + quotechar=self.quotechar, + quoting=csv.QUOTE_ALL, + lineterminator="\n", + extrasaction="ignore", + ) + self.writer.writeheader() + return self + + def write(self, row: dict[str, Any]) -> None: + assert self.writer is not None + self.writer.writerow({field: csv_safe(row.get(field)) for field in self.fieldnames}) + self.rows_written += 1 + + def __exit__(self, exc_type, exc, traceback) -> None: + if self.handle is not None: + self.handle.close() + + +def maybe_log_progress(label: str, rows: int, started: float, last_logged: float, every_rows: int = 100_000) -> float: + now = time.monotonic() + if rows == 1 or rows % every_rows == 0 or (now - last_logged) > 30: + elapsed = max(now - started, 0.001) + log.info("%s: %s rows in %.1fs; rss=%.1f MB", label, f"{rows:,}", elapsed, max_rss_mb()) + return now + return last_logged + + +def gpkg_geometry_to_geojson(blob: bytes | memoryview | None) -> dict[str, Any] | None: + if blob is None: + return None + from shapely import wkb + from shapely.geometry import mapping + + data = bytes(blob) + if data[:2] != b"GP": + geom = wkb.loads(data) + else: + flags = data[3] + if flags & 0b00010000: + return None + envelope_code = (flags >> 1) & 0b00000111 + envelope_bytes = {0: 0, 1: 32, 2: 48, 3: 48, 4: 64}.get(envelope_code) + if envelope_bytes is None: + raise ValueError(f"Unsupported GeoPackage geometry envelope code: {envelope_code}") + geom = wkb.loads(data[8 + envelope_bytes :]) + if geom.is_empty: + return None + return mapping(geom) + + +def point_geometry(lon: Any, lat: Any) -> dict[str, Any] | None: + if lon is None or lat is None: + return None + lon = float(lon) + lat = float(lat) + if math.isnan(lon) or math.isnan(lat): + return None + return {"type": "Point", "coordinates": [lon, lat]} + + +def gpkg_geometry_column(con: sqlite3.Connection, layer: str) -> str: + row = con.execute( + "SELECT column_name FROM gpkg_geometry_columns WHERE table_name = ?", + (layer,), + ).fetchone() + return str(row[0]) if row else "geom" + + +def gpkg_table_columns(con: sqlite3.Connection, layer: str) -> list[tuple[str, str]]: + return [(str(row[1]), str(row[2] or "")) for row in con.execute(f"PRAGMA table_info({sql_ident(layer)})")] + + +def configured_columns( + available_columns: Sequence[str], + geometry_column: str, + asset_cfg: dict[str, Any], +) -> tuple[list[str], dict[str, str]]: + properties_cfg = asset_cfg.get("properties") or {} + include = properties_cfg.get("include") or asset_cfg.get("columns") or ["*"] + exclude = set(properties_cfg.get("exclude") or []) + rename = dict(properties_cfg.get("rename") or {}) + skip = {"fid", geometry_column} + if include == ["*"] or include == "*": + selected = [column for column in available_columns if column not in skip and column not in exclude] + else: + selected = [str(column) for column in include if str(column) in available_columns and str(column) not in exclude] + output_names = {column: str(rename.get(column, column)) for column in selected} + return selected, output_names + + +def output_delimiter(upload_cfg: dict[str, Any]) -> str: + return str(upload_cfg.get("csv_delimiter", "\t")) + + +def output_quotechar(upload_cfg: dict[str, Any]) -> str: + return str(upload_cfg.get("csv_qualifier", '"')) + + +def column_specs(items: Sequence[Any]) -> list[tuple[str, str]]: + specs: list[tuple[str, str]] = [] + for item in items: + if isinstance(item, str): + specs.append((item, item)) + elif isinstance(item, dict): + if "source" in item: + specs.append((str(item["source"]), str(item.get("target") or item["source"]))) + elif len(item) == 1: + source, target = next(iter(item.items())) + specs.append((str(source), str(target))) + else: + raise ValueError(f"Column mapping must have source/target or one source: target pair: {item}") + else: + raise TypeError(f"Unsupported column config item: {item!r}") + return specs + + +def target_for_source(specs: Sequence[tuple[str, str]], source_name: str) -> str: + for source, target in specs: + if source == source_name: + return target + return source_name + + +def asset_output_path(repo_root: Path, asset_name: str, asset_cfg: dict[str, Any], outputs_cfg: dict[str, Any], suffix: str | None) -> Path: + if asset_cfg.get("output_table"): + path = repo_path(repo_root, asset_cfg["output_table"]) + else: + directory = repo_path(repo_root, outputs_cfg.get("directory", "data/gee/core_stack")) + extension = outputs_cfg.get("local_extension", ".tsv") + path = directory / f"{asset_name}{extension}" + return with_suffix_token(path, suffix) + + +def sqlite_fetchmany(cursor: sqlite3.Cursor, size: int) -> Iterator[sqlite3.Row]: + while True: + rows = cursor.fetchmany(size) + if not rows: + break + yield from rows + + +def export_gpkg_feature_layer( + repo_root: Path, + asset_name: str, + asset_cfg: dict[str, Any], + outputs_cfg: dict[str, Any], + upload_cfg: dict[str, Any], + *, + limit: int | None, + chunk_size: int, + overwrite: bool, + output_suffix: str | None, + max_rss_mb_limit: int, +) -> dict[str, Any]: + source_path = repo_path(repo_root, asset_cfg["source"]) + layer = asset_cfg["layer"] + output = asset_output_path(repo_root, asset_name, asset_cfg, outputs_cfg, output_suffix) + if output.exists() and not overwrite: + raise FileExistsError(f"Output exists; pass --overwrite to replace: {output}") + + started = time.monotonic() + with sqlite3.connect(source_path) as con: + con.row_factory = sqlite3.Row + geom_col = gpkg_geometry_column(con, layer) + all_columns = [column for column, _ in gpkg_table_columns(con, layer)] + selected_columns, output_names = configured_columns(all_columns, geom_col, asset_cfg) + property_fields = [output_names[column] for column in selected_columns] + geometry_field = asset_cfg.get("geometry_column", "geometry") + fieldnames = [*property_fields, geometry_field] + select_cols = [sql_ident(column) for column in selected_columns] + select_sql = ", ".join(["fid", *select_cols, sql_ident(geom_col)]) + total_rows = int(con.execute(f"SELECT COUNT(*) FROM {sql_ident(layer)}").fetchone()[0]) + remaining = limit + last_fid = 0 + rows_written = 0 + last_logged = started + with DelimitedTableWriter(output, fieldnames, output_delimiter(upload_cfg), output_quotechar(upload_cfg)) as writer: + while True: + current_chunk = chunk_size if remaining is None else min(chunk_size, remaining) + if current_chunk <= 0: + break + rows = con.execute( + f""" + SELECT {select_sql} + FROM {sql_ident(layer)} + WHERE fid > ? + ORDER BY fid + LIMIT ? + """, + (last_fid, current_chunk), + ).fetchall() + if not rows: + break + last_fid = int(rows[-1]["fid"]) + for row in rows: + geometry = gpkg_geometry_to_geojson(row[geom_col]) + if geometry is None: + continue + out = {output_names[column]: json_safe(row[column]) for column in selected_columns} + out[geometry_field] = compact_json(geometry) + writer.write(out) + rows_written += 1 + last_logged = maybe_log_progress(asset_name, rows_written, started, last_logged) + enforce_memory(max_rss_mb_limit, asset_name) + if remaining is not None: + remaining -= len(rows) + + return { + "asset": asset_name, + "kind": "feature_layer", + "source": str(source_path), + "layer": layer, + "rows_source": total_rows, + "rows": rows_written, + "columns": fieldnames, + "output": str(output), + "size_bytes": output.stat().st_size, + "elapsed_seconds": round(time.monotonic() - started, 3), + "max_rss_mb": round(max_rss_mb(), 1), + } + + +def iter_geojson_features(path: Path, max_in_memory_mb: int) -> Iterator[dict[str, Any]]: + suffix = path.suffix.lower() + if suffix in {".geojsonl", ".jsonl", ".ndjson"}: + with path.open("r", encoding="utf-8") as handle: + for line in handle: + if line.strip(): + yield json.loads(line) + return + try: + import ijson + except ImportError: + if path.stat().st_size > max_in_memory_mb * 1024 * 1024: + raise RuntimeError( + f"{path} is too large for in-memory GeoJSON parsing without ijson. " + "Run with `uv run --with ijson ...` or convert to GeoJSONL." + ) + with path.open("r", encoding="utf-8") as handle: + payload = json.load(handle) + for feature in payload.get("features", []): + yield feature + return + with path.open("rb") as handle: + yield from ijson.items(handle, "features.item") + + +def export_geojson_feature_layer( + repo_root: Path, + asset_name: str, + asset_cfg: dict[str, Any], + outputs_cfg: dict[str, Any], + upload_cfg: dict[str, Any], + *, + limit: int | None, + overwrite: bool, + output_suffix: str | None, + max_rss_mb_limit: int, +) -> dict[str, Any]: + source_path = repo_path(repo_root, asset_cfg["source"]) + output = asset_output_path(repo_root, asset_name, asset_cfg, outputs_cfg, output_suffix) + if output.exists() and not overwrite: + raise FileExistsError(f"Output exists; pass --overwrite to replace: {output}") + properties_cfg = asset_cfg.get("properties") or {} + include = properties_cfg.get("include") or ["*"] + exclude = set(properties_cfg.get("exclude") or []) + rename = dict(properties_cfg.get("rename") or {}) + geometry_field = asset_cfg.get("geometry_column", "geometry") + max_json_mb = int(asset_cfg.get("max_in_memory_geojson_mb", 512)) + + started = time.monotonic() + rows_written = 0 + fieldnames: list[str] | None = None + last_logged = started + writer_context = None + writer = None + try: + for feature in iter_geojson_features(source_path, max_json_mb): + props = feature.get("properties") or {} + if fieldnames is None: + if include == ["*"] or include == "*": + selected = [key for key in props.keys() if key not in exclude] + else: + selected = [key for key in include if key in props and key not in exclude] + fieldnames = [str(rename.get(column, column)) for column in selected] + fieldnames.append(geometry_field) + writer_context = DelimitedTableWriter(output, fieldnames, output_delimiter(upload_cfg), output_quotechar(upload_cfg)) + writer = writer_context.__enter__() + out = { + str(rename.get(column, column)): json_safe(props.get(column)) + for column in props.keys() + if (include == ["*"] or include == "*" or column in include) and column not in exclude + } + geometry = feature.get("geometry") + if not geometry: + continue + out[geometry_field] = compact_json(geometry) + writer.write(out) + rows_written += 1 + last_logged = maybe_log_progress(asset_name, rows_written, started, last_logged) + enforce_memory(max_rss_mb_limit, asset_name) + if limit and rows_written >= limit: + break + finally: + if writer_context is not None: + writer_context.__exit__(None, None, None) + if fieldnames is None: + fieldnames = [geometry_field] + with DelimitedTableWriter(output, fieldnames, output_delimiter(upload_cfg), output_quotechar(upload_cfg)): + pass + + return { + "asset": asset_name, + "kind": "geojson_feature_layer", + "source": str(source_path), + "rows": rows_written, + "columns": fieldnames, + "output": str(output), + "size_bytes": output.stat().st_size, + "elapsed_seconds": round(time.monotonic() - started, 3), + "max_rss_mb": round(max_rss_mb(), 1), + } + + +def export_facility_membership_points( + repo_root: Path, + asset_name: str, + asset_cfg: dict[str, Any], + outputs_cfg: dict[str, Any], + upload_cfg: dict[str, Any], + *, + limit: int | None, + chunk_size: int, + overwrite: bool, + output_suffix: str | None, + max_rss_mb_limit: int, +) -> dict[str, Any]: + gpkg = repo_path(repo_root, asset_cfg["source"]) + output = asset_output_path(repo_root, asset_name, asset_cfg, outputs_cfg, output_suffix) + if output.exists() and not overwrite: + raise FileExistsError(f"Output exists; pass --overwrite to replace: {output}") + columns_cfg = asset_cfg["columns"] + facility_specs = column_specs(columns_cfg["facility"]) + membership_specs = column_specs(columns_cfg["membership"]) + generated_cols = columns_cfg.get("generated", []) + geometry_field = asset_cfg.get("geometry_column", "geometry") + facility_targets = [target for _, target in facility_specs] + membership_targets = [target for _, target in membership_specs] + lon_target = target_for_source(facility_specs, "longitude") + lat_target = target_for_source(facility_specs, "latitude") + fieldnames = list(dict.fromkeys([*generated_cols, *facility_targets, *membership_targets, geometry_field])) + facility_select = ", ".join(f"f.{sql_ident(source)} AS {sql_ident(target)}" for source, target in facility_specs) + membership_select = ", ".join(f"m.{sql_ident(source)} AS {sql_ident(target)}" for source, target in membership_specs) + sql = f""" + SELECT {facility_select}, {membership_select} + FROM facility_memberships m + JOIN facilities f ON f.facility_uid = m.facility_uid + WHERE f.latitude IS NOT NULL + AND f.longitude IS NOT NULL + """ + if limit: + sql += f" LIMIT {int(limit)}" + + started = time.monotonic() + last_logged = started + rows_written = 0 + with sqlite3.connect(gpkg) as con: + con.row_factory = sqlite3.Row + cursor = con.execute(sql) + with DelimitedTableWriter(output, fieldnames, output_delimiter(upload_cfg), output_quotechar(upload_cfg)) as writer: + while True: + rows = cursor.fetchmany(chunk_size) + if not rows: + break + for row in rows: + props = dict(row) + geometry = point_geometry(props.get(lon_target), props.get(lat_target)) + if geometry is None: + continue + props["facility_membership_uid"] = "|".join( + str(props.get(col) or "") + for col in ("facility_uid", "class_l3_facility_class", "class_l4_facility_subtype", "class_k1") + ) + props[geometry_field] = compact_json(geometry) + writer.write(props) + rows_written += 1 + last_logged = maybe_log_progress(asset_name, rows_written, started, last_logged) + enforce_memory(max_rss_mb_limit, asset_name) + + return { + "asset": asset_name, + "kind": "facility_membership_points", + "source": str(gpkg), + "rows": rows_written, + "columns": fieldnames, + "output": str(output), + "size_bytes": output.stat().st_size, + "elapsed_seconds": round(time.monotonic() - started, 3), + "max_rss_mb": round(max_rss_mb(), 1), + } + + +def proximity_schema(con: sqlite3.Connection, asset_cfg: dict[str, Any]) -> dict[str, Any]: + class_rows = con.execute( + """ + SELECT DISTINCT class_l1_domain, class_l2_filter_group, class_l3_facility_class + FROM proximity_class_map + ORDER BY class_l1_domain, class_l2_filter_group, class_l3_facility_class + """ + ).fetchall() + l3_classes = [row[2] for row in class_rows] + l2_groups = sorted({(row[0], row[1]) for row in class_rows}) + l1_domains = sorted({row[0] for row in class_rows}) + include_l3 = asset_cfg["include_l3"] + include_l2 = asset_cfg["include_l2"] + l1_cfg = asset_cfg.get("include_l1_summary") or {} + l3_distance_cols = {cls: include_l3["distance_template"].format(class_name=slug(cls)) for cls in l3_classes} + l3_uid_cols = {cls: include_l3["facility_uid_template"].format(class_name=slug(cls)) for cls in l3_classes} + l2_distance_cols = {group: include_l2["distance_template"].format(group_name=slug(group[1])) for group in l2_groups} + l2_uid_cols = {group: include_l2["facility_uid_template"].format(group_name=slug(group[1])) for group in l2_groups} + l2_selected_cols = {group: include_l2["selected_l3_template"].format(group_name=slug(group[1])) for group in l2_groups} + l1_distance_cols = {} + l1_group_cols = {} + if l1_cfg.get("enabled", False): + l1_distance_cols = {domain: l1_cfg["nearest_distance_template"].format(domain_name=slug(domain)) for domain in l1_domains} + l1_group_cols = {domain: l1_cfg["nearest_l2_template"].format(domain_name=slug(domain)) for domain in l1_domains} + return { + "l3_classes": l3_classes, + "l2_groups": l2_groups, + "l1_domains": l1_domains, + "l3_distance_cols": l3_distance_cols, + "l3_uid_cols": l3_uid_cols, + "l2_distance_cols": l2_distance_cols, + "l2_uid_cols": l2_uid_cols, + "l2_selected_cols": l2_selected_cols, + "l1_distance_cols": l1_distance_cols, + "l1_group_cols": l1_group_cols, + } + + +def create_requested_villages(con: sqlite3.Connection, ids: Sequence[str]) -> None: + con.execute("DROP TABLE IF EXISTS temp.requested_villages") + con.execute("CREATE TEMP TABLE requested_villages (cs_feature_id TEXT PRIMARY KEY)") + con.executemany("INSERT OR IGNORE INTO requested_villages VALUES (?)", [(str(item),) for item in ids]) + + +def fetch_l3_metrics(con: sqlite3.Connection, schema: dict[str, Any]) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + for village_id, cls, distance, uid in con.execute( + """ + SELECT cs_feature_id, class_l3_facility_class, nearest_distance_km, nearest_facility_uid + FROM proximity_l3 + WHERE cs_feature_id IN (SELECT cs_feature_id FROM requested_villages) + """ + ): + row = out.setdefault(str(village_id), {}) + row[schema["l3_distance_cols"][cls]] = distance + row[schema["l3_uid_cols"][cls]] = uid + return out + + +def fetch_l2_metrics(con: sqlite3.Connection, schema: dict[str, Any]) -> dict[str, dict[str, Any]]: + out: dict[str, dict[str, Any]] = {} + l1_best: dict[str, dict[str, tuple[float, str]]] = {} + for village_id, l1, l2, distance, selected_l3, uid in con.execute( + """ + SELECT cs_feature_id, class_l1_domain, class_l2_filter_group, + logic_distance_km, selected_component_class, nearest_facility_uid + FROM proximity_l2_materialized + WHERE cs_feature_id IN (SELECT cs_feature_id FROM requested_villages) + """ + ): + village_id = str(village_id) + group_key = (l1, l2) + row = out.setdefault(village_id, {}) + row[schema["l2_distance_cols"][group_key]] = distance + row[schema["l2_uid_cols"][group_key]] = uid + row[schema["l2_selected_cols"][group_key]] = selected_l3 + if schema["l1_distance_cols"] and distance is not None: + best_by_domain = l1_best.setdefault(village_id, {}) + current = best_by_domain.get(l1) + if current is None or float(distance) < current[0]: + best_by_domain[l1] = (float(distance), l2) + for village_id, domains in l1_best.items(): + row = out.setdefault(village_id, {}) + for domain, (distance, l2_group) in domains.items(): + row[schema["l1_distance_cols"][domain]] = distance + row[schema["l1_group_cols"][domain]] = l2_group + return out + + +def export_village_proximity_wide( + repo_root: Path, + asset_name: str, + asset_cfg: dict[str, Any], + outputs_cfg: dict[str, Any], + upload_cfg: dict[str, Any], + *, + limit: int | None, + chunk_size: int, + overwrite: bool, + output_suffix: str | None, + max_rss_mb_limit: int, +) -> dict[str, Any]: + gpkg = repo_path(repo_root, asset_cfg["source"]) + output = asset_output_path(repo_root, asset_name, asset_cfg, outputs_cfg, output_suffix) + if output.exists() and not overwrite: + raise FileExistsError(f"Output exists; pass --overwrite to replace: {output}") + started = time.monotonic() + rows_written = 0 + with sqlite3.connect(gpkg) as con: + con.row_factory = sqlite3.Row + con.execute("PRAGMA temp_store = MEMORY") + schema = proximity_schema(con, asset_cfg) + admin_map = asset_cfg["admin_columns"] + metric_fields: list[str] = [] + for cls in schema["l3_classes"]: + metric_fields.extend([schema["l3_distance_cols"][cls], schema["l3_uid_cols"][cls]]) + for group in schema["l2_groups"]: + metric_fields.extend([schema["l2_distance_cols"][group], schema["l2_uid_cols"][group], schema["l2_selected_cols"][group]]) + for domain in schema["l1_domains"]: + if domain in schema["l1_distance_cols"]: + metric_fields.extend([schema["l1_distance_cols"][domain], schema["l1_group_cols"][domain]]) + geometry_field = asset_cfg.get("geometry_column", "geometry") + fieldnames = [*admin_map.values(), *metric_fields, geometry_field] + source_cols = list(admin_map.keys()) + select_cols = ", ".join(["fid", *[sql_ident(col) for col in source_cols], "geom"]) + last_fid = 0 + remaining = limit + last_logged = started + with DelimitedTableWriter(output, fieldnames, output_delimiter(upload_cfg), output_quotechar(upload_cfg)) as writer: + while True: + current_chunk = chunk_size if remaining is None else min(chunk_size, remaining) + if current_chunk <= 0: + break + rows = con.execute( + f""" + SELECT {select_cols} + FROM village_shapes + WHERE fid > ? + ORDER BY fid + LIMIT ? + """, + (last_fid, current_chunk), + ).fetchall() + if not rows: + break + last_fid = int(rows[-1]["fid"]) + ids = [str(row["cs_feature_id"]) for row in rows] + create_requested_villages(con, ids) + l3_metrics = fetch_l3_metrics(con, schema) + l2_metrics = fetch_l2_metrics(con, schema) + for row in rows: + village_id = str(row["cs_feature_id"]) + props = {target: json_safe(row[source]) for source, target in admin_map.items()} + props.update(l3_metrics.get(village_id, {})) + props.update(l2_metrics.get(village_id, {})) + geometry = gpkg_geometry_to_geojson(row["geom"]) + if geometry is None: + continue + props[geometry_field] = compact_json(geometry) + writer.write(props) + rows_written += 1 + last_logged = maybe_log_progress(asset_name, rows_written, started, last_logged) + enforce_memory(max_rss_mb_limit, asset_name) + if remaining is not None: + remaining -= len(rows) + return { + "asset": asset_name, + "kind": "village_proximity_wide", + "source": str(gpkg), + "rows": rows_written, + "columns": fieldnames, + "output": str(output), + "size_bytes": output.stat().st_size, + "elapsed_seconds": round(time.monotonic() - started, 3), + "max_rss_mb": round(max_rss_mb(), 1), + } + + +def export_asset( + repo_root: Path, + asset_name: str, + asset_cfg: dict[str, Any], + outputs_cfg: dict[str, Any], + upload_cfg: dict[str, Any], + args: argparse.Namespace, +) -> dict[str, Any]: + kind = asset_cfg.get("kind", "feature_layer") + common = { + "repo_root": repo_root, + "asset_name": asset_name, + "asset_cfg": asset_cfg, + "outputs_cfg": outputs_cfg, + "upload_cfg": upload_cfg, + "limit": args.limit, + "chunk_size": args.chunk_size, + "overwrite": args.overwrite, + "output_suffix": args.output_suffix, + "max_rss_mb_limit": args.max_rss_mb, + } + if kind == "feature_layer": + source_format = str(asset_cfg.get("format", "gpkg")).lower() + if source_format == "gpkg": + return export_gpkg_feature_layer(**common) + if source_format in {"geojson", "geojsonl", "jsonl"}: + return export_geojson_feature_layer( + repo_root, + asset_name, + asset_cfg, + outputs_cfg, + upload_cfg, + limit=args.limit, + overwrite=args.overwrite, + output_suffix=args.output_suffix, + max_rss_mb_limit=args.max_rss_mb, + ) + if kind == "facility_membership_points": + return export_facility_membership_points(**common) + if kind == "village_proximity_wide": + return export_village_proximity_wide(**common) + raise ValueError(f"Unsupported asset kind for {asset_name}: {kind}") + + +def selected_assets(config: dict[str, Any], requested: str) -> list[tuple[str, dict[str, Any]]]: + assets = config.get("assets") or {} + if requested == "all": + return list(assets.items()) + if requested not in assets: + raise KeyError(f"Unknown asset {requested!r}; choose one of: all, {', '.join(assets)}") + return [(requested, assets[requested])] + + +def load_config(args: argparse.Namespace) -> tuple[Path, Path, dict[str, Any]]: + config_path = args.config.resolve() + repo_root = find_repo_root(config_path) + return config_path, repo_root, read_yaml(config_path) + + +def build_assets(args: argparse.Namespace) -> dict[str, Any]: + config_path, repo_root, config = load_config(args) + outputs_cfg = config.get("outputs") or {} + upload_cfg = config.get("upload") or {} + assets = selected_assets(config, args.asset) + summary = { + "built_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), + "config": str(config_path), + "max_rss_mb_limit": args.max_rss_mb, + "assets": {}, + } + jobs = max(1, int(args.jobs or 1)) + if jobs == 1 or len(assets) <= 1: + for name, asset_cfg in assets: + summary["assets"][name] = export_asset(repo_root, name, asset_cfg, outputs_cfg, upload_cfg, args) + else: + with ThreadPoolExecutor(max_workers=jobs) as executor: + futures = { + executor.submit(export_asset, repo_root, name, asset_cfg, outputs_cfg, upload_cfg, args): name + for name, asset_cfg in assets + } + for future in as_completed(futures): + name = futures[future] + summary["assets"][name] = future.result() + summary_path = with_suffix_token(repo_path(repo_root, outputs_cfg.get("build_summary_yaml", "data/gee/core_stack/core_stack_gee_build_summary.yaml")), args.output_suffix) + write_yaml(summary_path, summary) + log.info("Wrote build summary: %s", summary_path) + return summary + + +def initialize_ee(service_account_json: Path) -> tuple[Any, Any, dict[str, Any]]: + try: + import ee + from google.oauth2 import service_account + except ImportError as exc: + raise RuntimeError( + "Upload requires earthengine-api and google-cloud-storage. " + "Run with `uv run --with earthengine-api --with google-cloud-storage --with pyyaml --with shapely ...`." + ) from exc + key = json.loads(service_account_json.read_text(encoding="utf-8")) + credentials = service_account.Credentials.from_service_account_file(service_account_json, scopes=GEE_OAUTH_SCOPES) + ee.Initialize(credentials=credentials, project=key.get("project_id")) + if hasattr(ee.data, "setUserAgent"): + ee.data.setUserAgent("core-stack-gee-ingest") + return ee, credentials, key + + +def gee_asset_exists(ee_module: Any, asset_id: str) -> bool: + try: + ee_module.data.getAsset(asset_id) + return True + except Exception: + return False + + +def ensure_gee_folder_path(ee_module: Any, folder_path: str) -> None: + if "/assets" not in folder_path: + raise ValueError(f"Invalid Earth Engine asset path: {folder_path}") + prefix, suffix = folder_path.rstrip("/").split("/assets", 1) + current = f"{prefix}/assets" + for part in [segment for segment in suffix.split("/") if segment]: + current = f"{current}/{part}" + if gee_asset_exists(ee_module, current): + continue + log.info("Creating Earth Engine folder: %s", current) + try: + ee_module.data.createAsset({"type": "FOLDER"}, current) + except Exception: + # Parallel uploads can race while creating the shared folder path. + if not gee_asset_exists(ee_module, current): + raise + time.sleep(1) + + +def upload_file_to_gcs(local_path: Path, credentials: Any, key: dict[str, Any], bucket_name: str, blob_name: str, chunk_size_mb: int) -> str: + from google.cloud import storage + + client = storage.Client(project=key.get("project_id"), credentials=credentials) + bucket = client.bucket(bucket_name) + blob = bucket.blob(blob_name) + blob.chunk_size = max(256 * 1024, int(chunk_size_mb) * 1024 * 1024) + log.info("Uploading %s to gs://%s/%s", local_path, bucket_name, blob_name) + blob.upload_from_filename(str(local_path)) + blob.reload() + return f"gs://{bucket_name}/{blob_name}" + + +def delete_gcs_blob(credentials: Any, key: dict[str, Any], bucket_name: str, blob_name: str) -> bool: + from google.api_core.exceptions import Forbidden, NotFound + from google.cloud import storage + + bucket = storage.Client(project=key.get("project_id"), credentials=credentials).bucket(bucket_name) + try: + bucket.blob(blob_name).delete() + return True + except NotFound: + return False + except Forbidden as exc: + log.warning("Could not delete gs://%s/%s: %s", bucket_name, blob_name, getattr(exc, "message", exc)) + return False + + +def ingestion_manifest(asset_id: str, gcs_uri: str, geometry_column: str, upload_cfg: dict[str, Any], properties: dict[str, Any]) -> dict[str, Any]: + source = { + "uris": [gcs_uri], + "charset": "UTF-8", + "primaryGeometryColumn": geometry_column, + "maxErrorMeters": float(upload_cfg.get("max_error_meters", 1.0)), + "crs": upload_cfg.get("crs", "EPSG:4326"), + "geodesic": bool(upload_cfg.get("geodesic", False)), + "csvDelimiter": upload_cfg.get("csv_delimiter", "\t"), + "csvQualifier": upload_cfg.get("csv_qualifier", '"'), + } + if int(upload_cfg.get("max_vertices", 0) or 0): + source["maxVertices"] = int(upload_cfg["max_vertices"]) + return {"name": asset_id, "sources": [source], "properties": properties} + + +def start_table_ingestion(ee_module: Any, manifest: dict[str, Any]) -> dict[str, Any]: + task_id = None + if hasattr(ee_module.data, "newTaskId"): + generated = ee_module.data.newTaskId() + task_id = generated[0] if isinstance(generated, (list, tuple)) else generated + response = ee_module.data.startTableIngestion(task_id, manifest) + operation_name = response if isinstance(response, str) else response.get("name") or response.get("id") + if not task_id and operation_name: + task_id = operation_name.rsplit("/", 1)[-1] + return {"task_id": task_id, "operation_name": operation_name, "raw_response": response} + + +def fetch_task_status(ee_module: Any, task_id: str | None, operation_name: str | None) -> dict[str, Any]: + operations = ee_module.data.listOperations() + if isinstance(operations, dict): + operations = operations.get("operations", []) + for operation in operations or []: + name = operation.get("name") + short_id = name.rsplit("/", 1)[-1] if name else None + if (operation_name and name == operation_name) or (task_id and short_id == task_id): + metadata = operation.get("metadata") or {} + return { + "task_id": short_id, + "operation_name": name, + "state": metadata.get("state"), + "metadata": metadata, + "error": operation.get("error"), + "error_message": (operation.get("error") or {}).get("message"), + } + return {"task_id": task_id, "operation_name": operation_name, "state": None} + + +def wait_for_task(ee_module: Any, task_id: str | None, operation_name: str | None, poll_interval: int, timeout: int | None) -> dict[str, Any]: + started = time.monotonic() + while True: + status = fetch_task_status(ee_module, task_id, operation_name) + log.info("GEE task status: %s", status) + if status.get("state") in TERMINAL_TASK_STATES: + return status + if timeout and (time.monotonic() - started) > timeout: + raise TimeoutError(f"Timed out waiting for GEE task {task_id or operation_name}") + time.sleep(max(1, poll_interval)) + + +def make_asset_public(ee_module: Any, asset_id: str) -> bool: + try: + acl = ee_module.data.getAssetAcl(asset_id) + acl["all_users_can_read"] = True + ee_module.data.setAssetAcl(asset_id, acl) + return bool(ee_module.data.getAssetAcl(asset_id).get("all_users_can_read")) + except Exception as exc: + log.warning("Could not make %s public: %s", asset_id, exc) + return False + + +def local_asset_file(repo_root: Path, asset_name: str, asset_cfg: dict[str, Any], outputs_cfg: dict[str, Any], output_suffix: str | None) -> Path: + return asset_output_path(repo_root, asset_name, asset_cfg, outputs_cfg, output_suffix) + + +def upload_one_asset( + repo_root: Path, + name: str, + asset_cfg: dict[str, Any], + outputs_cfg: dict[str, Any], + upload_cfg: dict[str, Any], + service_account_json: Path, + args: argparse.Namespace, +) -> dict[str, Any]: + local_file = local_asset_file(repo_root, name, asset_cfg, outputs_cfg, args.output_suffix) + if not local_file.exists(): + raise FileNotFoundError(f"Build output missing for {name}: {local_file}") + asset_id = asset_cfg["asset_id"] + bucket_name = args.gcs_bucket or upload_cfg.get("gcs_bucket", "core_stack") + gcs_extension = str(upload_cfg.get("gcs_object_extension") or ".csv") + if not gcs_extension.startswith("."): + gcs_extension = f".{gcs_extension}" + blob_name = ( + f"{str(upload_cfg.get('gcs_prefix', 'gee/core_stack')).strip('/')}/" + f"{slug(name)}_{uuid.uuid4().hex}{gcs_extension}" + ) + gcs_uri = f"gs://{bucket_name}/{blob_name}" + manifest = ingestion_manifest( + asset_id, + gcs_uri, + asset_cfg.get("geometry_column", "geometry"), + upload_cfg, + { + "corestack_asset": name, + "source_file": local_file.name, + "source_config": str(args.config), + "built_by": "utilities/scripts/gee/core_stack_gee_ingest.py", + }, + ) + manifest_path = repo_path(repo_root, outputs_cfg.get("manifest_dir", "data/gee/core_stack/manifests")) / f"{name}.manifest.json" + write_json(manifest_path, manifest) + if args.dry_run: + return {"asset_id": asset_id, "source_table": str(local_file), "gcs_uri": gcs_uri, "manifest": str(manifest_path), "dry_run": True} + + ee_module, credentials, key = initialize_ee(service_account_json) + ensure_gee_folder_path(ee_module, asset_id.rsplit("/", 1)[0]) + if gee_asset_exists(ee_module, asset_id): + if not args.replace_existing: + raise RuntimeError(f"GEE asset already exists: {asset_id}. Pass --replace-existing to overwrite.") + log.info("Deleting existing GEE asset: %s", asset_id) + ee_module.data.deleteAsset(asset_id) + time.sleep(1) + uploaded_gcs_uri = upload_file_to_gcs( + local_file, + credentials, + key, + bucket_name, + blob_name, + int(upload_cfg.get("chunk_size_mb", 64)), + ) + manifest["sources"][0]["uris"] = [uploaded_gcs_uri] + write_json(manifest_path, manifest) + try: + ingestion = start_table_ingestion(ee_module, manifest) + except Exception: + if args.cleanup_gcs: + delete_gcs_blob(credentials, key, bucket_name, blob_name) + raise + result = { + "asset_id": asset_id, + "source_table": str(local_file), + "gcs_uri": uploaded_gcs_uri, + "manifest": str(manifest_path), + **ingestion, + } + if args.wait or args.make_public or args.cleanup_gcs: + status = wait_for_task(ee_module, ingestion.get("task_id"), ingestion.get("operation_name"), args.poll_interval, args.timeout) + result["final_status"] = status + if status.get("state") not in SUCCESS_TASK_STATES: + raise RuntimeError(f"GEE ingestion failed for {asset_id}: {status}") + if args.make_public: + result["made_public"] = make_asset_public(ee_module, asset_id) + if args.cleanup_gcs: + result["cleaned_gcs"] = delete_gcs_blob(credentials, key, bucket_name, blob_name) + return result + + +def upload_assets(args: argparse.Namespace) -> dict[str, Any]: + config_path, repo_root, config = load_config(args) + outputs_cfg = config.get("outputs") or {} + upload_cfg = config.get("upload") or {} + service_account_json = repo_path(repo_root, args.service_account_json or config["service_account_json"]) + assets = selected_assets(config, args.asset) + summary = { + "uploaded_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), + "config": str(config_path), + "assets": {}, + } + jobs = max(1, int(args.jobs or 1)) + if jobs == 1 or len(assets) <= 1: + for name, asset_cfg in assets: + summary["assets"][name] = upload_one_asset(repo_root, name, asset_cfg, outputs_cfg, upload_cfg, service_account_json, args) + else: + with ThreadPoolExecutor(max_workers=jobs) as executor: + futures = { + executor.submit(upload_one_asset, repo_root, name, asset_cfg, outputs_cfg, upload_cfg, service_account_json, args): name + for name, asset_cfg in assets + } + for future in as_completed(futures): + summary["assets"][futures[future]] = future.result() + summary_path = with_suffix_token(repo_path(repo_root, outputs_cfg.get("upload_summary_yaml", "data/gee/core_stack/core_stack_gee_upload_summary.yaml")), args.output_suffix) + write_yaml(summary_path, summary) + log.info("Wrote upload summary: %s", summary_path) + return summary + + +def task_status(args: argparse.Namespace) -> dict[str, Any]: + config_path, repo_root, config = load_config(args) + outputs_cfg = config.get("outputs") or {} + summary_path = repo_path(repo_root, args.summary or outputs_cfg.get("upload_summary_yaml", "data/gee/core_stack/core_stack_gee_upload_summary.yaml")) + service_account_json = repo_path(repo_root, args.service_account_json or config["service_account_json"]) + ee_module, _, _ = initialize_ee(service_account_json) + summary = read_yaml(summary_path) + statuses = {"checked_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), "assets": {}} + for name, result in (summary.get("assets") or {}).items(): + statuses["assets"][name] = fetch_task_status(ee_module, result.get("task_id"), result.get("operation_name")) + status_path = repo_path( + repo_root, + args.output or outputs_cfg.get("status_summary_yaml", "data/gee/core_stack/core_stack_gee_status.yaml"), + ) + write_yaml(status_path, statuses) + log.info("Wrote task status summary: %s", status_path) + print(yaml.safe_dump(statuses, sort_keys=False)) + return statuses + + +def publish_assets(args: argparse.Namespace) -> dict[str, Any]: + config_path, repo_root, config = load_config(args) + outputs_cfg = config.get("outputs") or {} + service_account_json = repo_path(repo_root, args.service_account_json or config["service_account_json"]) + ee_module, _, _ = initialize_ee(service_account_json) + summary = { + "published_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), + "config": str(config_path), + "assets": {}, + } + for name, asset_cfg in selected_assets(config, args.asset): + asset_id = asset_cfg["asset_id"] + item = {"asset_id": asset_id, "exists": gee_asset_exists(ee_module, asset_id)} + if item["exists"]: + item["made_public"] = make_asset_public(ee_module, asset_id) + else: + item["made_public"] = False + item["error"] = "Asset does not exist yet; wait for ingestion to succeed." + summary["assets"][name] = item + summary_path = repo_path( + repo_root, + args.output or outputs_cfg.get("public_summary_yaml", "data/gee/core_stack/core_stack_gee_public_summary.yaml"), + ) + write_yaml(summary_path, summary) + log.info("Wrote public ACL summary: %s", summary_path) + print(yaml.safe_dump(summary, sort_keys=False)) + return summary + + +def verify_assets(args: argparse.Namespace) -> dict[str, Any]: + config_path, repo_root, config = load_config(args) + outputs_cfg = config.get("outputs") or {} + service_account_json = repo_path(repo_root, args.service_account_json or config["service_account_json"]) + ee_module, _, _ = initialize_ee(service_account_json) + summary = { + "verified_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(), + "config": str(config_path), + "assets": {}, + } + for name, asset_cfg in selected_assets(config, args.asset): + asset_id = asset_cfg["asset_id"] + item: dict[str, Any] = {"asset_id": asset_id} + try: + asset = ee_module.data.getAsset(asset_id) + item["exists"] = True + item["asset_type"] = asset.get("type") + item["name"] = asset.get("name") + collection = ee_module.FeatureCollection(asset_id) + if not args.skip_size: + item["feature_count"] = collection.size().getInfo() + item["property_names"] = collection.first().propertyNames().getInfo() + item["property_count"] = len(item["property_names"]) + except Exception as exc: + item["exists"] = False + item["error"] = str(exc) + summary["assets"][name] = item + summary_path = repo_path( + repo_root, + args.output or outputs_cfg.get("verify_summary_yaml", "data/gee/core_stack/core_stack_gee_verify_summary.yaml"), + ) + write_yaml(summary_path, summary) + log.info("Wrote GEE verify summary: %s", summary_path) + print(yaml.safe_dump(summary, sort_keys=False)) + return summary + + +def inspect_assets(args: argparse.Namespace) -> dict[str, Any]: + _, repo_root, config = load_config(args) + outputs_cfg = config.get("outputs") or {} + out = {"assets": {}} + for name, asset_cfg in selected_assets(config, args.asset): + source = repo_path(repo_root, asset_cfg["source"]) + output = local_asset_file(repo_root, name, asset_cfg, outputs_cfg, args.output_suffix) + item = { + "asset_id": asset_cfg["asset_id"], + "kind": asset_cfg.get("kind", "feature_layer"), + "source": str(source), + "source_exists": source.exists(), + "source_size_bytes": source.stat().st_size if source.exists() else None, + "output": str(output), + "output_exists": output.exists(), + "output_size_bytes": output.stat().st_size if output.exists() else None, + } + if source.exists() and asset_cfg.get("format", "gpkg") == "gpkg" and asset_cfg.get("layer"): + with sqlite3.connect(source) as con: + item["layer_rows"] = con.execute(f"SELECT COUNT(*) FROM {sql_ident(asset_cfg['layer'])}").fetchone()[0] + item["geometry_columns"] = [ + list(row) + for row in con.execute("SELECT table_name, column_name, geometry_type_name, srs_id FROM gpkg_geometry_columns") + ] + out["assets"][name] = item + print(yaml.safe_dump(out, sort_keys=False)) + return out + + +def add_common_build_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--asset", default="all") + parser.add_argument("--overwrite", action="store_true") + parser.add_argument("--chunk-size", type=int, default=25_000) + parser.add_argument("--limit", type=int, default=None, help="Smoke-test limit per asset.") + parser.add_argument("--jobs", type=int, default=1) + parser.add_argument("--output-suffix", default=None) + parser.add_argument("--max-rss-mb", type=int, default=DEFAULT_MAX_RSS_MB) + + +def add_common_upload_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--asset", default="all") + parser.add_argument("--service-account-json", default=None) + parser.add_argument("--gcs-bucket", default=None) + parser.add_argument("--jobs", type=int, default=1) + parser.add_argument("--output-suffix", default=None) + parser.add_argument("--replace-existing", action="store_true") + parser.add_argument("--wait", action="store_true") + parser.add_argument("--make-public", action="store_true") + parser.add_argument("--cleanup-gcs", action="store_true") + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--poll-interval", type=int, default=60) + parser.add_argument("--timeout", type=int, default=None) + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG) + parser.add_argument("--debug", action="store_true") + sub = parser.add_subparsers(dest="command", required=True) + + inspect_cmd = sub.add_parser("inspect", help="Inspect configured sources and local build outputs.") + inspect_cmd.add_argument("--asset", default="all") + inspect_cmd.add_argument("--output-suffix", default=None) + inspect_cmd.set_defaults(func=inspect_assets) + + build_cmd = sub.add_parser("build", help="Build local GEE ingestion tables.") + add_common_build_args(build_cmd) + build_cmd.set_defaults(func=build_assets) + + upload_cmd = sub.add_parser("upload", help="Stage local tables to GCS and start GEE table ingestion.") + add_common_upload_args(upload_cmd) + upload_cmd.set_defaults(func=upload_assets) + + all_cmd = sub.add_parser("all", help="Build local tables, then upload them.") + all_cmd.add_argument("--asset", default="all") + all_cmd.add_argument("--overwrite", action="store_true") + all_cmd.add_argument("--chunk-size", type=int, default=25_000) + all_cmd.add_argument("--limit", type=int, default=None, help="Smoke-test limit per asset.") + all_cmd.add_argument("--jobs", type=int, default=1) + all_cmd.add_argument("--output-suffix", default=None) + all_cmd.add_argument("--max-rss-mb", type=int, default=DEFAULT_MAX_RSS_MB) + all_cmd.add_argument("--service-account-json", default=None) + all_cmd.add_argument("--gcs-bucket", default=None) + all_cmd.add_argument("--replace-existing", action="store_true") + all_cmd.add_argument("--wait", action="store_true") + all_cmd.add_argument("--make-public", action="store_true") + all_cmd.add_argument("--cleanup-gcs", action="store_true") + all_cmd.add_argument("--dry-run", action="store_true") + all_cmd.add_argument("--poll-interval", type=int, default=60) + all_cmd.add_argument("--timeout", type=int, default=None) + all_cmd.set_defaults(func=lambda args: (build_assets(args), upload_assets(args))) + + status_cmd = sub.add_parser("status", help="Check task statuses from an upload summary YAML.") + status_cmd.add_argument("--summary", default=None) + status_cmd.add_argument("--service-account-json", default=None) + status_cmd.add_argument("--output", default=None) + status_cmd.set_defaults(func=task_status) + + public_cmd = sub.add_parser("make-public", help="Make existing configured GEE assets public.") + public_cmd.add_argument("--asset", default="all") + public_cmd.add_argument("--service-account-json", default=None) + public_cmd.add_argument("--output", default=None) + public_cmd.set_defaults(func=publish_assets) + + verify_cmd = sub.add_parser("verify", help="Verify existing GEE assets and sample schemas.") + verify_cmd.add_argument("--asset", default="all") + verify_cmd.add_argument("--service-account-json", default=None) + verify_cmd.add_argument("--output", default=None) + verify_cmd.add_argument("--skip-size", action="store_true") + verify_cmd.set_defaults(func=verify_assets) + + return parser + + +def main(argv: Sequence[str] | None = None) -> None: + args = build_parser().parse_args(argv) + config_path = args.config.resolve() + repo_root = find_repo_root(config_path) + config = read_yaml(config_path) if config_path.exists() else {} + log_file = repo_path(repo_root, (config.get("outputs") or {}).get("log_file", "logs/core_stack_gee_ingest.log")) + setup_logging(args.debug, log_file=log_file) + args.func(args) + + +if __name__ == "__main__": + main() From cfb84b7c7309eefbec02cda261d8b15e6fa19f90 Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:20 +0000 Subject: [PATCH 2/3] Configure Core Stack GEE assets This YAML declares the local source files, Earth Engine asset ids, layer names, geometry policy, and column handling for the core CS assets. Tracking it makes GEE uploads reproducible without committing generated staging files. --- .../scripts/gee/core_stack_gee_assets.yaml | 122 ++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 utilities/scripts/gee/core_stack_gee_assets.yaml diff --git a/utilities/scripts/gee/core_stack_gee_assets.yaml b/utilities/scripts/gee/core_stack_gee_assets.yaml new file mode 100644 index 00000000..958e76d2 --- /dev/null +++ b/utilities/scripts/gee/core_stack_gee_assets.yaml @@ -0,0 +1,122 @@ +schema_version: 1 +description: > + Config-driven Core Stack Earth Engine ingestion for standard GeoPackage and + GeoJSON vector assets. Local exports are tab-delimited tables with one + GeoJSON geometry column, then staged to GCS with a .csv suffix for Earth + Engine table ingestion. + +service_account_json: data/gee_confs/core-stack-learn-818963fa8f26.json + +outputs: + directory: data/gee/core_stack + local_extension: .tsv + build_summary_yaml: data/gee/core_stack/core_stack_gee_build_summary.yaml + upload_summary_yaml: data/gee/core_stack/core_stack_gee_upload_summary.yaml + status_summary_yaml: data/gee/core_stack/core_stack_gee_status.yaml + public_summary_yaml: data/gee/core_stack/core_stack_gee_public_summary.yaml + verify_summary_yaml: data/gee/core_stack/core_stack_gee_verify_summary.yaml + manifest_dir: data/gee/core_stack/manifests + log_file: logs/core_stack_gee_ingest.log + +upload: + gcs_bucket: core_stack + gcs_prefix: gee/core_stack_assets + gcs_object_extension: .csv + csv_delimiter: "\t" + csv_qualifier: '"' + crs: EPSG:4326 + geodesic: false + max_vertices: 0 + max_error_meters: 1.0 + chunk_size_mb: 64 + +assets: + cs_admin_standard: + asset_id: projects/corestack-datasets/assets/datasets/cs_admin_standard + kind: feature_layer + format: gpkg + source: data/admin-boundary/cs_admin_standard.gpkg + layer: cs_admin_standard + geometry_column: geometry + properties: + include: "*" + + cs_village_facility_proximity: + asset_id: projects/corestack-datasets/assets/datasets/cs_village_facility_proximity + kind: village_proximity_wide + source: data/facilities/outputs/cs_village_facility_proximity.gpkg + geometry_column: geometry + admin_columns: + cs_feature_id: cs_feature_id + state_name: state_name + district_name: district_name + TEHSIL: tehsil_name + pc11_village_id: village_id + NAME: village_name + include_l3: + distance_template: l3_{class_name}_distance_km + facility_uid_template: l3_{class_name}_facility_uid + include_l2: + distance_template: l2_{group_name}_distance_km + facility_uid_template: l2_{group_name}_facility_uid + selected_l3_template: l2_{group_name}_selected_l3 + include_l1_summary: + enabled: true + nearest_distance_template: l1_{domain_name}_nearest_l2_distance_km + nearest_l2_template: l1_{domain_name}_nearest_l2_group + + cs_pan_india_facilities: + asset_id: projects/corestack-datasets/assets/datasets/cs_pan_india_facilities + kind: facility_membership_points + source: data/facilities/outputs/pan_india_facilities.gpkg + geometry_column: geometry + columns: + facility: + - facility_uid + - facility_name + - facility_code + - source: latitude + target: facility_latitude + - source: longitude + target: facility_longitude + - urban_rural + - pincode + - establishment_year + - district_lgd + - village_census11 + - village_name + membership: + - class_l1_domain + - class_l2_filter_group + - class_l3_facility_class + - class_l4_facility_subtype + - class_k1 + - class_k2 + - class_k3 + - class_k4 + - class_k5 + - class_k6 + - class_k7 + - class_k8 + generated: + - facility_membership_uid + + cs_village_livestock_census_20: + asset_id: projects/corestack-datasets/assets/datasets/cs_village_livestock_census_20 + kind: feature_layer + format: gpkg + source: data/livestock/cs_village_livestock_census_20.gpkg + layer: cs_village_livestock_census_20 + geometry_column: geometry + properties: + include: "*" + + cs_antyodaya_2020: + asset_id: projects/corestack-datasets/assets/datasets/cs_antyodaya_2020 + kind: feature_layer + format: gpkg + source: data/antyodaya/output/cs_antyodaya_2020.gpkg + layer: cs_antyodaya_2020 + geometry_column: geometry + properties: + include: "*" From e3893adca0ecbcc43c01bad23629307cf7fc6dbf Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:21 +0000 Subject: [PATCH 3/3] Document Core Stack GEE ingestion This document records how to build, upload, monitor, verify, and publish the standard Core Stack GEE assets. It is included with the uploader so operators can rerun the flow without relying on local shell history. --- .../scripts/gee/core_stack_gee_ingest.md | 114 ++++++++++++++++++ 1 file changed, 114 insertions(+) create mode 100644 utilities/scripts/gee/core_stack_gee_ingest.md diff --git a/utilities/scripts/gee/core_stack_gee_ingest.md b/utilities/scripts/gee/core_stack_gee_ingest.md new file mode 100644 index 00000000..41f165ed --- /dev/null +++ b/utilities/scripts/gee/core_stack_gee_ingest.md @@ -0,0 +1,114 @@ +# Core Stack GEE Ingestion + +This is the standard ingestion path for large Core Stack GeoPackage and +GeoJSON vector assets. + +The script writes a local tab-delimited table with one GeoJSON geometry column, +stages the same bytes to GCS with a `.csv` object suffix, then starts an Earth +Engine table-ingestion task from a manifest. + +## Files + +- Script: `utilities/scripts/gee/core_stack_gee_ingest.py` +- Config: `utilities/scripts/gee/core_stack_gee_assets.yaml` +- Service account: `data/gee_confs/core-stack-learn-818963fa8f26.json` +- Local outputs: `data/gee/core_stack/` +- Manifests: `data/gee/core_stack/manifests/` +- Log: `logs/core_stack_gee_ingest.log` + +## Configured Assets + +- `cs_admin_standard` +- `cs_village_facility_proximity` +- `cs_pan_india_facilities` +- `cs_village_livestock_census_20` +- `cs_antyodaya_2020` + +`cs_pan_india_facilities` currently uses the existing source file +`data/facilities/outputs/pan_india_facilities.gpkg`; it is named with the `cs_` +asset key in the GEE config without making a duplicate 2 GB GeoPackage. + +## Commands + +Inspect configured sources: + +```bash +uv run --with pyyaml --with shapely \ + python utilities/scripts/gee/core_stack_gee_ingest.py inspect --asset all +``` + +Smoke-build all configured assets: + +```bash +uv run --with pyyaml --with shapely \ + python utilities/scripts/gee/core_stack_gee_ingest.py build \ + --asset all --limit 10 --jobs 2 --output-suffix .smoke \ + --overwrite --max-rss-mb 5000 +``` + +Dry-run manifests for the smoke outputs: + +```bash +uv run --with pyyaml --with shapely \ + python utilities/scripts/gee/core_stack_gee_ingest.py upload \ + --asset all --output-suffix .smoke --dry-run --jobs 2 +``` + +Build full local ingestion tables: + +```bash +nohup uv run --with pyyaml --with shapely \ + python utilities/scripts/gee/core_stack_gee_ingest.py build \ + --asset all --jobs 2 --overwrite --max-rss-mb 5000 \ + > logs/core_stack_gee_build_full_$(date -u +%Y%m%dT%H%M%SZ).log 2>&1 & +``` + +Start GEE ingestion after full tables are built: + +```bash +nohup uv run --with earthengine-api --with google-cloud-storage --with pyyaml --with shapely \ + python utilities/scripts/gee/core_stack_gee_ingest.py upload \ + --asset all --jobs 2 --replace-existing \ + > logs/core_stack_gee_upload_$(date -u +%Y%m%dT%H%M%SZ).log 2>&1 & +``` + +Check task status later: + +```bash +uv run --with earthengine-api --with google-cloud-storage --with pyyaml \ + python utilities/scripts/gee/core_stack_gee_ingest.py status +``` + +The status command also writes `data/gee/core_stack/core_stack_gee_status.yaml` +by default. + +Make existing ingested assets public after tasks succeed: + +```bash +uv run --with earthengine-api --with google-cloud-storage --with pyyaml \ + python utilities/scripts/gee/core_stack_gee_ingest.py make-public --asset all +``` + +Verify completed GEE assets and sample schemas: + +```bash +uv run --with earthengine-api --with google-cloud-storage --with pyyaml \ + python utilities/scripts/gee/core_stack_gee_ingest.py verify --asset all +``` + +## Notes + +- The runtime memory guard defaults to `5000` MB and is checked at chunk + boundaries. +- Special exporters accept either plain column names or `{source, target}` + column mappings in YAML. The facilities point asset aliases raw + `latitude`/`longitude` to `facility_latitude`/`facility_longitude` so Earth + Engine keeps them as ordinary properties while still using the GeoJSON + geometry column. +- For large GeoJSON FeatureCollections, install `ijson` or convert to GeoJSONL: + `uv run --with ijson --with pyyaml --with shapely ...`. +- `upload --make-public` is best-effort and waits for ingestion to finish + before applying ACLs. For long overnight uploads, submit tasks first, monitor + them, then use `make-public` or make assets public manually in Earth Engine if + IAM blocks the script. +- `verify` writes `data/gee/core_stack/core_stack_gee_verify_summary.yaml`.