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
+```
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
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.
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)),
+    }
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()
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()
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, "<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()