from __future__ import annotations
from typing import TYPE_CHECKING
import pandas as pd
from passengersim.reporting import report_figure
from .fare_class_mix import _fig_fare_class_mix
from .generic import GenericSimulationTables, MissingDataError, SimulationTableItem
from .paths import SimTabPaths
from .tools import aggregate_by_summing_dataframe
if TYPE_CHECKING:
from typing import Literal
import altair as alt
from passengersim import Simulation
[docs]
class SimTabPathClasses(GenericSimulationTables):
"""Container for summary tables and figures extracted from a Simulation.
This class is a subclass of GenericSimulationTables, which is defined in
the generic module. It lists the items that are available in the
SimulationTables class, and provides type hints and (optionally, but
ideally) documentation for the data that is stored in each item.
"""
pathclasses: pd.DataFrame = SimulationTableItem(
aggregation_func=aggregate_by_summing_dataframe("pathclasses", ["carrier", "orig", "dest"]),
extraction_func=extract_pathclasses,
doc="Path-Class summary data.",
)
paths: pd.DataFrame = SimTabPaths.paths
[docs]
@report_figure
def fig_od_fare_class_mix(
self, orig: str, dest: str, *, raw_df=False, also_df: bool = False, label_threshold=0.06
) -> alt.Chart | pd.DataFrame | tuple[alt.Chart, pd.DataFrame]:
"""
Plot the fare class mix data for a specific origin-destination pair.
Parameters
----------
orig, dest : str
The origin and destination airport codes.
raw_df : bool, optional
If True, return the raw dataframe instead of the figure.
also_df : bool, optional
If True, return the dataframe as well as the figure.
label_threshold : float, optional
The threshold for displaying labels on the bars. Default is 0.06.
Returns
-------
alt.Chart or pd.DataFrame or tuple[alt.Chart, pd.DataFrame]
The fare class mix figure or dataframe.
"""
df = (
self.pathclasses[(self.pathclasses["orig"] == orig) & (self.pathclasses["dest"] == dest)]
.groupby(["carrier", "booking_class"])["gt_sold"]
.sum()
/ self.n_total_samples
)
df = df.rename("avg_sold")
df = df.reset_index()[["carrier", "booking_class", "avg_sold"]]
if raw_df:
return df
fig = _fig_fare_class_mix(
df,
label_threshold=label_threshold,
title=f"Fare Class Mix for {orig} to {dest}",
)
if also_df:
return fig, df
return fig
@property
def market_segmentation(self) -> pd.DataFrame:
"""Computed DataFrame with market segmentation data."""
if "market_segmentation" not in self._data:
if self.pathclasses is None:
raise ValueError("pathclasses not found, cannot compute market_segmentation")
df = self.pathclasses.groupby(["carrier", "orig", "dest"]).sum() / self.n_total_samples
df.columns = df.columns.str.removeprefix("gt_")
d = {}
for metric in ["sold", "revenue"]:
dfm = df.filter(like=f"{metric}_by_segment")
dfm.columns = dfm.columns.str.replace(f"{metric}_by_segment_", "", regex=False)
dfm.columns.name = "segment"
d[metric] = dfm.stack()
self._data["market_segmentation"] = pd.concat(d, axis=1)
return self._data["market_segmentation"]
[docs]
def market_shares(
self,
by: Literal["sold", "revenue"] = "sold",
segments: bool = False,
agg: Literal["orig", "dest"] | None = None,
) -> pd.DataFrame:
"""Compute carrier market shares, in percent, for each market.
Parameters
----------
by : {'sold', 'revenue'}, default 'sold'
The metric to compute market shares by.
segments : bool, default False
Whether to also compute shares within each demand segment. This
requires path-class data with a per-segment breakdown.
agg : {'orig', 'dest'} or None, default None
If given, pool all markets sharing an origin (or destination)
before computing shares, instead of using directional O-D markets.
Returns
-------
pd.DataFrame
Shares in percent (0 to 100), indexed by ``orig`` and ``dest``, or
by the ``agg`` dimension alone. Without ``segments``, columns are
carriers. With ``segments``, columns are a ``(segment, carrier)``
MultiIndex, including a ``"Total"`` segment equal to the result
without ``segments``. Carriers absent from a market have a share of
0, and every carrier in a market with zero total volume has NaN.
Raises
------
ValueError
If ``by`` or ``agg`` is invalid, or if ``segments`` is requested
without path-class data that has a per-segment breakdown.
MissingDataError
If neither path-class nor path data is available.
"""
if by not in ["sold", "revenue"]:
raise ValueError(f"Invalid value for 'by': {by}. Must be 'sold' or 'revenue'.")
group_on = ["orig", "dest"]
if agg is not None:
if agg not in ["orig", "dest"]:
raise ValueError(f"Invalid value for 'agg': {agg}. Must be 'orig' or 'dest'.")
group_on = [agg]
# Totals come from the gt_ columns, which exist with or without a
# segment breakdown; path data is the fallback without path classes.
try:
source = self.pathclasses
except MissingDataError:
source = None
if source is None and segments:
raise ValueError("pathclasses not found, cannot compute market shares by segment")
if source is None:
source = self.paths
result = _percent_shares(source, f"gt_{by}", group_on, ["carrier"])
if not segments:
return result
mkt_seg = self.market_segmentation
if mkt_seg.empty:
raise ValueError("pathclasses has no segment breakdown, cannot compute market shares by segment")
by_segment = _percent_shares(mkt_seg, by, group_on, ["segment", "carrier"]).sort_index(axis=1)
result.columns = pd.MultiIndex.from_product([["Total"], result.columns], names=["segment", "carrier"])
return pd.concat([by_segment, result], axis=1)
def _percent_shares(frame: pd.DataFrame, value: str, rows: list[str], columns: list[str]) -> pd.DataFrame:
"""Return each group's percent of the total ``value``, reshaped to a table.
``rows`` and ``columns`` name columns or index levels of ``frame``. Shares
are computed across the last entry of ``columns`` (e.g. carrier), within
each combination of ``rows`` and the other ``columns`` (e.g. segment).
Combinations absent from ``frame`` are 0; groups with zero total are NaN.
"""
totals = frame.groupby(rows + columns)[value].sum()
shares = totals.groupby(rows + columns[:-1]).transform(lambda x: x / x.sum() * 100)
return pd.DataFrame(shares.unstack(columns, fill_value=0))