from __future__ import annotations
import sqlite3
from collections.abc import Iterable
from passengersim.utils._database_compat import deprecated_cnx_keyword
from .database import Database
[docs]
@deprecated_cnx_keyword()
def create_table_configs(db: Database | sqlite3.Connection):
sql = """
CREATE TABLE IF NOT EXISTS runtime_configs (
scenario TEXT PRIMARY KEY,
pxsim_version TEXT,
configs TEXT,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_leg_defs(db: Database | sqlite3.Connection, legs: Iterable | None = None):
sql = """
CREATE TABLE IF NOT EXISTS leg_defs
(
leg_id INTEGER PRIMARY KEY,
flt_no INTEGER,
carrier TEXT,
orig TEXT,
dest TEXT,
dep_time INTEGER,
arr_time INTEGER,
capacity INTEGER,
distance FLOAT
);
"""
db.execute(sql)
for leg in legs:
db.execute(
"""
INSERT OR REPLACE INTO leg_defs(
leg_id,
flt_no,
carrier,
orig,
dest,
dep_time,
arr_time,
capacity,
distance
) VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?
)
""",
(
leg.leg_id,
leg.flt_no,
leg.carrier_name if hasattr(leg, "carrier_name") else leg.carrier, # TODO remove this
leg.orig,
leg.dest,
leg.dep_time,
leg.arr_time,
leg.capacity,
leg.distance,
),
)
[docs]
@deprecated_cnx_keyword()
def create_table_fare_restriction_defs(db: Database | sqlite3.Connection, restrictions: list[str]) -> None:
"""
Create and populate a static database table of fare restrictions.
The contents of this table is static input data, used but not mutated
by the simulator.
Parameters
----------
db : Database or sqlite3.Connection
restrictions : list[str]
The restrictions to store in the database.
"""
sql = """
CREATE TABLE IF NOT EXISTS fare_restriction_defs
(
restriction_id INTEGER PRIMARY KEY,
restriction TEXT NOT NULL
);
"""
db.execute(sql)
for idx, restriction in enumerate(restrictions, start=1):
db.execute(
"""
INSERT OR REPLACE INTO fare_restriction_defs(
restriction_id,
restriction
) VALUES (
?, ?
)
""",
(idx, restriction),
)
[docs]
@deprecated_cnx_keyword()
def create_table_fare_defs(db: Database | sqlite3.Connection, fares: Iterable | None = None) -> None:
"""
Create and populate a static database table of fares.
The contents of this table is static input data, used but not mutated
by the simulator.
Parameters
----------
db : Database or sqlite3.Connection
fares : Iterable[Fare]
The fares to store in the database.
"""
sql = """
CREATE TABLE IF NOT EXISTS fare_defs
(
fare_id INTEGER PRIMARY KEY,
carrier VARCHAR(10) NOT NULL,
orig VARCHAR(10) NOT NULL,
dest VARCHAR(10) NOT NULL,
booking_class VARCHAR(10) NOT NULL,
price FLOAT NOT NULL,
restrictions VARCHAR(20) NOT NULL,
brand VARCHAR(20),
category VARCHAR(20)
);
"""
db.execute(sql)
sql2 = """
CREATE INDEX IF NOT EXISTS fare_defs_idx_2
ON fare_defs (
carrier, booking_class
);
"""
db.execute(sql2)
for fare in fares:
db.execute(
"""
INSERT OR REPLACE INTO fare_defs(
fare_id,
carrier,
orig,
dest,
booking_class,
brand,
price,
restrictions,
category
) VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?
)
""",
(
fare.fare_id,
fare.carrier_name,
fare.orig,
fare.dest,
fare.booking_class,
fare.brand,
fare.price,
",".join([str(r) for r in fare.get_restrictions()]),
fare.category,
),
)
[docs]
@deprecated_cnx_keyword()
def create_table_path_defs(db: Database | sqlite3.Connection, paths: Iterable | None = None):
sql = """
CREATE TABLE IF NOT EXISTS path_defs
(
path_id INTEGER PRIMARY KEY,
carrier TEXT,
orig TEXT,
stop1 TEXT,
dest TEXT,
leg1 INTEGER,
leg2 INTEGER,
distance FLOAT
);
"""
db.execute(sql)
for pth in paths:
connects = pth.num_legs() > 1
db.execute(
"""
INSERT OR REPLACE INTO path_defs(
path_id,
carrier,
orig,
stop1,
dest,
leg1,
leg2,
distance
) VALUES (
?, ?, ?, ?, ?, ?, ?, ?
)
""",
(
pth.path_id,
pth.get_leg_carrier(0),
pth.orig,
pth.get_leg_dest(0) if connects else None,
pth.dest,
pth.get_leg_id(0),
pth.get_leg_id(1) if connects else None,
pth.get_total_distance(),
),
)
[docs]
@deprecated_cnx_keyword()
def create_table_leg_detail(db: Database | sqlite3.Connection, primary_key: bool = False) -> None:
"""
Create the `leg_detail` table in the database.
Parameters
----------
db : Database or sqlite3.Connection
primary_key : bool, default False
"""
sql = """
CREATE TABLE IF NOT EXISTS leg_detail
(
scenario VARCHAR(20) NOT NULL,
iteration INT NOT NULL,
trial INT NOT NULL,
sample INT NOT NULL,
days_prior INT NOT NULL,
leg_id INT NOT NULL,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
sold INT,
revenue FLOAT,
local_sold INT,
local_revenue FLOAT,
q_demand FLOAT,
detruncated_demand FLOAT,
forecast_mean FLOAT,
bid_price FLOAT,
displacement FLOAT
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(
primary_key=", PRIMARY KEY(scenario, iteration, trial, sample, "
"carrier, orig, dest, flt_no, days_prior, dep_date)"
)
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_leg_bucket_detail(db: Database | sqlite3.Connection, primary_key: bool = False):
sql = """
CREATE TABLE IF NOT EXISTS leg_bucket_detail
(
scenario VARCHAR(20) NOT NULL,
iteration INT NOT NULL,
trial INT NOT NULL,
sample INT NOT NULL,
days_prior INT NOT NULL,
leg_id INT NOT NULL,
cabin_code VARCHAR(10) NOT NULL DEFAULT "",
bucket_number INT NOT NULL,
name VARCHAR(10) NOT NULL,
auth INT,
revenue FLOAT,
sold INT,
detruncated_demand FLOAT,
forecast_mean FLOAT,
forecast_stdev FLOAT,
forecast_closed_in_tf FLOAT,
forecast_closed_in_future FLOAT,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(
primary_key=", PRIMARY KEY(scenario, iteration, trial, sample, days_prior, "
"carrier, orig, dest, flt_no, dep_date, bucket_number)"
)
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_demand_detail(db: Database | sqlite3.Connection, primary_key: bool = False):
sql = """
CREATE TABLE IF NOT EXISTS demand_detail
(
scenario VARCHAR(20) NOT NULL,
iteration INT NOT NULL,
trial INT NOT NULL,
sample INT NOT NULL,
days_prior INT NOT NULL,
segment VARCHAR(10) NOT NULL,
orig VARCHAR(10) NOT NULL,
dest VARCHAR(10) NOT NULL,
updated_at DATETIME NOT NULL DEFAuLT CURRENT_TIMESTAMP,
sample_demand FLOAT,
sold INT,
no_go INT,
revenue FLOAT
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(
primary_key=", PRIMARY KEY(scenario, iteration, trial, sample, days_prior, segment, orig, dest)"
)
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_fare_detail(db: Database | sqlite3.Connection, primary_key: bool = False):
sql = """
CREATE TABLE IF NOT EXISTS fare_detail
(
scenario VARCHAR(20) NOT NULL,
iteration INT NOT NULL,
trial INT NOT NULL,
sample INT NOT NULL,
days_prior INT NOT NULL,
fare_id INT NOT NULL,
sold INT,
sold_business INT,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(
primary_key=", PRIMARY KEY(scenario, iteration, trial, sample, "
"days_prior, carrier, orig, dest, booking_class)"
)
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_bookings_by_timeframe(db: Database | sqlite3.Connection, primary_key: bool = False):
sql = """
CREATE TABLE IF NOT EXISTS bookings_by_timeframe
(
scenario VARCHAR(20) NOT NULL,
trial INT NOT NULL,
carrier VARCHAR(10) NOT NULL,
booking_class VARCHAR(10) NOT NULL,
days_prior INT NOT NULL,
tot_sold FLOAT,
avg_sold FLOAT,
avg_business FLOAT,
avg_leisure FLOAT,
avg_revenue FLOAT,
avg_price FLOAT,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(primary_key=", PRIMARY KEY(scenario, trial, carrier, booking_class, days_prior)")
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_booking_curve(db: Database | sqlite3.Connection, primary_key: bool = True):
sql = """
CREATE TABLE IF NOT EXISTS booking_curve (
scenario VARCHAR(20) NOT NULL,
carrier VARCHAR(10) NOT NULL,
orig VARCHAR(10) NOT NULL,
dest VARCHAR(10) NOT NULL,
flt_no INT NOT NULL,
days_prior INT NOT NULL,
ratio FLOAT NOT NULL
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(primary_key=", PRIMARY KEY(scenario, carrier, orig, dest, flt_no, days_prior)")
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_path_class_detail(db: Database | sqlite3.Connection, primary_key: bool = False):
sql = """
CREATE TABLE IF NOT EXISTS path_class_detail
(
scenario VARCHAR(20) NOT NULL,
iteration INT NOT NULL,
trial INT NOT NULL,
sample INT NOT NULL,
days_prior INT NOT NULL,
path_id INT NOT NULL,
booking_class VARCHAR(10) NOT NULL,
sold INT,
sold_priceable INT,
revenue FLOAT,
forecast_mean FLOAT,
forecast_stdev FLOAT,
forecast_closed_in_tf FLOAT,
forecast_closed_in_future FLOAT,
adjusted_price FLOAT,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(
primary_key=", PRIMARY KEY(scenario, iteration, trial, sample, days_prior, path_id, booking_class)"
)
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_edgar(db: Database | sqlite3.Connection, primary_key: bool = False):
"""Forecast accuracy data, modeled after UA's EDGAR approach"""
sql = """
CREATE TABLE IF NOT EXISTS edgar
(
scenario VARCHAR(20) NOT NULL,
iteration INT NOT NULL,
trial INT NOT NULL,
sample INT NOT NULL,
timeframe INT NOT NULL,
path_id INT NOT NULL,
booking_class VARCHAR(10) NOT NULL,
sold INT,
sold_priceable INT,
forecast_mean FLOAT,
forecast_stdev FLOAT,
closed FLOAT,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(
primary_key=", PRIMARY KEY(scenario, iteration, trial, sample, timeframe, path_id, booking_class)"
)
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_table_distance(db: Database | sqlite3.Connection, primary_key: bool = True):
sql = """
CREATE TABLE IF NOT EXISTS distance (
orig VARCHAR(10) NOT NULL,
dest VARCHAR(10) NOT NULL,
miles FLOAT
{primary_key}
);
"""
if primary_key is True:
sql = sql.format(primary_key=", PRIMARY KEY(orig, dest)")
else:
sql = sql.format(primary_key="")
db.execute(sql)
[docs]
@deprecated_cnx_keyword()
def create_tables(db: Database, primary_keys: dict[str, bool] | None = None):
pk = dict(
leg=False,
leg_bucket=False,
demand=False,
edgar=False,
fare=False,
booking_curve=True,
distance=True,
bookings=False,
path_class=False,
)
if primary_keys is not None:
pk.update(primary_keys)
create_table_configs(db)
create_table_leg_detail(db, pk["leg"])
create_table_leg_bucket_detail(db, pk["leg_bucket"])
create_table_demand_detail(db, pk["demand"])
create_table_fare_detail(db, pk["fare"])
create_table_bookings_by_timeframe(db, pk["bookings"])
create_table_booking_curve(db, pk["booking_curve"])
create_table_path_class_detail(db, pk["path_class"])
create_table_edgar(db, pk["edgar"])
create_table_distance(db, pk["distance"])
db._commit_raw()