#!/usr/bin/env python3 """ Test data population script for PPR database. Generates 30 random PPR records with various aircraft, airports, and other data. """ import random import csv from datetime import datetime, timedelta from pathlib import Path # Add the app directory to the Python path import sys sys.path.append(str(Path(__file__).parent / 'app')) from sqlalchemy.orm import Session from sqlalchemy import func from app.db.session import SessionLocal from app.models.ppr import PPRRecord, PPRStatus from app.core.config import settings def load_aircraft_data(): """Load aircraft data from CSV file.""" aircraft = [] csv_path = Path(__file__).parent.parent / 'aircraft_data.csv' try: with open(csv_path, 'r', encoding='utf-8') as f: reader = csv.reader(f) for row in reader: if len(row) >= 6: aircraft.append({ 'icao24': row[0], 'registration': row[1], 'manufacturer': row[2], 'type_code': row[3], 'manufacturer_name': row[4], 'model': row[5] }) except FileNotFoundError: print("Aircraft data file not found, using fallback data") # Fallback aircraft data aircraft = [ {'registration': 'G-ABCD', 'type_code': 'C172', 'manufacturer_name': 'Cessna', 'model': '172'}, {'registration': 'G-EFGH', 'type_code': 'PA28', 'manufacturer_name': 'Piper', 'model': 'PA-28'}, {'registration': 'G-IJKL', 'type_code': 'BE36', 'manufacturer_name': 'Beechcraft', 'model': 'Bonanza'}, {'registration': 'G-MNOP', 'type_code': 'R44', 'manufacturer_name': 'Robinson', 'model': 'R44'}, {'registration': 'G-QRST', 'type_code': 'C152', 'manufacturer_name': 'Cessna', 'model': '152'}, {'registration': 'G-UVWX', 'type_code': 'PA38', 'manufacturer_name': 'Piper', 'model': 'Tomahawk'}, {'registration': 'G-YZAB', 'type_code': 'C182', 'manufacturer_name': 'Cessna', 'model': '182'}, {'registration': 'G-CDEF', 'type_code': 'DR40', 'manufacturer_name': 'Robin', 'model': 'DR400'}, {'registration': 'G-GHIJ', 'type_code': 'TB20', 'manufacturer_name': 'Socata', 'model': 'TB-20'}, {'registration': 'G-KLMN', 'type_code': 'DA40', 'manufacturer_name': 'Diamond', 'model': 'DA-40'}, ] return aircraft def load_airport_data(): """Load airport data from CSV file.""" airports = [] csv_path = Path(__file__).parent.parent / 'airports_data_clean.csv' try: with open(csv_path, 'r', encoding='utf-8') as f: reader = csv.reader(f) for row in reader: if len(row) >= 4: airports.append({ 'icao': row[0], 'iata': row[1], 'name': row[2], 'country': row[3] }) except FileNotFoundError: print("Airport data file not found, using fallback data") # Fallback airport data airports = [ {'icao': 'EGFH', 'iata': '', 'name': 'Swansea Airport', 'country': 'GB'}, {'icao': 'EGFF', 'iata': 'CWL', 'name': 'Cardiff International Airport', 'country': 'GB'}, {'icao': 'EGTE', 'iata': 'EXT', 'name': 'Exeter International Airport', 'country': 'GB'}, {'icao': 'EGGD', 'iata': 'BRS', 'name': 'Bristol Airport', 'country': 'GB'}, {'icao': 'EGHH', 'iata': 'BOH', 'name': 'Bournemouth Airport', 'country': 'GB'}, {'icao': 'EGHI', 'iata': 'SOU', 'name': 'Southampton Airport', 'country': 'GB'}, {'icao': 'EGSS', 'iata': 'STN', 'name': 'London Stansted Airport', 'country': 'GB'}, {'icao': 'EGKK', 'iata': 'LGW', 'name': 'London Gatwick Airport', 'country': 'GB'}, {'icao': 'EGLL', 'iata': 'LHR', 'name': 'London Heathrow Airport', 'country': 'GB'}, {'icao': 'EIDW', 'iata': 'DUB', 'name': 'Dublin Airport', 'country': 'IE'}, ] return airports def generate_random_ppr(aircraft_data, airport_data): """Generate a random PPR record.""" # Select random aircraft aircraft = random.choice(aircraft_data) # Select random departure airport (not Swansea) departure_airports = [a for a in airport_data if a['icao'] != 'EGFH'] arrival_airport = random.choice(departure_airports) # Sometimes add a departure airport (50% chance) departure_airport = None if random.random() < 0.5: departure_airports = [a for a in airport_data if a['icao'] != arrival_airport['icao']] departure_airport = random.choice(departure_airports) # Generate random times now = datetime.now() base_date = now + timedelta(days=random.randint(-7, 14)) # Past week to 2 weeks future # ETA: sometime between 6 AM and 8 PM eta_hour = random.randint(6, 20) eta_minute = random.choice([0, 15, 30, 45]) eta = base_date.replace(hour=eta_hour, minute=eta_minute, second=0, microsecond=0) # ETD: 1-4 hours after ETA (if departure planned) etd = None if departure_airport: etd_hours = random.randint(1, 4) etd = eta + timedelta(hours=etd_hours) # Round ETD to 15-minute intervals etd_minute = ((etd.minute // 15) * 15) % 60 etd = etd.replace(minute=etd_minute, second=0, microsecond=0) # Random captain names captains = [ 'John Smith', 'Sarah Johnson', 'Michael Brown', 'Emma Davis', 'James Wilson', 'Olivia Taylor', 'William Anderson', 'Sophia Martinez', 'Benjamin Garcia', 'Isabella Lopez', 'Alexander Gonzalez', 'Charlotte Rodriguez', 'Daniel Lee', 'Amelia Walker', 'Matthew Hall' ] # Random fuel types fuel_types = [None, '100LL', 'JET A1', 'FULL'] # Random POB pob_in = random.randint(1, 4) pob_out = random.randint(1, 4) if departure_airport else None # Random status statuses = [PPRStatus.NEW, PPRStatus.CONFIRMED, PPRStatus.LANDED, PPRStatus.DEPARTED] status = random.choice(statuses) # Random contact info (sometimes) email = None phone = None if random.random() < 0.7: email = f"{random.choice(captains).lower().replace(' ', '.')}@example.com" if random.random() < 0.5: phone = f"07{random.randint(100000000, 999999999)}" # Random notes (sometimes) notes_options = [ None, "Medical flight - priority handling required", "VIP passenger on board", "Technical stop only", "Training flight", "Photo flight - low level operations", "Maintenance ferry flight", "Charter flight", "Private flight" ] notes = random.choice(notes_options) # Create PPR record ppr = PPRRecord( status=status, ac_reg=aircraft['registration'], ac_type=aircraft['type_code'] or 'UNKNOWN', ac_call=random.choice([None, f"CALL{random.randint(100, 999)}"]), captain=random.choice(captains), fuel=random.choice(fuel_types), in_from=arrival_airport['icao'], eta=eta, pob_in=pob_in, out_to=departure_airport['icao'] if departure_airport else None, etd=etd, pob_out=pob_out, email=email, phone=phone, notes=notes, submitted_dt=now - timedelta(days=random.randint(0, 30)) # Random submission date ) return ppr def main(): """Main function to populate test data.""" print("Loading aircraft and airport data...") aircraft_data = load_aircraft_data() airport_data = load_airport_data() print(f"Loaded {len(aircraft_data)} aircraft records") print(f"Loaded {len(airport_data)} airport records") # Create database session db: Session = SessionLocal() try: print("Generating and inserting 30 test PPR records...") # Generate and insert 30 PPR records for i in range(30): ppr = generate_random_ppr(aircraft_data, airport_data) db.add(ppr) if (i + 1) % 10 == 0: print(f"Generated {i + 1} records...") # Commit all changes db.commit() print("✅ Successfully inserted 30 test PPR records!") # Print summary total_count = db.query(PPRRecord).count() print(f"Total PPR records in database: {total_count}") # Show status breakdown status_counts = db.query(PPRRecord.status, func.count(PPRRecord.id)).group_by(PPRRecord.status).all() print("\nStatus breakdown:") for status, count in status_counts: print(f" {status}: {count}") except Exception as e: print(f"❌ Error: {e}") db.rollback() finally: db.close() if __name__ == "__main__": main()