Skip to content

Flight data

This script provides dataclasses storing arrays representing temporal data of flights.

FlightData dataclass

Base dataclass for derived classes holding time series data associated to a flight.

Parameters:

NameTypeDescriptionDefault
idstr

Flight ID to which time series data arrays are associated

required
timestamps_arrayList[datetime]

Sequence of time stamps with milliseconds granularity

required
Source code in src/rmlab/data/flight.py
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
@dataclass
class FlightData:
    """Base dataclass for derived classes holding time series data associated to a flight.

    Args:
        id (str): Flight ID to which time series data arrays are associated
        timestamps_array (List[datetime]): Sequence of time stamps with milliseconds granularity
    """

    id: str
    timestamps_array: List[datetime]

    def __post_init__(self):
        if not isinstance(self.timestamps_array, list):
            raise TypeError(f"Expected `list` type in timestamps array. Got `{type(self.timestamps_array)}`")
        if len(self.timestamps_array) > 0:
            if isinstance(self.timestamps_array[0], int):
                # NOTE Assumed input timestamps are in millisec
                self.timestamps_array = [
                    datetime.fromtimestamp(ts / _MSecFactor, tz=timezone.utc) for ts in self.timestamps_array
                ]
            elif not isinstance(self.timestamps_array[0], datetime):
                raise TypeError(f"Expected `datetime` or `int` time for timestamps. Got `{type(self.timestamps_array[0])}`")
            # else timestamps are already a datetime, do nothing
        # else empty array

    def id_serial_timestamps(self):

        if not isinstance(self.timestamps_array, list):
            raise TypeError(f"Expected `list` type in timestamps array. Got `{type(self.timestamps_array)}`")
        if len(self.timestamps_array) > 0:
            if isinstance(self.timestamps_array[0], int):
                return self.id, {"Timestamp": self.timestamps_array}
            elif isinstance(self.timestamps_array[0], datetime):
                return self.id, {"Timestamp": [round(ts.timestamp() * _MSecFactor) for ts in self.timestamps_array]}
            else:
                raise TypeError(f"Expected `datetime` or `int` time for timestamps. Got `{type(self.timestamps_array[0])}`")
        # else empty array

    def is_empty(self):
        return len(self.timestamps_array) == 0

FlightDataBooks dataclass

Bases: FlightData

Data class with arrays related to books associated to a flight

Parameters:

NameTypeDescriptionDefault
fares_arrayList[str]

Stores the enabled fares at each time stamp

required
pps_arrayList[int]

Stores the price per seat associated to each booking in cents

required
seats_arrayList[int]

Stores the seats booked for each booking

required
cumulated_seats_arrayList[int]

Stores the cumulated booked seats at each time stamp

required
cumulated_revenue_arrayList[int]

Stores the cumulated revenue at each time stamp in cents

required
Source code in src/rmlab/data/flight.py
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
@dataclass
class FlightDataBooks(FlightData):
    """Data class with arrays related to books associated to a flight

    Args:
        fares_array (List[str]): Stores the enabled fares at each time stamp
        pps_array (List[int]): Stores the price per seat associated to each booking in cents
        seats_array (List[int]): Stores the seats booked for each booking
        cumulated_seats_array (List[int]): Stores the cumulated booked seats at each time stamp
        cumulated_revenue_array (List[int]): Stores the cumulated revenue at each time stamp in cents
    """

    fares_array: List[str]
    pps_array: List[int]
    seats_array: List[int]
    cumulated_seats_array: List[int]
    cumulated_revenue_array: List[int]

    def __post_init__(self):
        super().__post_init__()
        if self.cumulated_seats_array is None:
            self.cumulated_seats_array = list(accumulate(self.seats_array))
        if self.cumulated_revenue_array is None:
            self.cumulated_revenue_array = list(accumulate([pps*s for pps, s in zip(self.pps_array, self.seats_array)]))

        assert len(self.fares_array) == len(self.timestamps_array)
        assert len(self.pps_array) == len(self.timestamps_array)
        assert len(self.seats_array) == len(self.timestamps_array)
        assert len(self.cumulated_revenue_array) == len(self.timestamps_array)
        assert len(self.cumulated_seats_array) == len(self.timestamps_array)

    def id_dict(self):
        id, d = self.id_serial_timestamps()
        return id, {**d,
                    "Seats": self.seats_array,
                    "PPS": self.pps_array,
                    "Fares": self.fares_array}

FlightDataEvents dataclass

