Spaces:
Runtime error
Runtime error
File size: 4,380 Bytes
d64c823 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """
Orchestrator: run all data processing steps in sequence and report status.
Usage: python -m src.data_processing.process_all
"""
import logging
import time
import sys
from pathlib import Path
logger = logging.getLogger(__name__)
def process_all():
"""Run all data processing steps in order."""
start = time.time()
print("=" * 60)
print(" InstaWarn — Data Processing Pipeline")
print("=" * 60)
print()
results = {}
# ── Step 1: Administrative Boundaries ─────────────────────────────────
print("[1/5] Processing administrative boundaries...")
try:
from src.data_processing.load_boundaries import load_and_filter_boundaries
unions = load_and_filter_boundaries()
results["boundaries"] = f"✅ {len(unions)} unions loaded"
except Exception as e:
results["boundaries"] = f"❌ Failed: {e}"
print(f" ERROR: {e}")
print(" This is a BLOCKER — cannot continue without boundaries.")
_print_summary(results, start)
return False
print()
# ── Step 2: Cyclone Track ─────────────────────────────────────────────
print("[2/5] Processing cyclone track data...")
try:
from src.data_processing.load_cyclone_track import load_cyclone_track
track = load_cyclone_track()
results["cyclone_track"] = f"✅ {len(track)} track points"
except Exception as e:
results["cyclone_track"] = f"❌ Failed: {e}"
print(f" ERROR: {e}")
print()
# ── Step 3: OSM Infrastructure ────────────────────────────────────────
print("[3/5] Extracting OSM infrastructure data...")
try:
from src.data_processing.extract_osm_data import extract_all_osm_data
osm = extract_all_osm_data()
counts = {k: len(v) for k, v in osm.items()}
results["osm_data"] = f"✅ {counts}"
except Exception as e:
results["osm_data"] = f"⚠️ Partial/Failed: {e}"
print(f" WARNING: {e}")
print()
# ── Step 4: Population ────────────────────────────────────────────────
print("[4/5] Processing population data...")
try:
from src.data_processing.load_population import load_population
pop = load_population()
total = pop["population"].sum()
results["population"] = f"✅ {len(pop)} unions, total pop: {total:,}"
except Exception as e:
results["population"] = f"⚠️ Failed: {e}"
print(f" WARNING: {e}")
print()
# ── Step 5: Shelters ──────────────────────────────────────────────────
print("[5/5] Processing shelter data...")
try:
from src.data_processing.load_shelters import load_shelters
shelters = load_shelters()
total_cap = shelters["capacity"].sum()
results["shelters"] = f"✅ {len(shelters)} shelters, capacity: {total_cap:,}"
except Exception as e:
results["shelters"] = f"⚠️ Failed: {e}"
print(f" WARNING: {e}")
print()
_print_summary(results, start)
return True
def _print_summary(results, start_time):
"""Print processing summary."""
elapsed = time.time() - start_time
print("=" * 60)
print(" Processing Summary")
print("=" * 60)
for step, status in results.items():
print(f" {step}: {status}")
print(f"\n Total time: {elapsed:.1f}s")
print("=" * 60)
# List processed files
processed_dir = Path(__file__).parent.parent.parent / "data" / "processed"
if processed_dir.exists():
files = sorted(processed_dir.glob("*"))
print(f"\n Processed files ({len(files)}):")
for f in files:
size_kb = f.stat().st_size / 1024
print(f" {f.name} ({size_kb:.1f} KB)")
print()
if __name__ == "__main__":
logging.basicConfig(
level=logging.INFO,
format="%(levelname)s: %(message)s",
)
success = process_all()
sys.exit(0 if success else 1)
|