From 131833f3c727b3ca04dcf9acccf175294b43a08e Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:25 +0000 Subject: [PATCH 1/7] Add facilities pipeline shared utilities Shared helpers keep path resolution, schema constants, CSV/GPKG writing, taxonomy handling, and QA metadata consistent across the facilities build scripts. --- .../facilities_utils/facility_utils.py | 251 ++++++++++++++++++ 1 file changed, 251 insertions(+) create mode 100644 utilities/scripts/facilities_utils/facility_utils.py diff --git a/utilities/scripts/facilities_utils/facility_utils.py b/utilities/scripts/facilities_utils/facility_utils.py new file mode 100644 index 00000000..ea70b5db --- /dev/null +++ b/utilities/scripts/facilities_utils/facility_utils.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Shared helpers for the facilities pipeline.""" + +from __future__ import annotations + +import hashlib +import json +import logging +import math +import re +import sqlite3 +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + +import pandas as pd + + +CLASS_K_COUNT = 8 +CLASS_K_COLUMNS = [f"class_k{i}" for i in range(1, CLASS_K_COUNT + 1)] + +LOG_FORMAT = "%(asctime)s [%(levelname)-7s] %(message)s" + + +def setup_logging(debug: bool = False, log_path: Optional[Path] = None) -> logging.Logger: + level = logging.DEBUG if debug else logging.INFO + handlers: List[logging.Handler] = [logging.StreamHandler()] + if log_path: + log_path.parent.mkdir(parents=True, exist_ok=True) + handlers.append(logging.FileHandler(log_path, encoding="utf-8")) + logging.basicConfig(level=level, format=LOG_FORMAT, datefmt="%H:%M:%S", handlers=handlers, force=True) + return logging.getLogger("facilities") + + +def find_repo_root(start: Path) -> Path: + for candidate in [start, *start.parents]: + if (candidate / "data").is_dir() and (candidate / "utilities").is_dir(): + return candidate + raise RuntimeError(f"Could not find repo root from {start}") + + +def resolve_path(value: str | Path, repo_root: Path) -> Path: + path = Path(value) + return path if path.is_absolute() else repo_root / path + + +def read_yaml(path: Path) -> Dict[str, Any]: + try: + import yaml + except ImportError as exc: + raise RuntimeError("PyYAML is required. Run with: uv run --with pyyaml ...") from exc + return yaml.safe_load(path.read_text(encoding="utf-8")) or {} + + +def write_yaml(path: Path, data: Dict[str, Any]) -> None: + try: + import yaml + except ImportError as exc: + raise RuntimeError("PyYAML is required. Run with: uv run --with pyyaml ...") from exc + + class Dumper(yaml.SafeDumper): + def ignore_aliases(self, data: Any) -> bool: + return True + + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(yaml.dump(data, Dumper=Dumper, sort_keys=False, allow_unicode=True, width=120), encoding="utf-8") + + +def slugify(value: Any) -> str: + text = "" if pd.isna(value) else str(value) + text = re.sub(r"[^A-Za-z0-9]+", "_", text.strip().lower()) + return re.sub(r"_+", "_", text).strip("_") + + +def clean_string_series(series: pd.Series) -> pd.Series: + return ( + series.astype("string") + .str.strip() + .replace({"": pd.NA, "nan": pd.NA, "None": pd.NA, "": pd.NA, "NaN": pd.NA}) + ) + + +TITLE_STOPWORDS = {"of", "the", "and", "in", "for", "at", "to", "a", "an", "by", "on", "or", "as", "with"} +KNOWN_ABBREVIATIONS = { + "IIT", + "IIM", + "IIIT", + "NIT", + "AIIMS", + "CBSE", + "ICSE", + "SSC", + "HSC", + "KV", + "JNV", + "DAV", + "PG", + "ITI", +} + + +def clean_text_value(value: Any) -> Any: + if pd.isna(value): + return pd.NA + text = re.sub(r"\s+", " ", str(value).strip()) + text = re.sub(r"^\d+[\.\)\-\s]+", "", text).strip() + if not text: + return pd.NA + words = [] + for index, word in enumerate(text.split(" ")): + stripped = word.strip() + bare = stripped.strip(".,;:!?()[]{}'\"") + if not bare: + words.append(stripped) + continue + if bare.upper() in KNOWN_ABBREVIATIONS: + words.append(stripped.replace(bare, bare.upper())) + elif index > 0 and bare.lower() in TITLE_STOPWORDS: + words.append(bare.lower()) + elif bare.isupper() and len(bare) <= 4: + words.append(bare) + else: + words.append(bare[:1].upper() + bare[1:].lower()) + return " ".join(words) + + +def clean_text_series(series: pd.Series) -> pd.Series: + return series.map(clean_text_value).astype("string") + + +def enforce_numeric(series: pd.Series) -> pd.Series: + return pd.to_numeric(series, errors="coerce") + + +def validate_pincode(series: pd.Series) -> pd.Series: + values = pd.to_numeric(series, errors="coerce") + return values.where(values.between(100000, 999999)).astype("Int64") + + +def validate_year(series: pd.Series) -> pd.Series: + values = pd.to_numeric(series, errors="coerce") + return values.where(values.between(1800, 2035)).astype("Int64") + + +def clean_identifier(series: pd.Series) -> pd.Series: + values = clean_string_series(series) + decimal_int = values.str.match(r"^-?\d+\.0+$", na=False) + values.loc[decimal_int] = values.loc[decimal_int].str.replace(r"\.0+$", "", regex=True) + return values + + +def parse_coordinate_pair(value: Any) -> Tuple[float, float]: + if pd.isna(value): + return (math.nan, math.nan) + numbers = re.findall(r"-?\d+(?:\.\d+)?", str(value)) + if len(numbers) < 2: + return (math.nan, math.nan) + lon = float(numbers[0]) + lat = float(numbers[1]) + return lon, lat + + +def coordinate_status(lat: pd.Series, lon: pd.Series) -> pd.Series: + valid = ( + lat.notna() + & lon.notna() + & lat.between(6.0, 38.8) + & lon.between(68.0, 98.5) + & (lat != 0) + & (lon != 0) + ) + return pd.Series(valid.map({True: "valid", False: "invalid_coordinate"}), index=lat.index, dtype="string") + + +def coordinate_key(series: pd.Series) -> pd.Series: + numeric = pd.to_numeric(series, errors="coerce").round(7) + return numeric.map(lambda value: "na" if pd.isna(value) else f"{value:.7f}").astype("string") + + +def file_fingerprint(paths: Sequence[Path]) -> str: + payload = [] + for path in paths: + stat = path.stat() + payload.append({"path": str(path), "size": stat.st_size, "mtime_ns": stat.st_mtime_ns}) + text = json.dumps(payload, sort_keys=True) + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def read_csv_selected(path: Path, usecols: Optional[Sequence[str]] = None, **kwargs: Any) -> pd.DataFrame: + if usecols: + header = pd.read_csv(path, nrows=0) + available = set(header.columns) + missing = [column for column in usecols if column not in available] + if missing: + logging.getLogger("facilities").warning("Missing columns in %s: %s", path.name, ", ".join(missing)) + usecols = [column for column in usecols if column in available] + return pd.read_csv(path, usecols=usecols, low_memory=False, **kwargs) + + +def taxonomy_rows(config: Dict[str, Any]) -> List[Dict[str, Any]]: + rows: List[Dict[str, Any]] = [] + filter_logic = config.get("filter_logic") or {} + order = 0 + for domain, groups in (config.get("taxonomy") or {}).items(): + for filter_group, classes in (groups or {}).items(): + logic = (filter_logic.get(domain) or {}).get(filter_group, "direct") + for facility_class, spec in (classes or {}).items(): + subtypes = (spec or {}).get("subtypes") or [] + rows.append( + { + "class_l1_domain": domain, + "class_l2_filter_group": filter_group, + "class_l3_facility_class": facility_class, + "configured_subtypes": ";".join(subtypes), + "filter_logic": logic, + "sort_order": order, + } + ) + order += 1 + return rows + + +def taxonomy_lookup(config: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: + return {row["class_l3_facility_class"]: row for row in taxonomy_rows(config)} + + +def add_attribute_table_to_gpkg(gpkg_path: Path, table_name: str, df: pd.DataFrame) -> None: + if df.empty: + return + with sqlite3.connect(gpkg_path) as con: + df.to_sql(table_name, con, if_exists="replace", index=False, chunksize=100_000) + con.execute( + """ + insert or replace into gpkg_contents + (table_name, data_type, identifier, description, last_change, min_x, min_y, max_x, max_y, srs_id) + values (?, 'attributes', ?, '', strftime('%Y-%m-%dT%H:%M:%fZ','now'), null, null, null, null, null) + """, + (table_name, table_name), + ) + + +def ensure_sqlite_indexes(gpkg_path: Path) -> None: + with sqlite3.connect(gpkg_path) as con: + con.execute("create index if not exists idx_facilities_uid on facilities(facility_uid)") + con.execute("create index if not exists idx_membership_uid on facility_memberships(facility_uid)") + con.execute( + """ + create index if not exists idx_membership_classes + on facility_memberships(class_l1_domain, class_l2_filter_group, class_l3_facility_class, class_l4_facility_subtype) + """ + ) + con.commit() From b3da5abbdeb84a0be01a11a2378f779da6fd1899 Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:26 +0000 Subject: [PATCH 2/7] Add facilities raw source cleaners The cleaner module contains the source-specific and generic raw-file transformations needed to standardize raw facility files before building pan_india_facilities.gpkg. --- .../facilities_utils/facility_cleaners.py | 457 ++++++++++++++++++ 1 file changed, 457 insertions(+) create mode 100644 utilities/scripts/facilities_utils/facility_cleaners.py diff --git a/utilities/scripts/facilities_utils/facility_cleaners.py b/utilities/scripts/facilities_utils/facility_cleaners.py new file mode 100644 index 00000000..e699ae3f --- /dev/null +++ b/utilities/scripts/facilities_utils/facility_cleaners.py @@ -0,0 +1,457 @@ +#!/usr/bin/env python3 +"""Raw-source processors for the facilities pipeline.""" + +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Any, Dict, Iterable, List, Optional, Tuple + +import pandas as pd + +from facility_utils import ( + CLASS_K_COLUMNS, + clean_identifier, + clean_string_series, + clean_text_series, + coordinate_key, + coordinate_status, + enforce_numeric, + parse_coordinate_pair, + read_csv_selected, + resolve_path, + slugify, + taxonomy_lookup, + validate_pincode, + validate_year, +) + + +log = logging.getLogger("facilities") + +def source_paths(repo_root: Path, config: Dict[str, Any], source_cfg: Dict[str, Any]) -> List[Path]: + raw_dir = resolve_path(config["pipeline"]["raw_dir"], repo_root) + return [raw_dir / name for name in source_cfg.get("raw_files") or []] + + +def load_pincode_centroids(repo_root: Path, config: Dict[str, Any]) -> Dict[int, Tuple[float, float]]: + path = resolve_path(config["references"]["pincode_centroids"], repo_root) + if not path.exists(): + return {} + df = read_csv_selected(path, usecols=["pin_code", "pin_lat", "pin_long"]) + df = df.dropna(subset=["pin_code", "pin_lat", "pin_long"]) + return { + int(row.pin_code): (float(row.pin_lat), float(row.pin_long)) + for row in df.itertuples(index=False) + } + + +def fill_coords_from_pincode(df: pd.DataFrame, lat_col: str, lon_col: str, pin_col: str, centroids: Dict[int, Tuple[float, float]]) -> pd.DataFrame: + if not centroids or pin_col not in df.columns: + return df + bad = df[lat_col].isna() | df[lon_col].isna() | (df[lat_col] == 0) | (df[lon_col] == 0) + if not bad.any(): + return df + pins = pd.to_numeric(df.loc[bad, pin_col], errors="coerce") + filled = 0 + for idx, pin in pins.dropna().items(): + pin_int = int(pin) + if pin_int in centroids: + df.at[idx, lat_col] = centroids[pin_int][0] + df.at[idx, lon_col] = centroids[pin_int][1] + filled += 1 + if filled: + log.info("Filled %d coordinates from pincode centroids", filled) + return df + + +def required_columns_for_table(source_cfg: Dict[str, Any]) -> List[str]: + columns = [ + source_cfg.get("id_col"), + source_cfg.get("name_col"), + source_cfg.get("lat_col"), + source_cfg.get("lon_col"), + source_cfg.get("subtype_col"), + source_cfg.get("facility_code_col"), + source_cfg.get("urban_rural_col"), + source_cfg.get("pincode_col"), + source_cfg.get("establishment_year_col"), + source_cfg.get("district_lgd_col"), + source_cfg.get("village_census11_col"), + source_cfg.get("village_name_col"), + source_cfg.get("coord_fallback_col"), + ] + for item in (source_cfg.get("class_k") or {}).values(): + columns.append((item or {}).get("source")) + columns.extend(source_cfg.get("clean_text_cols") or []) + return [column for column in dict.fromkeys(columns) if column] + + +def add_class_k_fields(out: pd.DataFrame, raw: pd.DataFrame, source_cfg: Dict[str, Any]) -> None: + for class_k in CLASS_K_COLUMNS: + spec = (source_cfg.get("class_k") or {}).get(class_k) or {} + source_col = spec.get("source") + out[class_k] = clean_string_series(raw[source_col]) if source_col and source_col in raw.columns else pd.NA + + +def standard_facility_frame( + raw: pd.DataFrame, + source_cfg: Dict[str, Any], + facility_class: str, + tax: Dict[str, Any], + subtype: Optional[pd.Series] = None, +) -> Tuple[pd.DataFrame, pd.DataFrame]: + id_col = source_cfg["id_col"] + lat_col = source_cfg["lat_col"] + lon_col = source_cfg["lon_col"] + source_id = clean_identifier(raw[id_col]) if id_col in raw.columns else pd.Series(pd.NA, index=raw.index, dtype="string") + missing_id = source_id.isna() + if missing_id.any(): + source_id = source_id.fillna("__row_" + raw.index.astype(str).astype("string")) + + lat = pd.to_numeric(raw[lat_col], errors="coerce") if lat_col in raw.columns else pd.Series(pd.NA, index=raw.index) + lon = pd.to_numeric(raw[lon_col], errors="coerce") if lon_col in raw.columns else pd.Series(pd.NA, index=raw.index) + coord_col = source_cfg.get("coord_fallback_col") + if coord_col and coord_col in raw.columns: + missing = lat.isna() | lon.isna() | (lat == 0) | (lon == 0) + if missing.any(): + parsed = raw.loc[missing, coord_col].map(parse_coordinate_pair) + lon.loc[missing] = parsed.map(lambda item: item[0]) + lat.loc[missing] = parsed.map(lambda item: item[1]) + + uid = source_cfg["facility_uid_prefix"] + ":" + source_id.astype(str) + if source_cfg.get("uid_strategy") == "id_coord": + uid = uid + ":" + coordinate_key(lat) + ":" + coordinate_key(lon) + + out = pd.DataFrame(index=raw.index) + out["facility_uid"] = uid + out["source_id"] = source_id + out["latitude"] = lat + out["longitude"] = lon + out["facility_name"] = clean_string_series(raw[source_cfg["name_col"]]) if source_cfg.get("name_col") in raw.columns else pd.NA + for col in source_cfg.get("clean_text_cols") or []: + if col == source_cfg.get("name_col") and "facility_name" in out: + out["facility_name"] = clean_text_series(raw[col]) + out["facility_code"] = clean_string_series(raw[source_cfg["facility_code_col"]]) if source_cfg.get("facility_code_col") in raw.columns else pd.NA + out["class_l1_domain"] = tax["class_l1_domain"] + out["class_l2_filter_group"] = tax["class_l2_filter_group"] + out["class_l3_facility_class"] = facility_class + if subtype is not None: + out["class_l4_facility_subtype"] = clean_string_series(subtype) + elif source_cfg.get("subtype_col") in raw.columns: + out["class_l4_facility_subtype"] = clean_string_series(raw[source_cfg["subtype_col"]]) + else: + out["class_l4_facility_subtype"] = pd.NA + add_class_k_fields(out, raw, source_cfg) + out["urban_rural"] = clean_string_series(raw[source_cfg["urban_rural_col"]]) if source_cfg.get("urban_rural_col") in raw.columns else pd.NA + out["pincode"] = validate_pincode(raw[source_cfg["pincode_col"]]) if source_cfg.get("pincode_col") in raw.columns else pd.NA + out["establishment_year"] = validate_year(raw[source_cfg["establishment_year_col"]]) if source_cfg.get("establishment_year_col") in raw.columns else pd.NA + out["district_lgd"] = enforce_numeric(raw[source_cfg["district_lgd_col"]]).astype("Int64") if source_cfg.get("district_lgd_col") in raw.columns else pd.NA + out["village_census11"] = clean_identifier(raw[source_cfg["village_census11_col"]]) if source_cfg.get("village_census11_col") in raw.columns else pd.NA + out["village_name"] = clean_string_series(raw[source_cfg["village_name_col"]]) if source_cfg.get("village_name_col") in raw.columns else pd.NA + out["coordinate_status"] = coordinate_status(out["latitude"], out["longitude"]) + return out, membership_from_facilities(out) + + +def membership_from_facilities(facilities: pd.DataFrame) -> pd.DataFrame: + columns = [ + "facility_uid", + "class_l1_domain", + "class_l2_filter_group", + "class_l3_facility_class", + "class_l4_facility_subtype", + *CLASS_K_COLUMNS, + ] + return facilities[columns].copy() + + +def append_csv(df: pd.DataFrame, path: Path, header: bool) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + df.to_csv(path, mode="a", index=False, header=header) + + +def split_valid_invalid(facilities: pd.DataFrame, memberships: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: + valid_mask = facilities["coordinate_status"] == "valid" + invalid = facilities.loc[~valid_mask].copy() + valid = facilities.loc[valid_mask].copy() + memberships = memberships[memberships["facility_uid"].isin(set(valid["facility_uid"]))].copy() + valid = valid.drop(columns=["coordinate_status", "source_id"], errors="ignore") + return valid, memberships, invalid + + +def process_table_source( + repo_root: Path, + config: Dict[str, Any], + source_key: str, + source_cfg: Dict[str, Any], + sample_rows: Optional[int], +) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: + path = source_paths(repo_root, config, source_cfg)[0] + raw = read_csv_selected(path, usecols=required_columns_for_table(source_cfg), nrows=sample_rows) + if source_cfg.get("pincode_col") and source_cfg.get("lat_col") in raw.columns and source_cfg.get("lon_col") in raw.columns: + centroids = load_pincode_centroids(repo_root, config) + raw[source_cfg["lat_col"]] = pd.to_numeric(raw[source_cfg["lat_col"]], errors="coerce") + raw[source_cfg["lon_col"]] = pd.to_numeric(raw[source_cfg["lon_col"]], errors="coerce") + raw = fill_coords_from_pincode(raw, source_cfg["lat_col"], source_cfg["lon_col"], source_cfg["pincode_col"], centroids) + lookup = taxonomy_lookup(config) + facility_class = source_cfg["facility_class"] + facilities, memberships = standard_facility_frame(raw, source_cfg, facility_class, lookup[facility_class]) + return split_valid_invalid(facilities, memberships) + + +def process_apmc_source( + repo_root: Path, + config: Dict[str, Any], + source_key: str, + source_cfg: Dict[str, Any], + sample_rows: Optional[int], +) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: + path = source_paths(repo_root, config, source_cfg)[0] + raw = read_csv_selected( + path, + usecols=["gid", "mandi_code", "fatehabad", "market_cat", "district_n", "lat", "long"], + nrows=sample_rows, + ) + apmc_mapping = (config.get("mappings") or {}).get("apmc") or {} + category_normalise = apmc_mapping.get("category_normalise") or {} + category_labels = apmc_mapping.get("category_labels") or {} + raw["apmc_category"] = raw["market_cat"].fillna("Other").replace(category_normalise).map(category_labels).fillna("Other") + raw["apmc_name"] = clean_text_series(raw["fatehabad"]).fillna("") + " APMC" + raw = raw.rename(columns={"gid": "id", "lat": "latitude", "long": "longitude", "mandi_code": "facility_code", "district_n": "village_name"}) + local_cfg = { + **source_cfg, + "id_col": "id", + "name_col": "apmc_name", + "lat_col": "latitude", + "lon_col": "longitude", + "facility_code_col": "facility_code", + "village_name_col": "village_name", + } + lookup = taxonomy_lookup(config) + facilities, memberships = standard_facility_frame(raw, local_cfg, "apmc", lookup["apmc"], subtype=raw["apmc_category"]) + return split_valid_invalid(facilities, memberships) + + +def process_health_source( + repo_root: Path, + config: Dict[str, Any], + source_key: str, + source_cfg: Dict[str, Any], + sample_rows: Optional[int], +) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: + path = source_paths(repo_root, config, source_cfg)[0] + raw = read_csv_selected(path, usecols=["FID", "facility_t", "facility_n", "coordinates"], nrows=sample_rows) + parsed = raw["coordinates"].map(parse_coordinate_pair) + raw["longitude"] = parsed.map(lambda item: item[0]) + raw["latitude"] = parsed.map(lambda item: item[1]) + raw["facility_t"] = clean_string_series(raw["facility_t"]) + frames: List[pd.DataFrame] = [] + memberships: List[pd.DataFrame] = [] + invalid: List[pd.DataFrame] = [] + lookup = taxonomy_lookup(config) + for source_type, facility_class in source_cfg["type_to_class"].items(): + sub = raw[raw["facility_t"] == source_type].copy() + if sub.empty: + continue + local_cfg = { + **source_cfg, + "id_col": "FID", + "name_col": "facility_n", + "lat_col": "latitude", + "lon_col": "longitude", + "facility_uid_prefix": f"health_{source_type}", + } + f, m = standard_facility_frame(sub, local_cfg, facility_class, lookup[facility_class]) + vf, vm, inv = split_valid_invalid(f, m) + frames.append(vf) + memberships.append(vm) + invalid.append(inv) + return ( + pd.concat(frames, ignore_index=True, sort=False) if frames else pd.DataFrame(), + pd.concat(memberships, ignore_index=True, sort=False) if memberships else pd.DataFrame(), + pd.concat(invalid, ignore_index=True, sort=False) if invalid else pd.DataFrame(), + ) + + +def process_agri_source( + repo_root: Path, + config: Dict[str, Any], + source_key: str, + source_cfg: Dict[str, Any], + sample_rows: Optional[int], +) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: + frames = [] + for path in source_paths(repo_root, config, source_cfg): + part = read_csv_selected(path, usecols=["facility_i", "fac_desc", "dt_lgd", "lattitude", "longitude", "subtype"], nrows=sample_rows) + part["raw_file"] = path.name + frames.append(part) + raw = pd.concat(frames, ignore_index=True, sort=False) + agri_mapping = (config.get("mappings") or {}).get("agri_industry") or {} + subtype_to_category = { + subtype: category + for category, subtypes in (agri_mapping.get("subtype_categories") or {}).items() + for subtype in (subtypes or []) + } + category_to_class = agri_mapping.get("category_to_class") or {} + raw["reclassified_category"] = raw["subtype"].map(subtype_to_category).fillna("Industrial Manufacturing") + raw["facility_class"] = raw["reclassified_category"].map(category_to_class).fillna("agri_industry_industrial_manufacturing") + raw = raw.rename( + columns={ + "facility_i": "id", + "fac_desc": "facility_name", + "dt_lgd": "district_lgd", + "lattitude": "latitude", + "longitude": "longitude", + } + ) + raw["facility_name"] = clean_text_series(raw["facility_name"]) + lookup = taxonomy_lookup(config) + frames_out: List[pd.DataFrame] = [] + memberships_out: List[pd.DataFrame] = [] + invalid_out: List[pd.DataFrame] = [] + for facility_class, sub in raw.groupby("facility_class", sort=True): + local_cfg = { + **source_cfg, + "id_col": "id", + "name_col": "facility_name", + "lat_col": "latitude", + "lon_col": "longitude", + "district_lgd_col": "district_lgd", + "uid_strategy": "id_coord", + } + f, m = standard_facility_frame(sub, local_cfg, facility_class, lookup[facility_class], subtype=sub["subtype"]) + vf, vm, inv = split_valid_invalid(f, m) + frames_out.append(vf) + memberships_out.append(vm) + invalid_out.append(inv) + return ( + pd.concat(frames_out, ignore_index=True, sort=False), + pd.concat(memberships_out, ignore_index=True, sort=False), + pd.concat(invalid_out, ignore_index=True, sort=False), + ) + + +def process_school_source( + repo_root: Path, + config: Dict[str, Any], + source_key: str, + source_cfg: Dict[str, Any], + sample_rows: Optional[int], + chunksize: int, + facility_path: Path, + membership_path: Path, + invalid_path: Path, +) -> Dict[str, Any]: + path = source_paths(repo_root, config, source_cfg)[0] + usecols = ["lgd_distri", "vilcode11", "vilname", "schcd", "schname", "school_cat", "schcat", "management", "schmgt", "latitude", "longitude"] + lookup = taxonomy_lookup(config) + school_mapping = (config.get("mappings") or {}).get("school") or {} + category_labels = {int(key): value for key, value in (school_mapping.get("category_labels") or {}).items()} + category_to_classes = { + int(key): list(value or []) + for key, value in (school_mapping.get("category_to_classes") or {}).items() + } + header_f = header_m = header_i = True + input_rows = valid_rows = membership_rows = invalid_rows = 0 + reader = pd.read_csv(path, usecols=usecols, chunksize=chunksize, low_memory=False, nrows=sample_rows) + for chunk in reader: + input_rows += len(chunk) + chunk["schcat_num"] = pd.to_numeric(chunk["schcat"], errors="coerce").astype("Int64") + missing_category = clean_string_series(chunk["school_cat"]).isna() + if missing_category.any(): + chunk.loc[missing_category, "school_cat"] = chunk.loc[missing_category, "schcat_num"].map(category_labels) + levels = chunk["schcat_num"].map(lambda code: category_to_classes.get(int(code), []) if pd.notna(code) else []) + primary_class = levels.map(lambda items: items[0] if items else pd.NA) + keep = primary_class.notna() + chunk = chunk.loc[keep].copy() + levels = levels.loc[keep] + primary_class = primary_class.loc[keep] + if chunk.empty: + continue + + local_cfg = { + **source_cfg, + "id_col": "schcd", + "name_col": "schname", + "lat_col": "latitude", + "lon_col": "longitude", + "facility_code_col": "schcd", + "district_lgd_col": "lgd_distri", + "village_census11_col": "vilcode11", + "village_name_col": "vilname", + "clean_text_cols": ["schname"], + } + facilities, _ = standard_facility_frame(chunk, local_cfg, primary_class.iloc[0], lookup[primary_class.iloc[0]], subtype=chunk["school_cat"]) + facilities["class_l1_domain"] = primary_class.map(lambda cls: lookup[cls]["class_l1_domain"]) + facilities["class_l2_filter_group"] = primary_class.map(lambda cls: lookup[cls]["class_l2_filter_group"]) + facilities["class_l3_facility_class"] = primary_class.values + + memberships = [] + for facility_class in sorted(set(cls for items in levels for cls in items)): + mask = levels.map(lambda items: facility_class in items) + part = facilities.loc[mask].copy() + part["class_l1_domain"] = lookup[facility_class]["class_l1_domain"] + part["class_l2_filter_group"] = lookup[facility_class]["class_l2_filter_group"] + part["class_l3_facility_class"] = facility_class + memberships.append(membership_from_facilities(part)) + membership_df = pd.concat(memberships, ignore_index=True, sort=False) + valid, membership_df, invalid = split_valid_invalid(facilities, membership_df) + append_csv(valid, facility_path, header_f) + append_csv(membership_df, membership_path, header_m) + if not invalid.empty: + append_csv(invalid, invalid_path, header_i) + header_i = False + header_f = header_m = False + valid_rows += len(valid) + membership_rows += len(membership_df) + invalid_rows += len(invalid) + log.info("Processed school rows=%d valid=%d memberships=%d", input_rows, valid_rows, membership_rows) + return { + "source_key": source_key, + "input_rows": input_rows, + "valid_rows": valid_rows, + "membership_rows": membership_rows, + "invalid_rows": invalid_rows, + } + + +PROCESSORS = { + "table": process_table_source, + "apmc": process_apmc_source, + "health_center_split": process_health_source, + "agri_industry_reclassified": process_agri_source, +} + + +def process_source( + repo_root: Path, + config: Dict[str, Any], + source_key: str, + sample_rows: Optional[int], + chunksize: int, +) -> Dict[str, Any]: + source_cfg = config["sources"][source_key] + intermediate = resolve_path(config["pipeline"]["intermediate_dir"], repo_root) + facility_path = intermediate / "source_facilities" / f"{source_key}.csv" + membership_path = intermediate / "source_memberships" / f"{source_key}.csv" + invalid_path = intermediate / "invalid" / f"{source_key}.csv" + for path in [facility_path, membership_path, invalid_path]: + if path.exists(): + path.unlink() + + if source_cfg["processor"] == "school_mhrd": + return process_school_source(repo_root, config, source_key, source_cfg, sample_rows, chunksize, facility_path, membership_path, invalid_path) + + processor = PROCESSORS[source_cfg["processor"]] + facilities, memberships, invalid = processor(repo_root, config, source_key, source_cfg, sample_rows) + append_csv(facilities, facility_path, True) + append_csv(memberships, membership_path, True) + if not invalid.empty: + append_csv(invalid, invalid_path, True) + return { + "source_key": source_key, + "input_rows": int(len(facilities) + len(invalid)), + "valid_rows": int(len(facilities)), + "membership_rows": int(len(memberships)), + "invalid_rows": int(len(invalid)), + } From 70f0a8c4128912567bad2c38d7dcdfb0ff315c6b Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:27 +0000 Subject: [PATCH 3/7] Add facilities dataset build CLI This CLI orchestrates raw cleaning, pan-India facility asset creation, proximity rebuilds, and monitoring commands from tracked YAML config instead of one-off local scripts. --- .../facilities_utils/facility_pipeline.py | 368 ++++++++++++++++++ 1 file changed, 368 insertions(+) create mode 100644 utilities/scripts/facilities_utils/facility_pipeline.py diff --git a/utilities/scripts/facilities_utils/facility_pipeline.py b/utilities/scripts/facilities_utils/facility_pipeline.py new file mode 100644 index 00000000..5f1ffbbd --- /dev/null +++ b/utilities/scripts/facilities_utils/facility_pipeline.py @@ -0,0 +1,368 @@ +#!/usr/bin/env python3 +"""CLI orchestrator for the pan-India facilities pipeline.""" + +from __future__ import annotations + +import argparse +import json +import logging +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, List, Optional + +import pandas as pd + +from facility_cleaners import process_source, source_paths +from facility_utils import ( + CLASS_K_COLUMNS, + add_attribute_table_to_gpkg, + ensure_sqlite_indexes, + file_fingerprint, + find_repo_root, + read_yaml, + resolve_path, + setup_logging, + taxonomy_rows, + write_yaml, +) + + +log = logging.getLogger("facilities") + + +def load_config(config_path: Path) -> tuple[Path, Dict[str, Any]]: + repo_root = find_repo_root(config_path.resolve()) + return repo_root, read_yaml(config_path) + + +def state_path(repo_root: Path, config: Dict[str, Any]) -> Path: + return resolve_path(config["pipeline"]["state_path"], repo_root) + + +def load_state(path: Path) -> Dict[str, Any]: + if not path.exists(): + return {"sources": {}, "runs": []} + return json.loads(path.read_text(encoding="utf-8")) + + +def save_state(path: Path, state: Dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(state, indent=2, sort_keys=True), encoding="utf-8") + + +def source_config_fingerprint(repo_root: Path, config: Dict[str, Any], source_key: str, config_path: Path) -> str: + paths = source_paths(repo_root, config, config["sources"][source_key]) + paths = [*paths, config_path] + return file_fingerprint(paths) + + +def active_sources(config: Dict[str, Any], requested: Optional[str]) -> List[str]: + sources = [] + for key, source in config.get("sources", {}).items(): + if requested and key != requested: + continue + if source.get("active", True): + sources.append(key) + if requested and not sources: + raise KeyError(f"Unknown or inactive source: {requested}") + return sources + + +def run_clean(args: argparse.Namespace) -> None: + repo_root, config = load_config(args.config) + setup_logging(args.debug, resolve_path(config["pipeline"]["log_path"], repo_root)) + state_file = state_path(repo_root, config) + state = load_state(state_file) + summaries = [] + + for source_key in active_sources(config, args.source): + fingerprint = source_config_fingerprint(repo_root, config, source_key, args.config) + previous = state["sources"].get(source_key, {}) + if not args.force and not args.sample_rows and previous.get("fingerprint") == fingerprint: + log.info("Skipping unchanged source: %s", source_key) + summaries.append(previous.get("summary", {"source_key": source_key, "skipped": True})) + continue + log.info("Processing source: %s", source_key) + summary = process_source( + repo_root=repo_root, + config=config, + source_key=source_key, + sample_rows=args.sample_rows, + chunksize=args.chunksize, + ) + summary["fingerprint"] = fingerprint + summary["processed_at_utc"] = datetime.now(timezone.utc).replace(microsecond=0).isoformat() + summary["sample_rows"] = args.sample_rows + state["sources"][source_key] = {"fingerprint": fingerprint, "summary": summary} + summaries.append(summary) + save_state(state_file, state) + + source_summary_path = resolve_path(config["outputs"]["source_summary_csv"], repo_root) + source_summary_path.parent.mkdir(parents=True, exist_ok=True) + pd.DataFrame([item.get("summary", item) for item in state["sources"].values()]).to_csv(source_summary_path, index=False) + log.info("Saved source summary: %s", source_summary_path) + if not getattr(args, "skip_monitor", False): + from facility_metadata_monitor import write_monitor + + monitor_path = write_monitor(args.config) + log.info("Updated metadata monitor: %s", monitor_path) + + +def concat_csvs(paths: List[Path], out_path: Path, columns: Optional[List[str]] = None) -> None: + out_path.parent.mkdir(parents=True, exist_ok=True) + if out_path.exists(): + out_path.unlink() + header = True + for path in paths: + if not path.exists(): + continue + for chunk in pd.read_csv(path, chunksize=250_000, low_memory=False): + if columns: + for column in columns: + if column not in chunk.columns: + chunk[column] = pd.NA + chunk = chunk[columns] + chunk.to_csv(out_path, mode="a", index=False, header=header) + header = False + + +def membership_counts(membership_csv: Path) -> Dict[str, int]: + counts: Dict[str, int] = {} + for chunk in pd.read_csv(membership_csv, usecols=["facility_uid", "class_l3_facility_class"], chunksize=500_000, low_memory=False): + chunk = chunk.drop_duplicates() + part = chunk.groupby("facility_uid", sort=False).size() + for uid, count in part.items(): + counts[uid] = counts.get(uid, 0) + int(count) + return counts + + +def build_facility_csv(repo_root: Path, config: Dict[str, Any]) -> None: + intermediate = resolve_path(config["pipeline"]["intermediate_dir"], repo_root) + output_facilities = resolve_path(config["outputs"]["facilities_csv"], repo_root) + output_memberships = resolve_path(config["outputs"]["memberships_csv"], repo_root) + invalid_output = resolve_path(config["outputs"]["invalid_coordinates_csv"], repo_root) + taxonomy_output = resolve_path(config["outputs"]["taxonomy_csv"], repo_root) + + source_keys = [key for key, source in config["sources"].items() if source.get("active", True)] + facility_paths = [intermediate / "source_facilities" / f"{key}.csv" for key in source_keys] + membership_paths = [intermediate / "source_memberships" / f"{key}.csv" for key in source_keys] + invalid_paths = [intermediate / "invalid" / f"{key}.csv" for key in source_keys] + + concat_csvs(membership_paths, output_memberships, config["schema"]["membership_columns"]) + counts = membership_counts(output_memberships) + + output_facilities.parent.mkdir(parents=True, exist_ok=True) + if output_facilities.exists(): + output_facilities.unlink() + header = True + columns = config["schema"]["facility_columns"] + seen: set[str] = set() + for path in facility_paths: + if not path.exists(): + log.warning("Missing source facility intermediate: %s", path) + continue + for chunk in pd.read_csv(path, chunksize=250_000, low_memory=False): + if "facility_uid" not in chunk.columns: + continue + chunk = chunk[~chunk["facility_uid"].isin(seen)].copy() + seen.update(chunk["facility_uid"].dropna().astype(str).tolist()) + chunk["membership_count"] = chunk["facility_uid"].map(counts).fillna(1).astype("Int64") + for column in columns: + if column not in chunk.columns: + chunk[column] = pd.NA + chunk[columns].to_csv(output_facilities, mode="a", index=False, header=header) + header = False + concat_csvs(invalid_paths, invalid_output) + pd.DataFrame(taxonomy_rows(config)).to_csv(taxonomy_output, index=False) + log.info("Saved facilities CSV: %s", output_facilities) + log.info("Saved memberships CSV: %s", output_memberships) + log.info("Saved taxonomy CSV: %s", taxonomy_output) + + +def write_large_attribute_table(gpkg_path: Path, table_name: str, csv_path: Path) -> None: + if not csv_path.exists(): + return + with sqlite3.connect(gpkg_path) as con: + first = True + for chunk in pd.read_csv(csv_path, chunksize=250_000, low_memory=False): + chunk.to_sql(table_name, con, if_exists="replace" if first else "append", index=False) + first = False + con.execute( + """ + insert or replace into gpkg_contents + (table_name, data_type, identifier, description, last_change, min_x, min_y, max_x, max_y, srs_id) + values (?, 'attributes', ?, '', strftime('%Y-%m-%dT%H:%M:%fZ','now'), null, null, null, null, null) + """, + (table_name, table_name), + ) + con.commit() + + +def write_gpkg(repo_root: Path, config: Dict[str, Any], chunksize: int) -> None: + try: + import geopandas as gpd + import pyogrio + except ImportError as exc: + raise RuntimeError("GeoPackage output requires geopandas, shapely, and pyogrio.") from exc + + facility_csv = resolve_path(config["outputs"]["facilities_csv"], repo_root) + membership_csv = resolve_path(config["outputs"]["memberships_csv"], repo_root) + taxonomy_csv = resolve_path(config["outputs"]["taxonomy_csv"], repo_root) + source_summary_csv = resolve_path(config["outputs"]["source_summary_csv"], repo_root) + invalid_csv = resolve_path(config["outputs"]["invalid_coordinates_csv"], repo_root) + gpkg_path = resolve_path(config["outputs"]["facilities_gpkg"], repo_root) + gpkg_path.parent.mkdir(parents=True, exist_ok=True) + if gpkg_path.exists(): + gpkg_path.unlink() + + append = False + total = 0 + text_columns = [ + "facility_uid", + "facility_name", + "facility_code", + "class_l1_domain", + "class_l2_filter_group", + "class_l3_facility_class", + "class_l4_facility_subtype", + *CLASS_K_COLUMNS, + "urban_rural", + "village_census11", + "village_name", + ] + integer_columns = ["pincode", "establishment_year", "district_lgd", "membership_count"] + for chunk in pd.read_csv(facility_csv, chunksize=chunksize, low_memory=False): + chunk["latitude"] = pd.to_numeric(chunk["latitude"], errors="coerce") + chunk["longitude"] = pd.to_numeric(chunk["longitude"], errors="coerce") + chunk = chunk.dropna(subset=["latitude", "longitude"]) + for column in text_columns: + if column in chunk.columns: + chunk[column] = chunk[column].astype("string") + for column in integer_columns: + if column in chunk.columns: + chunk[column] = pd.to_numeric(chunk[column], errors="coerce").astype("Int64") + geometry = gpd.points_from_xy(chunk["longitude"], chunk["latitude"], crs=config["schema"]["crs"]) + gdf = gpd.GeoDataFrame(chunk, geometry=geometry, crs=config["schema"]["crs"]) + pyogrio.write_dataframe(gdf, gpkg_path, layer=config["schema"]["facility_layer"], driver="GPKG", append=append) + append = True + total += len(gdf) + log.info("Wrote %d facility points", total) + + write_large_attribute_table(gpkg_path, config["schema"]["membership_table"], membership_csv) + add_attribute_table_to_gpkg(gpkg_path, config["schema"]["taxonomy_table"], pd.read_csv(taxonomy_csv)) + if source_summary_csv.exists(): + add_attribute_table_to_gpkg(gpkg_path, config["schema"]["source_summary_table"], pd.read_csv(source_summary_csv)) + if invalid_csv.exists(): + add_attribute_table_to_gpkg(gpkg_path, config["schema"]["invalid_table"], pd.read_csv(invalid_csv, low_memory=False)) + ensure_sqlite_indexes(gpkg_path) + log.info("Saved facilities GeoPackage: %s (%d points)", gpkg_path, total) + + +def run_build(args: argparse.Namespace) -> None: + repo_root, config = load_config(args.config) + setup_logging(args.debug, resolve_path(config["pipeline"]["log_path"], repo_root)) + if not getattr(args, "gpkg_only", False): + build_facility_csv(repo_root, config) + if not args.skip_gpkg: + write_gpkg(repo_root, config, args.gpkg_chunksize) + if not getattr(args, "skip_monitor", False): + from facility_metadata_monitor import write_monitor + + monitor_path = write_monitor(args.config) + log.info("Updated metadata monitor: %s", monitor_path) + + +def run_proximity(args: argparse.Namespace) -> None: + from facility_proximity_finder import run_from_args + + run_from_args(args) + + +def run_all(args: argparse.Namespace) -> None: + original_skip_monitor = getattr(args, "skip_monitor", False) + args.skip_monitor = True + run_clean(args) + run_build(args) + if args.with_proximity: + run_proximity(args) + args.skip_monitor = original_skip_monitor + if not args.skip_monitor: + from facility_metadata_monitor import write_monitor + + monitor_path = write_monitor(args.config) + log.info("Updated metadata monitor: %s", monitor_path) + + +def run_monitor(args: argparse.Namespace) -> None: + from facility_metadata_monitor import write_monitor + + monitor_path = write_monitor(args.config) + print(monitor_path) + + +def build_parser() -> argparse.ArgumentParser: + repo_root = find_repo_root(Path(__file__).resolve()) + default_config = repo_root / "utilities" / "scripts" / "facilities_utils" / "config" / "facilities_master.yaml" + parser = argparse.ArgumentParser(description="Build and maintain the pan-India facilities pipeline.") + parser.add_argument("--config", type=Path, default=default_config) + parser.add_argument("--debug", action="store_true") + sub = parser.add_subparsers(dest="command", required=True) + + clean = sub.add_parser("clean", help="Process raw sources into per-source intermediates.") + clean.add_argument("--source", type=str, default=None) + clean.add_argument("--sample-rows", type=int, default=None) + clean.add_argument("--chunksize", type=int, default=500_000) + clean.add_argument("--force", action="store_true") + clean.add_argument("--skip-monitor", action="store_true") + clean.set_defaults(func=run_clean) + + build = sub.add_parser("build", help="Build CSV outputs and the facilities GPKG from intermediates.") + build.add_argument("--skip-gpkg", action="store_true") + build.add_argument("--gpkg-only", action="store_true") + build.add_argument("--gpkg-chunksize", type=int, default=250_000) + build.add_argument("--skip-monitor", action="store_true") + build.set_defaults(func=run_build) + + prox = sub.add_parser("proximity", help="Build village-to-facility proximity GPKG.") + add_proximity_args(prox) + prox.set_defaults(func=run_proximity) + + all_cmd = sub.add_parser("all", help="Run clean, build, and optionally proximity.") + all_cmd.add_argument("--source", type=str, default=None) + all_cmd.add_argument("--sample-rows", type=int, default=None) + all_cmd.add_argument("--chunksize", type=int, default=500_000) + all_cmd.add_argument("--force", action="store_true") + all_cmd.add_argument("--skip-gpkg", action="store_true") + all_cmd.add_argument("--gpkg-chunksize", type=int, default=250_000) + all_cmd.add_argument("--with-proximity", action="store_true") + add_proximity_args(all_cmd, include_force=False) + all_cmd.set_defaults(func=run_all) + + monitor = sub.add_parser("monitor", help="Refresh utilities/scripts/facilities_utils/config/facilities_metadata_monitor.yaml.") + monitor.set_defaults(func=run_monitor) + return parser + + +def add_proximity_args(parser: argparse.ArgumentParser, include_force: bool = True) -> None: + parser.add_argument("--classes", type=str, default=None, help="Comma-separated L3 classes to compute.") + parser.add_argument("--sample-villages", type=int, default=None) + parser.add_argument("--sample-classes", type=int, default=None) + parser.add_argument("--village-chunksize", type=int, default=100_000) + parser.add_argument("--output-gpkg", type=Path, default=None) + parser.add_argument("--materialize-derived", action="store_true", help="Legacy flag; L2 is materialized by default after full L3 completion.") + parser.add_argument("--no-derived-views", action="store_true") + parser.add_argument("--refresh-derived-only", action="store_true", help="Rebuild class-map and materialized L2 output from existing L3 rows.") + parser.add_argument("--skip-monitor", action="store_true") + if include_force: + parser.add_argument("--force", action="store_true") + + +def main() -> None: + parser = build_parser() + args = parser.parse_args() + args.func(args) + + +if __name__ == "__main__": + main() From ab8ed157e839a72c399565bd9d4254101a4de80b Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:28 +0000 Subject: [PATCH 4/7] Build facilities proximity on CS village polygons This script builds cs_village_facility_proximity.gpkg from cs_admin_standard and pan_india_facilities.gpkg, storing actual village polygons and materialized proximity tables for downstream API and GEE use. --- .../facility_proximity_finder.py | 975 ++++++++++++++++++ 1 file changed, 975 insertions(+) create mode 100644 utilities/scripts/facilities_utils/facility_proximity_finder.py diff --git a/utilities/scripts/facilities_utils/facility_proximity_finder.py b/utilities/scripts/facilities_utils/facility_proximity_finder.py new file mode 100644 index 00000000..6bee080f --- /dev/null +++ b/utilities/scripts/facilities_utils/facility_proximity_finder.py @@ -0,0 +1,975 @@ +#!/usr/bin/env python3 +"""Build lean village-to-facility proximity outputs from facilities and village GPKGs.""" + +from __future__ import annotations + +import argparse +import logging +import sqlite3 +from pathlib import Path +from typing import Any, Dict, List, Optional + +import numpy as np +import pandas as pd +from scipy.spatial import cKDTree + +from facility_utils import find_repo_root, read_yaml, resolve_path, setup_logging, slugify, taxonomy_rows + + +EARTH_RADIUS_KM = 6371.0088 +log = logging.getLogger("facilities") + +INTEGER_VILLAGE_COLUMNS = { + "pc11_village_id", + "village_id", + "pc11_state_id", + "pc11_district_id", + "pc11_subdistrict_id", + "feature_part_index", + "feature_part_count", +} + + +def sql_quote(value: Any) -> str: + if value is None or pd.isna(value): + return "null" + return "'" + str(value).replace("'", "''") + "'" + + +def sql_ident(value: str) -> str: + return '"' + str(value).replace('"', '""') + '"' + + +def lonlat_to_xyz(lat: np.ndarray, lon: np.ndarray) -> np.ndarray: + lat_rad = np.radians(lat) + lon_rad = np.radians(lon) + return np.column_stack((np.cos(lat_rad) * np.cos(lon_rad), np.cos(lat_rad) * np.sin(lon_rad), np.sin(lat_rad))) + + +def chord_to_km(chord: np.ndarray) -> np.ndarray: + chord = np.clip(chord, 0, 2) + return EARTH_RADIUS_KM * 2 * np.arcsin(chord / 2) + + +def village_shapes_table(config: Dict[str, Any]) -> str: + return config["proximity"]["tables"].get("village_shapes") or "village_shapes" + + +def coerce_integer_village_columns(villages: Any) -> Any: + for column in INTEGER_VILLAGE_COLUMNS.intersection(villages.columns): + numeric = pd.to_numeric(villages[column], errors="coerce") + fractional = numeric.notna() & (numeric != np.floor(numeric)) + if bool(fractional.any()): + raise ValueError(f"Column {column} has non-integer values and cannot be written as INTEGER") + villages[column] = numeric.astype("Int64") + return villages + + +def load_villages(config: Dict[str, Any], repo_root: Path, sample_villages: Optional[int]) -> Any: + import geopandas as gpd + + prox = config["proximity"] + rows = slice(0, sample_villages) if sample_villages else None + villages = gpd.read_file(resolve_path(prox["village_gpkg"], repo_root), layer=prox["village_layer"], rows=rows, engine="pyogrio") + if villages.crs is None: + villages = villages.set_crs(config["schema"]["crs"]) + villages = villages.to_crs(config["schema"]["crs"]) + points = villages.geometry.representative_point() + villages = villages.copy() + villages = coerce_integer_village_columns(villages) + villages["_village_latitude"] = points.y + villages["_village_longitude"] = points.x + return villages + + +def register_attribute_table(con: sqlite3.Connection, table_name: str) -> None: + con.execute( + """ + insert or replace into gpkg_contents + (table_name, data_type, identifier, description, last_change, min_x, min_y, max_x, max_y, srs_id) + values (?, 'attributes', ?, '', strftime('%Y-%m-%dT%H:%M:%fZ','now'), null, null, null, null, null) + """, + (table_name, table_name), + ) + + +def drop_derived_outputs(con: sqlite3.Connection, config: Dict[str, Any]) -> None: + tables = config["proximity"]["tables"] + con.execute(f"drop view if exists {sql_ident(tables['l2_view'])}") + con.execute(f"drop view if exists {sql_ident(tables['l1_view'])}") + con.execute(f"drop table if exists {sql_ident(tables['l2_materialized'])}") + con.execute(f"drop table if exists {sql_ident(tables['l1_materialized'])}") + + +def output_village_count(output_gpkg: Path, village_layer: str) -> Optional[int]: + if not output_gpkg.exists(): + return None + try: + with sqlite3.connect(output_gpkg) as con: + return int(con.execute(f"select count(*) from {sql_ident(village_layer)}").fetchone()[0]) + except sqlite3.Error: + return None + + +def village_context_table(config: Dict[str, Any]) -> str: + tables = config["proximity"]["tables"] + return tables.get("village_shapes", "village_shapes") + + +def prepare_output_gpkg(output_gpkg: Path, villages: Any, config: Dict[str, Any], force_full: bool) -> None: + tables = config["proximity"]["tables"] + village_layer = village_shapes_table(config) + existing_villages = output_village_count(output_gpkg, village_layer) + if output_gpkg.exists() and not force_full and existing_villages == len(villages): + log.info("Resuming existing proximity GeoPackage: %s", output_gpkg) + return + if output_gpkg.exists(): + output_gpkg.unlink() + village_cols = [ + config["proximity"]["village_id_col"], + *(config["proximity"].get("village_context_cols") or []), + "_village_latitude", + "_village_longitude", + "geometry", + ] + village_cols = [col for col in village_cols if col in villages.columns] + villages[village_cols].to_file(output_gpkg, layer=village_layer, driver="GPKG", engine="pyogrio") + with sqlite3.connect(output_gpkg) as con: + con.execute(f"drop table if exists {sql_ident(tables['l3'])}") + con.commit() + + +def l3_groups(config: Dict[str, Any], facilities_gpkg: Path, classes: Optional[List[str]], sample_classes: Optional[int]) -> List[Dict[str, Any]]: + requested = set(classes or []) + df = pd.DataFrame(taxonomy_rows(config)) + df = df[["class_l1_domain", "class_l2_filter_group", "class_l3_facility_class", "filter_logic"]].drop_duplicates() + if requested: + df = df[df["class_l3_facility_class"].isin(requested)] + if sample_classes: + df = df.head(sample_classes) + return df.to_dict(orient="records") + + +def completed_l3_classes(output_gpkg: Path, config: Dict[str, Any], expected_villages: int) -> set[str]: + if not output_gpkg.exists(): + return set() + try: + with sqlite3.connect(output_gpkg) as con: + rows = con.execute( + f""" + select class_l3_facility_class, count(*) + from {sql_ident(config['proximity']['tables']['l3'])} + group by class_l3_facility_class + having count(*) >= ? + """, + (expected_villages,), + ).fetchall() + return {row[0] for row in rows} + except sqlite3.Error: + return set() + + +def load_l3_facility_points(config: Dict[str, Any], facilities_gpkg: Path, facility_class: str) -> pd.DataFrame: + sql = f""" + select distinct + f.facility_uid, + f.facility_name, + f.facility_code, + f.latitude, + f.longitude, + m.class_l4_facility_subtype + from {config['schema']['facility_layer']} f + join {config['schema']['membership_table']} m on m.facility_uid = f.facility_uid + where m.class_l3_facility_class = ? + and f.latitude is not null + and f.longitude is not null + """ + with sqlite3.connect(facilities_gpkg) as con: + points = pd.read_sql_query(sql, con, params=(facility_class,)) + points["latitude"] = pd.to_numeric(points["latitude"], errors="coerce") + points["longitude"] = pd.to_numeric(points["longitude"], errors="coerce") + return points.dropna(subset=["latitude", "longitude"]).reset_index(drop=True) + + +def append_table(output_gpkg: Path, table_name: str, frame: pd.DataFrame) -> None: + with sqlite3.connect(output_gpkg) as con: + frame.to_sql(table_name, con, if_exists="append", index=False) + + +def nearest_lookup_table(config: Dict[str, Any]) -> str: + return config["proximity"]["tables"].get("nearest_facilities", "proximity_nearest_facilities") + + +def ensure_nearest_lookup_table(con: sqlite3.Connection, config: Dict[str, Any]) -> None: + table = nearest_lookup_table(config) + con.execute( + f""" + create table if not exists {sql_ident(table)} ( + facility_uid text primary key, + facility_name text, + facility_code text, + latitude real, + longitude real, + class_l4_facility_subtype text + ) + """ + ) + register_attribute_table(con, table) + + +def append_nearest_lookup(output_gpkg: Path, config: Dict[str, Any], nearest: pd.DataFrame) -> None: + columns = [ + "facility_uid", + "facility_name", + "facility_code", + "latitude", + "longitude", + "class_l4_facility_subtype", + ] + lookup = nearest[columns].drop_duplicates(subset=["facility_uid"]).where(pd.notna(nearest[columns]), None) + if lookup.empty: + return + table = nearest_lookup_table(config) + with sqlite3.connect(output_gpkg) as con: + ensure_nearest_lookup_table(con, config) + con.executemany( + f""" + insert or ignore into {sql_ident(table)} + (facility_uid, facility_name, facility_code, latitude, longitude, class_l4_facility_subtype) + values (?, ?, ?, ?, ?, ?) + """, + lookup.itertuples(index=False, name=None), + ) + con.commit() + + +def nearest_lookup_missing(output_gpkg: Path, config: Dict[str, Any]) -> bool: + if not output_gpkg.exists(): + return True + table = nearest_lookup_table(config) + try: + with sqlite3.connect(output_gpkg) as con: + if not table_exists(con, table): + return True + return int(con.execute(f"select count(*) from {sql_ident(table)}").fetchone()[0]) == 0 + except sqlite3.Error: + return True + + +def rebuild_nearest_lookup(output_gpkg: Path, config: Dict[str, Any], facilities_gpkg: Path, batch_size: int = 900) -> None: + l3_table = config["proximity"]["tables"]["l3"] + with sqlite3.connect(output_gpkg) as out_con: + if not table_exists(out_con, l3_table): + ensure_nearest_lookup_table(out_con, config) + return + uids = [ + row[0] + for row in out_con.execute( + f""" + select distinct nearest_facility_uid + from {sql_ident(l3_table)} + where nearest_facility_uid is not null and nearest_facility_uid != '' + """ + ).fetchall() + ] + out_con.execute(f"drop table if exists {sql_ident(lookup_table)}") + ensure_nearest_lookup_table(out_con, config) + out_con.commit() + if not uids: + return + facility_table = config["schema"]["facility_layer"] + query_cols = [ + "facility_uid", + "facility_name", + "facility_code", + "latitude", + "longitude", + "class_l4_facility_subtype", + ] + with sqlite3.connect(facilities_gpkg) as facility_con: + for start in range(0, len(uids), batch_size): + batch = uids[start : start + batch_size] + placeholders = ", ".join("?" for _ in batch) + sql = ( + f"select {', '.join(query_cols)} " + f"from {sql_ident(facility_table)} " + f"where facility_uid in ({placeholders})" + ) + frame = pd.read_sql_query(sql, facility_con, params=batch) + append_nearest_lookup(output_gpkg, config, frame) + log.info("Rebuilt %s with %d nearest facility ids", lookup_table, len(uids)) + + +def table_exists(con: sqlite3.Connection, table_name: str) -> bool: + return ( + con.execute( + "select 1 from sqlite_master where type in ('table', 'view') and name = ?", + (table_name,), + ).fetchone() + is not None + ) + + +def delete_l3_classes(output_gpkg: Path, config: Dict[str, Any], classes: List[str]) -> None: + if not output_gpkg.exists() or not classes: + return + table = config["proximity"]["tables"]["l3"] + placeholders = ", ".join("?" for _ in classes) + with sqlite3.connect(output_gpkg) as con: + drop_derived_outputs(con, config) + if table_exists(con, table): + con.execute( + f"delete from {sql_ident(table)} where class_l3_facility_class in ({placeholders})", + classes, + ) + log.info("Deleted existing L3 proximity rows for: %s", ", ".join(classes)) + con.commit() + + +def build_l3_for_group( + output_gpkg: Path, + config: Dict[str, Any], + villages: Any, + group: Dict[str, Any], + points: pd.DataFrame, + village_chunksize: int, +) -> int: + tables = config["proximity"]["tables"] + tree = cKDTree(lonlat_to_xyz(points["latitude"].to_numpy(), points["longitude"].to_numpy())) + query_xyz = lonlat_to_xyz(villages["_village_latitude"].to_numpy(), villages["_village_longitude"].to_numpy()) + id_col = config["proximity"]["village_id_col"] + written = 0 + for start in range(0, len(villages), village_chunksize): + end = min(start + village_chunksize, len(villages)) + chord, index = tree.query(query_xyz[start:end], k=1) + nearest = points.iloc[index].reset_index(drop=True) + chunk = villages.iloc[start:end] + out = pd.DataFrame() + out[id_col] = chunk[id_col].astype("string").to_numpy() + out["class_l3_facility_class"] = group["class_l3_facility_class"] + out["nearest_distance_km"] = np.round(chord_to_km(chord), config["proximity"].get("distance_precision", 6)) + out["nearest_facility_uid"] = nearest["facility_uid"].to_numpy() + append_table(output_gpkg, tables["l3"], out) + append_nearest_lookup(output_gpkg, config, nearest) + written += len(out) + return written + + +def class_map_table(config: Dict[str, Any]) -> str: + return config["proximity"]["tables"].get("class_map", "proximity_class_map") + + +def class_map_frame(config: Dict[str, Any]) -> pd.DataFrame: + rows = [] + for row in taxonomy_rows(config): + rows.append( + { + "class_l3_facility_class": row["class_l3_facility_class"], + "class_l2_filter_group": row["class_l2_filter_group"], + "class_l1_domain": row["class_l1_domain"], + "filter_logic": row["filter_logic"], + } + ) + return pd.DataFrame(rows).drop_duplicates().sort_values( + ["class_l1_domain", "class_l2_filter_group", "class_l3_facility_class"] + ) + + +def write_class_map(output_gpkg: Path, config: Dict[str, Any]) -> None: + table = class_map_table(config) + frame = class_map_frame(config) + with sqlite3.connect(output_gpkg) as con: + frame.to_sql(table, con, if_exists="replace", index=False) + register_attribute_table(con, table) + con.execute(f"create index if not exists idx_{table}_l3 on {sql_ident(table)}(class_l3_facility_class)") + con.execute(f"create index if not exists idx_{table}_l2 on {sql_ident(table)}(class_l1_domain, class_l2_filter_group)") + con.execute(f"create index if not exists idx_{table}_l1 on {sql_ident(table)}(class_l1_domain)") + con.commit() + + +def ensure_proximity_indexes(con: sqlite3.Connection, config: Dict[str, Any]) -> None: + """Create the indexes needed by both derivation and tehsil export.""" + tables = config["proximity"]["tables"] + id_col = config["proximity"]["village_id_col"] + village_table = village_context_table(config) + l3_table = tables["l3"] + map_table = class_map_table(config) + + if table_exists(con, village_table): + con.execute( + f"create index if not exists idx_{village_table}_location " + f"on {sql_ident(village_table)}(state_name, district_name, TEHSIL)" + ) + con.execute( + f"create index if not exists idx_{village_table}_id " + f"on {sql_ident(village_table)}({sql_ident(id_col)})" + ) + if table_exists(con, l3_table): + con.execute( + f"create index if not exists idx_{l3_table}_village " + f"on {sql_ident(l3_table)}({sql_ident(id_col)})" + ) + con.execute( + f"create index if not exists idx_{l3_table}_class " + f"on {sql_ident(l3_table)}(class_l3_facility_class)" + ) + if table_exists(con, map_table): + con.execute( + f"create index if not exists idx_{map_table}_l3 " + f"on {sql_ident(map_table)}(class_l3_facility_class)" + ) + con.execute( + f"create index if not exists idx_{map_table}_l2 " + f"on {sql_ident(map_table)}(class_l1_domain, class_l2_filter_group)" + ) + + +def context_select_sql(config: Dict[str, Any]) -> str: + parts = [] + for col in config["proximity"].get("village_context_cols") or []: + parts.append(f",\n v.{sql_ident(col)} as {sql_ident(col)}") + return "".join(parts) + + +def outer_context_sql(config: Dict[str, Any]) -> str: + parts = [] + for col in config["proximity"].get("village_context_cols") or []: + parts.append(f", {sql_ident(col)}") + return "".join(parts) + + +def create_l2_view_sql(config: Dict[str, Any]) -> str: + id_col = config["proximity"]["village_id_col"] + l3_table = config["proximity"]["tables"]["l3"] + village_table = village_shapes_table(config) + map_table = class_map_table(config) + lookup_table = nearest_lookup_table(config) + return f""" + with ranked as ( + select + p.{sql_ident(id_col)} as {sql_ident(id_col)}{context_select_sql(config)}, + m.class_l1_domain, + m.class_l2_filter_group, + m.filter_logic, + case + when m.filter_logic = 'max' + then max(p.nearest_distance_km) over ( + partition by p.{sql_ident(id_col)}, m.class_l1_domain, m.class_l2_filter_group + ) + else min(p.nearest_distance_km) over ( + partition by p.{sql_ident(id_col)}, m.class_l1_domain, m.class_l2_filter_group + ) + end as logic_distance_km, + p.class_l3_facility_class as selected_component_class, + p.nearest_facility_uid, + nf.facility_name as nearest_facility_name, + nf.facility_code as nearest_facility_code, + nf.latitude as nearest_facility_latitude, + nf.longitude as nearest_facility_longitude, + nf.class_l4_facility_subtype as nearest_class_l4_facility_subtype, + row_number() over ( + partition by p.{sql_ident(id_col)}, m.class_l1_domain, m.class_l2_filter_group + order by + case when m.filter_logic = 'max' then p.nearest_distance_km end desc, + case when coalesce(m.filter_logic, 'min') != 'max' then p.nearest_distance_km end asc, + p.class_l3_facility_class + ) as rn + from {sql_ident(l3_table)} p + join {sql_ident(map_table)} m + on m.class_l3_facility_class = p.class_l3_facility_class + left join {sql_ident(village_table)} v + on cast(v.{sql_ident(id_col)} as text) = cast(p.{sql_ident(id_col)} as text) + left join {sql_ident(lookup_table)} nf + on nf.facility_uid = p.nearest_facility_uid + ) + select + {sql_ident(id_col)}{outer_context_sql(config)}, + class_l1_domain, + class_l2_filter_group, + filter_logic, + logic_distance_km, + selected_component_class, + nearest_facility_uid, + nearest_facility_name, + nearest_facility_code, + nearest_facility_latitude, + nearest_facility_longitude, + nearest_class_l4_facility_subtype + from ranked + where rn = 1 + """ + + +def create_l1_view_sql(config: Dict[str, Any]) -> str: + id_col = config["proximity"]["village_id_col"] + l3_table = config["proximity"]["tables"]["l3"] + village_table = village_shapes_table(config) + map_table = class_map_table(config) + lookup_table = nearest_lookup_table(config) + return f""" + with ranked as ( + select + p.{sql_ident(id_col)} as {sql_ident(id_col)}{context_select_sql(config)}, + m.class_l1_domain, + p.nearest_distance_km as closest_domain_distance_km, + m.class_l2_filter_group as selected_filter_group, + p.class_l3_facility_class as selected_component_class, + p.nearest_facility_uid, + nf.facility_name as nearest_facility_name, + nf.facility_code as nearest_facility_code, + nf.latitude as nearest_facility_latitude, + nf.longitude as nearest_facility_longitude, + nf.class_l4_facility_subtype as nearest_class_l4_facility_subtype, + row_number() over ( + partition by p.{sql_ident(id_col)}, m.class_l1_domain + order by p.nearest_distance_km asc, p.class_l3_facility_class + ) as rn + from {sql_ident(l3_table)} p + join {sql_ident(map_table)} m + on m.class_l3_facility_class = p.class_l3_facility_class + left join {sql_ident(village_table)} v + on cast(v.{sql_ident(id_col)} as text) = cast(p.{sql_ident(id_col)} as text) + left join {sql_ident(lookup_table)} nf + on nf.facility_uid = p.nearest_facility_uid + ) + select + {sql_ident(id_col)}{outer_context_sql(config)}, + class_l1_domain, + closest_domain_distance_km, + selected_filter_group, + selected_component_class, + nearest_facility_uid, + nearest_facility_name, + nearest_facility_code, + nearest_facility_latitude, + nearest_facility_longitude, + nearest_class_l4_facility_subtype + from ranked + where rn = 1 + """ + + +def materialized_insert_columns(config: Dict[str, Any], metric_columns: List[str]) -> str: + id_col = config["proximity"]["village_id_col"] + columns = [id_col, *(config["proximity"].get("village_context_cols") or []), *metric_columns] + return ", ".join(sql_ident(column) for column in columns) + + +def materialized_context_sql(config: Dict[str, Any]) -> str: + parts = [] + for col in config["proximity"].get("village_context_cols") or []: + parts.append(f", v.{sql_ident(col)}") + return "".join(parts) + + +def ensure_l3_derived_indexes(con: sqlite3.Connection, config: Dict[str, Any]) -> None: + id_col = config["proximity"]["village_id_col"] + l3_table = config["proximity"]["tables"]["l3"] + con.execute( + f"create index if not exists idx_{l3_table}_class_village_distance " + f"on {sql_ident(l3_table)}(class_l3_facility_class, {sql_ident(id_col)}, nearest_distance_km)" + ) + con.execute( + f"create index if not exists idx_{l3_table}_village_class " + f"on {sql_ident(l3_table)}({sql_ident(id_col)}, class_l3_facility_class)" + ) + + +def create_empty_materialized_tables(con: sqlite3.Connection, config: Dict[str, Any]) -> None: + tables = config["proximity"]["tables"] + l2_sql = create_l2_view_sql(config) + l1_sql = create_l1_view_sql(config) + con.execute(f"create table {sql_ident(tables['l2_materialized'])} as {l2_sql} limit 0") + con.execute(f"create table {sql_ident(tables['l1_materialized'])} as {l1_sql} limit 0") + + +def materialize_l2_groups(con: sqlite3.Connection, config: Dict[str, Any]) -> None: + id_col = config["proximity"]["village_id_col"] + tables = config["proximity"]["tables"] + l3_table = tables["l3"] + village_table = village_shapes_table(config) + map_table = class_map_table(config) + lookup_table = nearest_lookup_table(config) + output_table = tables["l2_materialized"] + output_columns = materialized_insert_columns( + config, + [ + "class_l1_domain", + "class_l2_filter_group", + "filter_logic", + "logic_distance_km", + "selected_component_class", + "nearest_facility_uid", + "nearest_facility_name", + "nearest_facility_code", + "nearest_facility_latitude", + "nearest_facility_longitude", + "nearest_class_l4_facility_subtype", + ], + ) + context_sql = materialized_context_sql(config) + groups = class_map_frame(config)[["class_l1_domain", "class_l2_filter_group", "filter_logic"]].drop_duplicates() + for row in groups.itertuples(index=False): + aggregate = "max" if row.filter_logic == "max" else "min" + con.execute( + f""" + insert into {sql_ident(output_table)} ({output_columns}) + with target as ( + select + p.{sql_ident(id_col)} as join_id, + {aggregate}(p.nearest_distance_km) as logic_distance_km + from {sql_ident(l3_table)} p + join {sql_ident(map_table)} m + on m.class_l3_facility_class = p.class_l3_facility_class + where m.class_l1_domain = ? + and m.class_l2_filter_group = ? + group by p.{sql_ident(id_col)} + ), + selected as ( + select + p.{sql_ident(id_col)} as join_id, + min(p.class_l3_facility_class) as selected_component_class + from {sql_ident(l3_table)} p + join {sql_ident(map_table)} m + on m.class_l3_facility_class = p.class_l3_facility_class + join target t + on t.join_id = p.{sql_ident(id_col)} + and t.logic_distance_km = p.nearest_distance_km + where m.class_l1_domain = ? + and m.class_l2_filter_group = ? + group by p.{sql_ident(id_col)} + ) + select + t.join_id{context_sql}, + ? as class_l1_domain, + ? as class_l2_filter_group, + ? as filter_logic, + t.logic_distance_km, + s.selected_component_class, + p.nearest_facility_uid, + nf.facility_name as nearest_facility_name, + nf.facility_code as nearest_facility_code, + nf.latitude as nearest_facility_latitude, + nf.longitude as nearest_facility_longitude, + nf.class_l4_facility_subtype as nearest_class_l4_facility_subtype + from target t + join selected s + on s.join_id = t.join_id + join {sql_ident(l3_table)} p + on p.{sql_ident(id_col)} = t.join_id + and p.class_l3_facility_class = s.selected_component_class + and p.nearest_distance_km = t.logic_distance_km + left join {sql_ident(village_table)} v + on v.{sql_ident(id_col)} = t.join_id + left join {sql_ident(lookup_table)} nf + on nf.facility_uid = p.nearest_facility_uid + """, + ( + row.class_l1_domain, + row.class_l2_filter_group, + row.class_l1_domain, + row.class_l2_filter_group, + row.class_l1_domain, + row.class_l2_filter_group, + row.filter_logic, + ), + ) + con.commit() + log.info("Materialized L2 group %s/%s", row.class_l1_domain, row.class_l2_filter_group) + + +def materialize_l1_groups(con: sqlite3.Connection, config: Dict[str, Any]) -> None: + id_col = config["proximity"]["village_id_col"] + tables = config["proximity"]["tables"] + l3_table = tables["l3"] + village_table = village_shapes_table(config) + map_table = class_map_table(config) + lookup_table = nearest_lookup_table(config) + output_table = tables["l1_materialized"] + output_columns = materialized_insert_columns( + config, + [ + "class_l1_domain", + "closest_domain_distance_km", + "selected_filter_group", + "selected_component_class", + "nearest_facility_uid", + "nearest_facility_name", + "nearest_facility_code", + "nearest_facility_latitude", + "nearest_facility_longitude", + "nearest_class_l4_facility_subtype", + ], + ) + context_sql = materialized_context_sql(config) + domains = class_map_frame(config)["class_l1_domain"].drop_duplicates().tolist() + for domain in domains: + con.execute( + f""" + insert into {sql_ident(output_table)} ({output_columns}) + with target as ( + select + p.{sql_ident(id_col)} as join_id, + min(p.nearest_distance_km) as closest_domain_distance_km + from {sql_ident(l3_table)} p + join {sql_ident(map_table)} m + on m.class_l3_facility_class = p.class_l3_facility_class + where m.class_l1_domain = ? + group by p.{sql_ident(id_col)} + ), + selected as ( + select + p.{sql_ident(id_col)} as join_id, + min(p.class_l3_facility_class) as selected_component_class + from {sql_ident(l3_table)} p + join {sql_ident(map_table)} m + on m.class_l3_facility_class = p.class_l3_facility_class + join target t + on t.join_id = p.{sql_ident(id_col)} + and t.closest_domain_distance_km = p.nearest_distance_km + where m.class_l1_domain = ? + group by p.{sql_ident(id_col)} + ) + select + t.join_id{context_sql}, + ? as class_l1_domain, + t.closest_domain_distance_km, + m.class_l2_filter_group as selected_filter_group, + s.selected_component_class, + p.nearest_facility_uid, + nf.facility_name as nearest_facility_name, + nf.facility_code as nearest_facility_code, + nf.latitude as nearest_facility_latitude, + nf.longitude as nearest_facility_longitude, + nf.class_l4_facility_subtype as nearest_class_l4_facility_subtype + from target t + join selected s + on s.join_id = t.join_id + join {sql_ident(l3_table)} p + on p.{sql_ident(id_col)} = t.join_id + and p.class_l3_facility_class = s.selected_component_class + and p.nearest_distance_km = t.closest_domain_distance_km + join {sql_ident(map_table)} m + on m.class_l3_facility_class = p.class_l3_facility_class + left join {sql_ident(village_table)} v + on v.{sql_ident(id_col)} = t.join_id + left join {sql_ident(lookup_table)} nf + on nf.facility_uid = p.nearest_facility_uid + """, + (domain, domain, domain), + ) + con.commit() + log.info("Materialized L1 domain %s", domain) + + +def refresh_derived_outputs(output_gpkg: Path, config: Dict[str, Any], materialize: bool) -> None: + tables = config["proximity"]["tables"] + write_class_map(output_gpkg, config) + l2_sql = create_l2_view_sql(config) + l1_sql = create_l1_view_sql(config) + with sqlite3.connect(output_gpkg) as con: + drop_derived_outputs(con, config) + ensure_proximity_indexes(con, config) + if config["proximity"].get("create_derived_views", True): + con.execute(f"create view {sql_ident(tables['l2_view'])} as {l2_sql}") + con.execute(f"create view {sql_ident(tables['l1_view'])} as {l1_sql}") + if materialize: + ensure_l3_derived_indexes(con, config) + create_empty_materialized_tables(con, config) + con.commit() + materialize_l2_groups(con, config) + materialize_l1_groups(con, config) + con.execute( + f"create index if not exists idx_{tables['l2_materialized']}_village " + f"on {sql_ident(tables['l2_materialized'])}({sql_ident(config['proximity']['village_id_col'])})" + ) + con.execute( + f"create index if not exists idx_{tables['l1_materialized']}_village " + f"on {sql_ident(tables['l1_materialized'])}({sql_ident(config['proximity']['village_id_col'])})" + ) + register_attribute_table(con, tables["l2_materialized"]) + register_attribute_table(con, tables["l1_materialized"]) + con.commit() + + +def write_proximity_metadata(output_gpkg: Path, config: Dict[str, Any], completed: int, total_groups: int) -> None: + rows = pd.DataFrame( + [ + {"key": "stored_level", "value": config["proximity"]["stored_level"]}, + {"key": "l2_l1_method", "value": "derived_from_proximity_l3"}, + {"key": "class_map_table", "value": class_map_table(config)}, + {"key": "village_layer", "value": village_shapes_table(config)}, + {"key": "nearest_facility_lookup_table", "value": nearest_lookup_table(config)}, + {"key": "l3_groups_completed", "value": str(completed)}, + {"key": "l3_groups_total", "value": str(total_groups)}, + {"key": "materialized_l2_table", "value": str(config["proximity"]["tables"].get("l2_materialized"))}, + {"key": "derive_l1_from_l3", "value": str(config["proximity"].get("derive_l1_from_l3", False))}, + ] + ) + with sqlite3.connect(output_gpkg) as con: + rows.to_sql("proximity_metadata", con, if_exists="replace", index=False) + register_attribute_table(con, "proximity_metadata") + con.commit() + + +def run_proximity( + config_path: Path, + classes: Optional[str] = None, + sample_villages: Optional[int] = None, + sample_classes: Optional[int] = None, + village_chunksize: int = 100_000, + output_gpkg: Optional[Path] = None, + materialize_derived: bool = False, + no_derived_views: bool = False, + refresh_derived_only: bool = False, + skip_monitor: bool = False, + force: bool = False, + debug: bool = False, +) -> None: + repo_root = find_repo_root(config_path.resolve()) + config = read_yaml(config_path) + setup_logging(debug, resolve_path(config["pipeline"]["log_path"], repo_root)) + if no_derived_views: + config["proximity"]["create_derived_views"] = False + effective_materialize = materialize_derived or bool(config["proximity"].get("materialize_derived_tables", False)) + config["proximity"]["materialize_derived_tables"] = effective_materialize + facilities_gpkg = resolve_path(config["outputs"]["facilities_gpkg"], repo_root) + out_gpkg = output_gpkg or resolve_path(config["outputs"]["proximity_gpkg"], repo_root) + out_gpkg = out_gpkg if out_gpkg.is_absolute() else repo_root / out_gpkg + class_filter = [item.strip() for item in classes.split(",") if item.strip()] if classes else None + + if refresh_derived_only: + village_layer = village_shapes_table(config) + expected_villages = output_village_count(out_gpkg, village_layer) + if expected_villages is None: + raise RuntimeError( + f"Cannot refresh derived proximity outputs because {out_gpkg} " + f"has no {village_layer} layer." + ) + all_groups = l3_groups(config, facilities_gpkg, None, None) + completed_after = completed_l3_classes(out_gpkg, config, expected_villages) + write_class_map(out_gpkg, config) + if nearest_lookup_missing(out_gpkg, config): + rebuild_nearest_lookup(out_gpkg, config, facilities_gpkg) + if len(completed_after) >= len(all_groups): + refresh_derived_outputs( + out_gpkg, + config, + materialize=effective_materialize, + ) + log.info("Refreshed L2 materialized output from existing L3 proximity rows.") + else: + log.info( + "Skipping L2 materialization until all L3 classes are complete (%d/%d complete)", + len(completed_after), + len(all_groups), + ) + write_proximity_metadata(out_gpkg, config, len(completed_after), len(all_groups)) + if not skip_monitor: + from facility_metadata_monitor import write_monitor + + monitor_path = write_monitor(config_path) + log.info("Updated metadata monitor: %s", monitor_path) + return + + villages = load_villages(config, repo_root, sample_villages) + prepare_output_gpkg(out_gpkg, villages, config, force_full=force and not class_filter) + if force and class_filter: + delete_l3_classes(out_gpkg, config, class_filter) + + groups = l3_groups(config, facilities_gpkg, class_filter, sample_classes) + done = completed_l3_classes(out_gpkg, config, len(villages)) + log.info("Prepared %d L3 proximity groups; %d already complete", len(groups), len(done)) + written = 0 + completed_now = 0 + for group in groups: + facility_class = group["class_l3_facility_class"] + if facility_class in done: + continue + points = load_l3_facility_points(config, facilities_gpkg, facility_class) + if points.empty: + log.warning("No facility points for %s", facility_class) + continue + rows = build_l3_for_group(out_gpkg, config, villages, group, points, village_chunksize) + written += rows + completed_now += 1 + log.info("Wrote L3 proximity for %s (%d facilities, %d rows)", facility_class, len(points), rows) + + with sqlite3.connect(out_gpkg) as con: + l3_table = config["proximity"]["tables"]["l3"] + if table_exists(con, l3_table): + register_attribute_table(con, l3_table) + ensure_proximity_indexes(con, config) + con.commit() + write_class_map(out_gpkg, config) + all_groups = l3_groups(config, facilities_gpkg, None, None) + completed_after = completed_l3_classes(out_gpkg, config, len(villages)) + full_l3_complete = len(completed_after) >= len(all_groups) + if full_l3_complete: + if nearest_lookup_missing(out_gpkg, config): + rebuild_nearest_lookup(out_gpkg, config, facilities_gpkg) + refresh_derived_outputs( + out_gpkg, + config, + materialize=effective_materialize, + ) + else: + log.info( + "Skipping L2 materialization until all L3 classes are complete (%d/%d complete)", + len(completed_after), + len(all_groups), + ) + write_proximity_metadata(out_gpkg, config, len(completed_after), len(all_groups)) + log.info("Saved proximity GeoPackage: %s (%d new L3 rows)", out_gpkg, written) + if not skip_monitor: + from facility_metadata_monitor import write_monitor + + monitor_path = write_monitor(config_path) + log.info("Updated metadata monitor: %s", monitor_path) + + +def run_from_args(args: argparse.Namespace) -> None: + run_proximity( + config_path=args.config, + classes=args.classes, + sample_villages=args.sample_villages, + sample_classes=args.sample_classes, + village_chunksize=args.village_chunksize, + output_gpkg=args.output_gpkg, + materialize_derived=args.materialize_derived, + no_derived_views=args.no_derived_views, + refresh_derived_only=args.refresh_derived_only, + skip_monitor=args.skip_monitor, + force=getattr(args, "force", False), + debug=args.debug, + ) + + +def build_parser() -> argparse.ArgumentParser: + repo_root = find_repo_root(Path(__file__).resolve()) + parser = argparse.ArgumentParser(description="Build lean L3-based village-to-facility proximity GPKG.") + parser.add_argument( + "--config", + type=Path, + default=repo_root / "utilities" / "scripts" / "facilities_utils" / "config" / "facilities_master.yaml", + ) + parser.add_argument("--classes", type=str, default=None, help="Comma-separated L3 classes to compute.") + parser.add_argument("--sample-villages", type=int, default=None) + parser.add_argument("--sample-classes", type=int, default=None) + parser.add_argument("--village-chunksize", type=int, default=100_000) + parser.add_argument("--output-gpkg", type=Path, default=None) + parser.add_argument("--materialize-derived", action="store_true", help="Legacy flag; L2 is materialized by default when all L3 classes are complete.") + parser.add_argument("--no-derived-views", action="store_true") + parser.add_argument("--refresh-derived-only", action="store_true", help="Rebuild class-map and materialized L2 output from existing L3 rows.") + parser.add_argument("--skip-monitor", action="store_true", help="Skip metadata monitor generation for this run.") + parser.add_argument("--force", action="store_true") + parser.add_argument("--debug", action="store_true") + return parser + + +def main() -> None: + run_from_args(build_parser().parse_args()) + + +if __name__ == "__main__": + main() From dbdca1bcb6852eb41f1ef67a9253d744edd6126b Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:28 +0000 Subject: [PATCH 5/7] Configure facilities build and proximity outputs This YAML is the tracked source of truth for raw facility source definitions, output schema, taxonomy, and proximity settings. It is required to reproduce pan_india_facilities.gpkg and cs_village_facility_proximity.gpkg. --- .../config/facilities_master.yaml | 550 ++++++++++++++++++ 1 file changed, 550 insertions(+) create mode 100644 utilities/scripts/facilities_utils/config/facilities_master.yaml diff --git a/utilities/scripts/facilities_utils/config/facilities_master.yaml b/utilities/scripts/facilities_utils/config/facilities_master.yaml new file mode 100644 index 00000000..8d7c4af9 --- /dev/null +++ b/utilities/scripts/facilities_utils/config/facilities_master.yaml @@ -0,0 +1,550 @@ +pipeline: + name: pan_india_facilities + raw_dir: data/facilities/raw + reference_dir: data/facilities/reference + intermediate_dir: data/facilities/intermediate + output_dir: data/facilities/outputs + state_path: data/facilities/state/pipeline_state.json + log_path: data/facilities/state/pipeline.log + +schema: + crs: EPSG:4326 + class_k_count: 8 + facility_layer: facilities + membership_table: facility_memberships + taxonomy_table: facility_taxonomy + source_summary_table: source_summary + invalid_table: invalid_coordinates + facility_columns: + - facility_uid + - latitude + - longitude + - facility_name + - facility_code + - 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 + - urban_rural + - pincode + - establishment_year + - district_lgd + - village_census11 + - village_name + - membership_count + membership_columns: + - facility_uid + - 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 + dropped_from_facility_layer: + - source_dataset + - source_clean_file + - source_clean_file_list + - coordinate_status + - class_l1_domain_list + - class_l2_filter_group_list + - class_l3_facility_class_list + - class_l4_facility_subtype_list + - class_k_list_columns + - facility_type + - facility_type_code + - source_category_detail + - management + - management_code + - ownership + +outputs: + facilities_csv: data/facilities/outputs/pan_india_facilities.csv + memberships_csv: data/facilities/outputs/facility_memberships.csv + taxonomy_csv: data/facilities/outputs/facility_taxonomy.csv + source_summary_csv: data/facilities/outputs/source_summary.csv + invalid_coordinates_csv: data/facilities/outputs/invalid_coordinates.csv + facilities_gpkg: data/facilities/outputs/pan_india_facilities.gpkg + proximity_gpkg: data/facilities/outputs/cs_village_facility_proximity.gpkg + metadata_monitor_yaml: utilities/scripts/facilities_utils/config/facilities_metadata_monitor.yaml + +references: + pincode_centroids: data/facilities/reference/india_pincodes_centroid.csv + +proximity: + village_gpkg: data/admin-boundary/cs_admin_standard.gpkg + village_layer: cs_admin_standard + village_id_col: cs_feature_id + village_context_cols: + - state_name + - district_name + - TEHSIL + - pc11_village_id + - NAME + backend: kdtree + stored_level: l3_facility_class + derive_l2_from_l3: true + derive_l1_from_l3: false + create_derived_views: true + materialize_derived_tables: true + write_village_shapes: true + output_mode: l3_base_with_derived_views + tables: + village_shapes: village_shapes + village_points: village_shapes + class_map: proximity_class_map + nearest_facilities: proximity_nearest_facilities + l3: proximity_l3 + l2_view: proximity_l2 + l1_view: proximity_l1 + l2_materialized: proximity_l2_materialized + l1_materialized: proximity_l1_materialized + distance_precision: 6 + l3_base_columns: + - village_id + - class_l3_facility_class + - nearest_distance_km + - nearest_facility_uid + nearest_facility_lookup_columns: + - facility_uid + - facility_name + - facility_code + - latitude + - longitude + - class_l4_facility_subtype + l1_l2_derivation: + class_map_table: proximity_class_map + method: Rebuild proximity_class_map from taxonomy in this YAML, then derive L1/L2 from proximity_l3. + +filter_logic: + education: + essential_education: max + higher_education: min + health: + essential_health: max + advanced_health: min + food_security: + essential_services: direct + finance: + financial_inclusion: max + agriculture: + apmc_access: min + post_harvest: min + cooperative: direct + livestock: direct + agri_support_infra: direct + +mappings: + school: + category_labels: + 1: Primary + 2: Primary with Upper Primary + 3: Pr. with Up.Pr. Sec. and H.Sec. + 4: Upper Primary only + 5: Up. Pr. Secondary and Higher Sec + 6: Pr. Up Pr. and Secondary Only + 7: Upper Pr. and Secondary + 8: Secondary Only + 10: Secondary with Higher Secondary + 11: Higher Secondary only/Jr. College + 12: Pre-Primary Only + category_to_classes: + 1: [school_primary] + 2: [school_primary, school_upper_primary] + 3: [school_primary, school_upper_primary, school_secondary, school_higher_secondary] + 4: [school_upper_primary] + 5: [school_upper_primary, school_secondary, school_higher_secondary] + 6: [school_primary, school_upper_primary, school_secondary] + 7: [school_upper_primary, school_secondary] + 8: [school_secondary] + 10: [school_secondary, school_higher_secondary] + 11: [school_higher_secondary] + 12: [school_primary] + apmc: + category_normalise: + Farmers Market/Village Market: FCM + rural market: FCM + Others: Other + "": Other + category_labels: + PMY: Primary Market Yard + SMY: Sub Market Yard + RPM: Regulated Primary Market + RSM: Regulated Sub Market + NRM: Non-Regulated Market + FCM: Farmers Consumer Market + Other: Other + agri_industry: + category_to_class: + Markets & Trading: agri_industry_markets_trading + Storage & Warehousing: agri_industry_storage_warehousing + Distribution & Utilities: agri_industry_distribution_utilities + Agri-Processing: agri_industry_agri_processing + Industrial Manufacturing: agri_industry_industrial_manufacturing + Co-operatives & Societies: agri_industry_co_operatives_societies + Dairy & Animal Husbandry: agri_industry_dairy_animal_husbandry + Agri-Support & Infrastructure: agri_industry_agri_support_infrastructure + subtype_categories: + Agri-Processing: + - Agro Industry + - Aqua Industry + - Bakery + - Cashew Industry + - Coconut Industry + - Dal Mill + - Lamon grass oil production ce* + - Maize Milling Unit + - Paraboiled Industries + - Rice/Flour/Oil Mill + - Sugar Mill + - Tea Industry + Agri-Support & Infrastructure: + - Fertilizer Depot + - Green House + - IFFCO Centre + - Krishi Bhavan + - Pesticide Depot + - Rice Research Centre + - Soil Testing Lab + - Sugar Cane Centre + Co-operatives & Societies: + - Agricultural Co operative Soc* + - Co operative Society + - Farmer Welfare Society + - MPCS + - Primary Agricultural Cooperat + Dairy & Animal Husbandry: + - Animal Epicenter + - Animal Husbandry + - Cattle Farm + - Cattle Feed + - Chicken Center + - Cow Farm + - Dairy Farm + - Fish Farm + - Fisheries + - Goat Farm + - Honey Farm + - Kshreera Bhavan + - Milk Centre + - Milk Chilling Centre + - Milk Co Operative Society + - Pig Farm + - Poultry Farm + - Pultry Farm + - Sheep Farm + Distribution & Utilities: + - Civil supply Corporation + - Distribution Centre + - Echoshop + - Electricity Distribution Cent* + - Food Corporation of India + - Indian Gas Distribution + Industrial Manufacturing: + - Brick Industry + - Camphor Manufacturing Unit + - Coir Industry + - Cotton Industry + - Jute Industry + - Latex Industry + - Resham Farm + - Rubber Industry + - Rubber Processing Unit + - Saw Mill + - Silk Farm + - Tobaco Industry + - Wood Industry + Markets & Trading: + - APMC + - Bazaar + - Collection Centre + - Dhan Kharidi Kendra + - Direct Purchase Centre + - Fish Mart + - GRaM + - GRaM (Notified) + - Mandi + - Mandi (Notified) + - Market + - PACCS GRaM + - Vegetable /fruit store and Shop + Storage & Warehousing: + - Cold Storage + - FCI Godown + - Fish Storage + - GSSS Godown + - Godown + - LPG Godown + - Warehouse + +taxonomy: + education: + essential_education: + school_primary: + subtypes: + - Primary + - Primary with Upper Primary + - Pr. Up Pr. and Secondary Only + - Pr. with Up.Pr. Sec. and H.Sec. + - Pre-Primary Only + school_upper_primary: + subtypes: + - Primary with Upper Primary + - Upper Primary only + - Pr. Up Pr. and Secondary Only + - Pr. with Up.Pr. Sec. and H.Sec. + - Upper Pr. and Secondary + - Up. Pr. Secondary and Higher Sec + school_secondary: + subtypes: + - Pr. Up Pr. and Secondary Only + - Pr. with Up.Pr. Sec. and H.Sec. + - Upper Pr. and Secondary + - Up. Pr. Secondary and Higher Sec + - Secondary Only + - Secondary with Higher Secondary + higher_education: + school_higher_secondary: + subtypes: + - Pr. with Up.Pr. Sec. and H.Sec. + - Up. Pr. Secondary and Higher Sec + - Secondary with Higher Secondary + - Higher Secondary only/Jr. College + college: + subtypes: + - Affiliated College + - Constituent / University College + - Recognized Center + - PG Center / Off-Campus Center + - Autonomous College + universities: + subtypes: + - Technical/Polytechnic + - Nursing + - Teacher Training + - Paramedical + - PGDM Institutes + - Institutes under Ministries + - Hotel Management and Catering + health: + essential_health: + health_sub_cen: + subtypes: [] + health_phc: + subtypes: [] + advanced_health: + health_chc: + subtypes: [] + health_dis_h: + subtypes: [] + health_s_t_h: + subtypes: [] + food_security: + essential_services: + pds: + subtypes: [] + finance: + financial_inclusion: + csc: + subtypes: [] + bank_mitra: + subtypes: [] + bank_branch: + subtypes: [] + bank_atm: + subtypes: [] + agriculture: + apmc_access: + apmc: + subtypes: + - Primary Market Yard + - Sub Market Yard + - Other + - Regulated Primary Market + - Regulated Sub Market + - Farmers Consumer Market + - Non-Regulated Market + agri_industry_markets_trading: + subtypes: [] + post_harvest: + agri_industry_storage_warehousing: + subtypes: [] + agri_industry_distribution_utilities: + subtypes: [] + agri_industry_agri_processing: + subtypes: [] + agri_industry_industrial_manufacturing: + subtypes: [] + cooperative: + agri_industry_co_operatives_societies: + subtypes: [] + livestock: + agri_industry_dairy_animal_husbandry: + subtypes: [] + agri_support_infra: + agri_industry_agri_support_infrastructure: + subtypes: [] + +sources: + school: + active: true + processor: school_mhrd + raw_files: + - School__Source___MHRD_2023__4.csv + facility_uid_prefix: school + class_k: + class_k1: + source: management + label: school_management + class_k2: + source: schmgt + label: school_management_code + college: + active: true + processor: table + raw_files: [college.csv] + facility_class: college + facility_uid_prefix: college + id_col: aishecode + name_col: collegeins + lat_col: latitude + lon_col: longitude + subtype_col: collegetyp + facility_code_col: affiliatin + urban_rural_col: locationid + pincode_col: pincode + establishment_year_col: yearofesta + district_lgd_col: districtlg + clean_text_cols: [collegeins, affiliatin] + class_k: + class_k1: + source: management + label: management + class_k2: + source: ownerships + label: ownership + universities: + active: true + processor: table + raw_files: [universities.csv] + facility_class: universities + facility_uid_prefix: universities + id_col: aishecode + name_col: name + lat_col: latitude + lon_col: longitude + subtype_col: typeofinst + facility_code_col: aishe_code + urban_rural_col: locationid + pincode_col: pincode + establishment_year_col: yearofesta + district_lgd_col: districtlg + clean_text_cols: [name] + class_k: + class_k1: + source: management + label: management + class_k2: + source: ownerships + label: ownership + health_center: + active: true + processor: health_center_split + raw_files: + - Health_Center_Source___Data_gov_2022__6.csv + facility_uid_prefix: health + type_to_class: + sub_cen: health_sub_cen + phc: health_phc + chc: health_chc + dis_h: health_dis_h + s_t_h: health_s_t_h + pds: + active: true + processor: table + raw_files: [PDS__Sources___DoFD_2023__7.csv] + facility_class: pds + facility_uid_prefix: pds + id_col: fps_code + name_col: fpsname + lat_col: latitude + lon_col: longitude + facility_code_col: fps_code + village_census11_col: villagecod + village_name_col: villagenam + uid_strategy: id_coord + csc: + active: true + processor: table + raw_files: [CSC-MeitY-2022.csv] + facility_class: csc + facility_uid_prefix: csc + id_col: csc_id + lat_col: latitude + lon_col: longitude + uid_strategy: id_coord + bank_mitra: + active: true + processor: table + raw_files: [bank_mitra.csv] + facility_class: bank_mitra + facility_uid_prefix: bank_mitra + id_col: objectid + name_col: bank_name + lat_col: bk_m_lat + lon_col: bk_m_long + coord_fallback_col: coordinates + facility_code_col: bk_mitr_cd + clean_text_cols: [bank_name] + bank_branch: + active: true + processor: table + raw_files: [bank_branch.csv] + facility_class: bank_branch + facility_uid_prefix: bank_branch + id_col: objectid + name_col: bank_name + lat_col: br_lat + lon_col: br_long + facility_code_col: br_ifsc_cd + pincode_col: br_pin_cod + clean_text_cols: [bank_name] + bank_atm: + active: true + processor: table + raw_files: [bank_atm.csv] + facility_class: bank_atm + facility_uid_prefix: bank_atm + id_col: objectid + name_col: bank_name + lat_col: atm_lat + lon_col: atm_long + facility_code_col: atm_cd + pincode_col: atm_pin_cd + clean_text_cols: [bank_name] + apmc: + active: true + processor: apmc + raw_files: [apmc.csv] + facility_class: apmc + facility_uid_prefix: apmc + agri_industry: + active: true + processor: agri_industry_reclassified + raw_files: + - Industries__Sources___PMGSY_2023__11.csv + - Agri_Resources__Sources___PMGSY_2023__10.csv + facility_uid_prefix: agri_industry From ccfedd11c1102068dfb7db34fc691b358839dbeb Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:29 +0000 Subject: [PATCH 6/7] Add template for new facilities sources The template documents the config block required to add new raw facility datasets safely without editing pipeline code. --- .../config/new_dataset_schema_template.yaml | 37 +++++++++++++++++++ 1 file changed, 37 insertions(+) create mode 100644 utilities/scripts/facilities_utils/config/new_dataset_schema_template.yaml diff --git a/utilities/scripts/facilities_utils/config/new_dataset_schema_template.yaml b/utilities/scripts/facilities_utils/config/new_dataset_schema_template.yaml new file mode 100644 index 00000000..2ef8cb02 --- /dev/null +++ b/utilities/scripts/facilities_utils/config/new_dataset_schema_template.yaml @@ -0,0 +1,37 @@ +new_source_key: + active: true + processor: table + raw_files: + - new_raw_file.csv + facility_class: add_matching_class_from_taxonomy + facility_uid_prefix: short_stable_prefix + id_col: source_unique_id_column + lat_col: latitude_column + lon_col: longitude_column + + # Optional useful row properties. + name_col: null + subtype_col: null + facility_code_col: null + urban_rural_col: null + pincode_col: null + establishment_year_col: null + district_lgd_col: null + village_census11_col: null + village_name_col: null + coord_fallback_col: null + uid_strategy: id + + # Optional text cleanup. + clean_text_cols: [] + + # Optional source-specific classes/fields to keep in class_k columns. + # Use only compact, useful fields. Do not duplicate class_l4_facility_subtype + # or standard columns such as urban_rural, pincode, and district_lgd. + class_k: + class_k1: + source: source_column_name + label: human_readable_meaning + +# If the source needs short code-to-label or subtype-to-class rules, add them +# under mappings in facilities_master.yaml. Keep large lookup tables as files. From 3cf26731d1bb9379928514dd2c43920def633d16 Mon Sep 17 00:00:00 2001 From: amit-spatial Date: Tue, 7 Jul 2026 10:40:30 +0000 Subject: [PATCH 7/7] Document facilities build workflow The README explains how to run the facilities build and proximity steps, what outputs are expected, and how to add new sources without tracking raw files or generated outputs. --- utilities/scripts/facilities_utils/README.md | 120 +++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 utilities/scripts/facilities_utils/README.md diff --git a/utilities/scripts/facilities_utils/README.md b/utilities/scripts/facilities_utils/README.md new file mode 100644 index 00000000..21786f94 --- /dev/null +++ b/utilities/scripts/facilities_utils/README.md @@ -0,0 +1,120 @@ +# Facilities Pipeline + +This directory contains the standalone scripts for building and maintaining the +pan-India facilities dataset. The source of truth is: + +- `data/facilities/raw/` +- `utilities/scripts/facilities_utils/config/facilities_master.yaml` + +*Facilities asset creation pipeline scripts do not depend on Django.* + +## Main CLI + +```bash +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio --with scipy \ + python utilities/scripts/facilities_utils/facility_pipeline.py all +``` + +Useful subcommands: + +```bash +# Process raw source files into per-source intermediates. +uv run --with pandas --with numpy --with pyyaml \ + python utilities/scripts/facilities_utils/facility_pipeline.py clean + +# Build lean CSV outputs and data/facilities/outputs/pan_india_facilities.gpkg. +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio \ + python utilities/scripts/facilities_utils/facility_pipeline.py build + +# Build data/facilities/outputs/cs_village_facility_proximity.gpkg. +# This stores L3 proximity as the durable base and derives L2/L1 from L3. +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio --with scipy \ + python utilities/scripts/facilities_utils/facility_pipeline.py proximity --materialize-derived +``` + +## Outputs + +`data/facilities/outputs/pan_india_facilities.gpkg` + +- `facilities`: one valid facility point per physical facility. +- `facility_memberships`: non-spatial class membership table. +- `facility_taxonomy`: non-spatial taxonomy and filter-logic table. +- `source_summary`: non-spatial source QA table. +- `invalid_coordinates`: non-spatial invalid-coordinate audit table. + +`data/facilities/outputs/cs_village_facility_proximity.gpkg` + +- `village_shapes`: actual village polygons from `cs_admin_standard` with representative-point latitude/longitude helper columns. +- `proximity_l3`: lean durable base table with village id, L3 class, distance, and nearest facility uid. +- `proximity_l2_materialized`: lean physical L2 table derived from L3 using the configured min/max group logic. +- `proximity_nearest_facilities`: deduplicated facility detail lookup for nearest facilities referenced by L3. +- `proximity_class_map`: small YAML-derived class map from L3 to L2/L1/filter logic. +- `proximity_l2`: SQL view derived from L3 plus `proximity_class_map`. +- `proximity_l1`: SQL view derived from L3 plus `proximity_class_map`. +- Materialized L1/L2 tables for API and GeoServer export after all L3 classes are complete. + +The L1/L2 views are retained as rebuildable derivations. The API source asset +also keeps physical L1/L2 tables for repeated local reads. + +## Adding Data + +Use `utilities/scripts/facilities_utils/config/new_dataset_schema_template.yaml` as the starting +point for a new source block. Add the block under `sources` in +`utilities/scripts/facilities_utils/config/facilities_master.yaml`, add the class under `taxonomy` +if needed, then run: + +```bash +uv run --with pandas --with numpy --with pyyaml \ + python utilities/scripts/facilities_utils/facility_pipeline.py clean --source new_source_key +``` + +Then rebuild the GPKG: + +```bash +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio \ + python utilities/scripts/facilities_utils/facility_pipeline.py build +``` + +Recompute one changed L3 class without deleting the whole proximity asset: + +```bash +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio --with scipy \ + python utilities/scripts/facilities_utils/facility_pipeline.py proximity \ + --classes school_primary --force +``` + +Refresh L2 output after only taxonomy/filter-logic YAML changes: + +```bash +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio --with scipy \ + python utilities/scripts/facilities_utils/facility_pipeline.py proximity --refresh-derived-only +``` + +For repeated class-by-class proximity rebuilds, skip monitor generation until the +last run: + +```bash +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio --with scipy \ + python utilities/scripts/facilities_utils/facility_pipeline.py proximity \ + --classes school_primary --force --skip-monitor + +uv run --with pandas --with pyyaml \ + python utilities/scripts/facilities_utils/facility_pipeline.py monitor +``` + +## Smoke Tests + +Keep smoke outputs separate: + +```bash +uv run --with pandas --with numpy --with pyyaml \ + python utilities/scripts/facilities_utils/facility_pipeline.py clean --sample-rows 1000 --force + +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio \ + python utilities/scripts/facilities_utils/facility_pipeline.py build --gpkg-chunksize 5000 + +uv run --with pandas --with numpy --with pyyaml --with geopandas --with shapely --with pyogrio --with scipy \ + python utilities/scripts/facilities_utils/facility_pipeline.py proximity \ + --sample-villages 5 --sample-classes 3 \ + --output-gpkg /tmp/cs_village_facility_proximity.gpkg +```