Test data script
This commit is contained in:
James Pattinson
237
backend/populate_test_data.py
Executable file
237
backend/populate_test_data.py
Executable file
@@ -0,0 +1,237 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user