Bases: FlightData

Data class holding a sequence of events associated to a flight

Parameters:

NameTypeDescriptionDefault
events_arrayList[FlightEvent]

Stores the events at each time stamp.

required
Source code in src/rmlab/data/flight.py
155
156
157
158
159
160
161
162
163
164
165
166
167
@dataclass
class FlightDataEvents(FlightData):
    """Data class holding a sequence of events associated to a flight

    Args:
        events_array (List[FlightEvent]): Stores the events at each time stamp.
    """

    events_array: List[FlightEvent]

    def __post_init__(self):
        super().__post_init__()
        assert len(self.events_array) == len(self.timestamps_array)

FlightDataForecastedBooks dataclass

Bases: FlightData

Data class with arrays related to forecasted books associated to a flight

Parameters:

NameTypeDescriptionDefault
pps_arrayList[float]

Stores the forecasted price per seat at each time stamp

required
seats_arrayList[float]

Stores the forecasted cumulated booked seats at each time stamp

required
Source code in src/rmlab/data/flight.py
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@dataclass
class FlightDataForecastedBooks(FlightData):
    """Data class with arrays related to forecasted books associated to a flight

    Args:
        pps_array (List[float]): Stores the forecasted price per seat at each time stamp
        seats_array (List[float]): Stores the forecasted cumulated booked seats at each time stamp
    """

    pps_array: List[float]
    seats_array: List[float]

    def __post_init__(self):
        super().__post_init__()
        assert len(self.pps_array) == len(self.timestamps_array)
        assert len(self.seats_array) == len(self.timestamps_array)

FlightDataPricePerSeatSettings dataclass

Bases: FlightData

Data class with arrays of prices per seat for each fare associated to a flight

Parameters:

NameTypeDescriptionDefault
fare_to_pps_arrayMapping[str], List[int]

Maps each fare identifier to an array storing the prices per seat at each time stamp.

required
Source code in src/rmlab/data/flight.py
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@dataclass
class FlightDataPricePerSeatSettings(FlightData):
    """Data class with arrays of prices per seat for each fare associated to a flight

    Args:
        fare_to_pps_array (Mapping[str], List[int]): Maps each fare identifier
            to an array storing the prices per seat at each time stamp.
    """

    fare_to_pps_array: Mapping[str, List[int]]

    def id_dict(self):
        id, d = self.id_serial_timestamps()
        return id, {**d, **self.fare_to_pps_array}

    def __post_init__(self):
        super().__post_init__()
        assert all([len(pps) == len(self.timestamps_array)
                    for pps in self.fare_to_pps_array.values()])

FlightDataQForecast dataclass

Bases: FlightData

Data class with arrays result of q-forecast.

Parameters:

NameTypeDescriptionDefault
frat5_arrayList[float]

Array of FRAT5 values

required
book_qequivalent_arrayList[float]

Array of Q-equivalent books.

required
Source code in src/rmlab/data/flight.py
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
@dataclass
class FlightDataQForecast(FlightData):
    """Data class with arrays result of q-forecast.

    Args:
        frat5_array (List[float]): Array of FRAT5 values
        book_qequivalent_array: Array of Q-equivalent books.
    """
    frat5_array: List[float]
    book_qequivalent_array: List[float]

    def __post_init__(self):
        super().__post_init__()
        assert len(self.frat5_array) == len(self.timestamps_array)
        assert len(self.book_qequivalent_array) == len(self.timestamps_array)

FlightDataThresholdSettings dataclass

Bases: FlightData

Data class with arrays of seat thresholds for each fare associated to a flight

Parameters:

NameTypeDescriptionDefault
fare_to_threshold_arrayMapping[str], List[int]

Maps each fare identifier to an array storing the seat thresholds at each time stamp.

required
Source code in src/rmlab/data/flight.py
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@dataclass
class FlightDataThresholdSettings(FlightData):
    """Data class with arrays of seat thresholds for each fare associated to a flight

    Args:
        fare_to_threshold_array (Mapping[str], List[int]): Maps each fare identifier
            to an array storing the seat thresholds at each time stamp.
    """

    fare_to_threshold_array: Mapping[str, List[int]]

    def id_dict(self):
        id, d = self.id_serial_timestamps()
        return id, {**d, **self.fare_to_threshold_array}

    def __post_init__(self):
        super().__post_init__()
        assert all([len(threshold) == len(self.timestamps_array)
                    for threshold in self.fare_to_threshold_array.values()])