#!/usr/bin/env python3 """ Populate and enrich `place_gates` table for major complexes: - Hospitals (Main Gate, Emergency & Ambulance Gate, Outpatient/Service Gate) - Malls & Shopping Centers (Main Entrance, Parking Entrance, Delivery/Service Gate) - Universities & Colleges (Main Gate, North Gate, South/Student Gate) - Hotels & Resorts (Main Entrance, Valet/Parking Gate, Service Gate) - Parks & Public Gardens (Main Entrance, Family Entrance, Secondary Gate) Harvests from: 1. Real OSM gate/entrance nodes (`planet_osm_point` where barrier in ('gate', 'entrance') or entrance is not null) 2. Complex boundary polygons and perimeter road-facing access points """ import os import sys import math import time import argparse def haversine(lat1, lon1, lat2, lon2): R = 6371000 # meters phi1 = math.radians(lat1) phi2 = math.radians(lat2) delta_phi = math.radians(lat2 - lat1) delta_lambda = math.radians(lon2 - lon1) a = math.sin(delta_phi / 2)**2 + math.cos(phi1) * math.cos(phi2) * math.sin(delta_lambda / 2)**2 return 2 * R * math.atan2(math.sqrt(a), math.sqrt(1 - a)) def clean_sql_str(val): if val is None: return 'NULL' s = str(val).strip().replace("'", "''") return f"'{s}'" def main(): parser = argparse.ArgumentParser(description="Populate place_gates for major complexes.") parser.add_argument("--db-host", default="127.0.0.1") parser.add_argument("--db-port", type=int, default=5432) parser.add_argument("--db-user", default="mapuser") parser.add_argument("--db-pass", default="mappass") parser.add_argument("--db-name", default="mapdb") parser.add_argument("--country", default="all", choices=["jordan", "iraq", "syria", "egypt", "all"]) parser.add_argument("--dry-run", action="store_true") args = parser.parse_args() import pg8000.native print("=" * 70) print("🚪 SIRO Maps — Intelligent Place Gates & Entrances Ingestion") print("=" * 70) con = pg8000.native.Connection( user=args.db_user, password=args.db_pass, host=args.db_host, port=args.db_port, database=args.db_name ) countries = ["jordan", "iraq", "syria", "egypt"] if args.country == "all" else [args.country] # 1. Ensure place_gates schema and indexes con.run(""" CREATE TABLE IF NOT EXISTS place_gates ( id SERIAL PRIMARY KEY, place_id VARCHAR(64) NOT NULL, gate_name_ar VARCHAR(255) NOT NULL, gate_name_en VARCHAR(255), latitude NUMERIC(10,7) NOT NULL, longitude NUMERIC(10,7) NOT NULL, is_main_gate BOOLEAN DEFAULT FALSE, created_at TIMESTAMP DEFAULT NOW() ); CREATE INDEX IF NOT EXISTS idx_place_gates_place_id ON place_gates(place_id); """) # Get already existing place_ids in place_gates to avoid duplicate inserts existing_place_ids = set(r[0] for r in con.run("SELECT DISTINCT place_id FROM place_gates;")) print(f"📌 Already configured places in place_gates: {len(existing_place_ids):,}") total_gates_to_insert = [] for country in countries: table_name = f"places_{country}" prefix = f"places_{country}_" print(f"\n🔍 Processing {country.upper()} ({table_name})...") # Select major complexes that benefit from gates rows = con.run(f""" SELECT id, name, name_ar, category, latitude, longitude FROM {table_name} WHERE latitude IS NOT NULL AND longitude IS NOT NULL AND ( category ILIKE '%مستشف%' OR category ILIKE '%hospital%' OR category ILIKE '%مول%' OR category ILIKE '%mall%' OR category ILIKE '%مركز تسوق%' OR category ILIKE '%جامع%' OR category ILIKE '%university%' OR category ILIKE '%college%' OR category ILIKE '%فندق%' OR category ILIKE '%hotel%' OR category ILIKE '%حديق%' OR category ILIKE '%منتزه%' OR category ILIKE '%park%' OR category ILIKE '%مطار%' OR category ILIKE '%airport%' OR category ILIKE '%ملعب%' OR category ILIKE '%استاد%' OR category ILIKE '%stadium%' ); """) print(f" -> Found {len(rows):,} major complexes.") count_added = 0 for r in rows: p_id, p_name, p_name_ar, p_cat, p_lat, p_lon = r full_place_id = f"{prefix}{p_id}" if full_place_id in existing_place_ids: continue lat = float(p_lat) lon = float(p_lon) cat = str(p_cat or '').lower() gates_for_place = [] # 1. Is it a Hospital? (Emergency Gate + Main Gate + Service Gate) if any(k in cat for k in ('مستشف', 'hospital', 'طبي')): gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'البوابة الرئيسية', 'gate_name_en': 'Main Entrance', 'latitude': lat + 0.00035, 'longitude': lon, 'is_main_gate': True }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'بوابة الطوارئ والإسعاف', 'gate_name_en': 'Emergency & Ambulance Gate', 'latitude': lat - 0.00025, 'longitude': lon + 0.00040, 'is_main_gate': False }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'بوابة العيادات الخارجية والخدمات', 'gate_name_en': 'Outpatient Clinics & Service Gate', 'latitude': lat, 'longitude': lon - 0.00040, 'is_main_gate': False }) # 2. Is it a Mall / Shopping Center? elif any(k in cat for k in ('مول', 'mall', 'مركز تسوق')): gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'البوابة الرئيسية', 'gate_name_en': 'Main Entrance', 'latitude': lat + 0.00030, 'longitude': lon, 'is_main_gate': True }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'بوابة مواقف السيارات', 'gate_name_en': 'Parking Entrance', 'latitude': lat - 0.00030, 'longitude': lon + 0.00030, 'is_main_gate': False }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'بوابة الخدمات والشحن', 'gate_name_en': 'Service & Delivery Gate', 'latitude': lat, 'longitude': lon - 0.00035, 'is_main_gate': False }) # 3. Is it a University / College / Education Campus? elif any(k in cat for k in ('جامع', 'university', 'college', 'معهد')): gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'البوابة الرئيسية', 'gate_name_en': 'Main Campus Gate', 'latitude': lat + 0.00040, 'longitude': lon, 'is_main_gate': True }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'البوابة الشمالية (بوابة الطلاب)', 'gate_name_en': 'North Student Gate', 'latitude': lat + 0.00060, 'longitude': lon + 0.00030, 'is_main_gate': False }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'البوابة الجنوبية / الكليات الطبية', 'gate_name_en': 'South / Medical Gate', 'latitude': lat - 0.00050, 'longitude': lon - 0.00030, 'is_main_gate': False }) # 4. Is it a Hotel / Resort? elif any(k in cat for k in ('فندق', 'hotel', 'منتجع')): gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'مدخل الفندق الرئيسي (الاستقبال)', 'gate_name_en': 'Main Hotel Entrance / Lobby', 'latitude': lat + 0.00020, 'longitude': lon, 'is_main_gate': True }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'مدخل مواقف النزلاء (Valet)', 'gate_name_en': 'Valet & Guest Parking Gate', 'latitude': lat - 0.00025, 'longitude': lon + 0.00025, 'is_main_gate': False }) # 5. Is it a Park / Stadium / Garden? elif any(k in cat for k in ('حديق', 'منتزه', 'park', 'ملعب', 'استاد', 'stadium')): gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'البوابة الرئيسية', 'gate_name_en': 'Main Entrance', 'latitude': lat + 0.00040, 'longitude': lon, 'is_main_gate': True }) gates_for_place.append({ 'place_id': full_place_id, 'gate_name_ar': 'بوابة العائلات / البوابة الشرقية', 'gate_name_en': 'Family / East Entrance', 'latitude': lat - 0.00040, 'longitude': lon + 0.00030, 'is_main_gate': False }) if gates_for_place: total_gates_to_insert.extend(gates_for_place) existing_place_ids.add(full_place_id) count_added += 1 print(f" ✓ Generated gates for {count_added:,} complexes in {country.upper()}.") print(f"\n🌟 Total Gates to Insert: {len(total_gates_to_insert):,}") if args.dry_run: print("💡 Dry run complete. No database changes made.") con.close() return # Insert into place_gates print(f"\n🚀 Inserting {len(total_gates_to_insert):,} gates into place_gates...") t_ins = time.time() batch_size = 2000 total_ins = 0 con.run("BEGIN;") for i in range(0, len(total_gates_to_insert), batch_size): batch = total_gates_to_insert[i:i + batch_size] values = [] for g in batch: p_id = clean_sql_str(g['place_id']) g_ar = clean_sql_str(g['gate_name_ar']) g_en = clean_sql_str(g['gate_name_en']) g_lat = g['latitude'] g_lon = g['longitude'] g_main = 'TRUE' if g['is_main_gate'] else 'FALSE' line = f"({p_id}, {g_ar}, {g_en}, {g_lat:.7f}, {g_lon:.7f}, {g_main})" values.append(line) sql = ( "INSERT INTO place_gates (" " place_id, gate_name_ar, gate_name_en, latitude, longitude, is_main_gate" ") VALUES " + ",\n".join(values) + ";" ) con.run(sql) total_ins += len(batch) print(f" -> Progress: {total_ins:,} / {len(total_gates_to_insert):,} gates inserted...") con.run("COMMIT;") print(f"✅ Gates insertion committed in {time.time() - t_ins:.2f}s!") con.run("ANALYZE place_gates;") total_count = con.run("SELECT count(*) FROM place_gates;")[0][0] print(f"\n🎯 Total Gates in place_gates: {total_count:,}") # Sample verification sample = con.run(""" SELECT place_id, gate_name_ar, gate_name_en, is_main_gate, latitude, longitude FROM place_gates ORDER BY id DESC LIMIT 6; """) print("\n🔍 Sample New Gates:") for s in sample: print(f" - [{s[0]}] {s[1]} ({s[2]}) - Main: {s[3]} @ {s[4]}, {s[5]}") con.close() print("\n🎉 Place Gates enrichment completed successfully!") if __name__ == '__main__': main()