Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Table of Contents

altrios

to_pydict

def to_pydict(self, data_fmt: str = "msg_pack", flatten: bool = False) -> Dict

Returns self converted to pure python dictionary with no nested Rust objects Arguments

  • - flatten: if True, returns dict without any hierarchy
  • - data_fmt: data format for intermediate conversion step

get_hist_len

def get_hist_len(obj: Dict) -> Optional[int]

Finds nested history and gets lenth of first element

get_flattened

def get_flattened(obj: Dict | List,
                  hist_len: int,
                  prepend_str: str = "") -> Dict

Flattens and returns dictionary, separating keys and indices with a "." Arguments

- obj: object to flatten

- hist_len: length of any lists storing history data

- prepend_str: prepend this to all keys in the returned flat dict

from_pydict

@classmethod
def from_pydict(cls,
                pydict: Dict,
                data_fmt: str = "msg_pack",
                skip_init: bool = False) -> Self

Instantiates Self from pure python dictionary Arguments

  • - pydict: dictionary to be converted to ALTRIOS object
  • - data_fmt: data format for intermediate conversion step
  • - skip_init: passed to SerdeAPI methods to control whether initialization is skipped

to_dataframe

def to_dataframe(
        self,
        pandas: bool = False,
        allow_partial: bool = False) -> Union[pd.DataFrame, pl.DataFrame]

Returns time series results from ALTRIOS object as a Polars or Pandas dataframe.

Arguments

  • - pandas: returns pandas dataframe if True; otherwise, returns polars dataframe by default
  • - allow_partial: tries to return dataframe of length equal to solved time steps if simulation fails early

altrios.rollout

altrios.defaults

Module for default modeling assumption constants.

LHV_DIESEL_KJ_PER_KG

https://www.engineeringtoolbox.com/fuels-higher-calorific-values-d_169.html

RHO_DIESEL_KG_PER_M3

https://www.engineeringtoolbox.com/fuels-densities-specific-volumes-d_166.html

DIESEL_REFUEL_RATE_J_PER_HR

300 gallons per minute -> joules per hour

BEL_CHARGER_COST_USD

NLR Cost of Charging (Borlaug) showing ~linear trend on kW; ICCT report showing little change through 2030

DIESEL_LOCO_COST_USD

Zenith et al.

altrios.plot

altrios.train_planner.data_prep

load_freight_demand

def load_freight_demand(
    demand_table: Union[pl.DataFrame, pl.LazyFrame, Path,
                        str], config: planner_config.TrainPlannerConfig
) -> Tuple[pl.DataFrame, pl.Series, int]

Load the user input csv file into a dataframe for later processing

Arguments:

  • user_input_file - path to the input csv file that user import to the module Example Input: Origin Destination Train_Type Number_of_Cars Number_of_Containers Barstow Stockton Unit 2394 0 Barstow Stockton Manifest 2588 0 Barstow Stockton Intermodal 2221 2221

    Outputs:

  • df_annual_demand - dataframe with all pair information including: origin, destination, train type, number of cars

  • node_list - List of origin or destination demand nodes

build_locopool

def build_locopool(config: planner_config.TrainPlannerConfig,
                   demand_file: Union[pl.DataFrame, pl.LazyFrame, Path, str],
                   dispatch_schedule: Union[pl.DataFrame, pl.LazyFrame]
                   | None = None,
                   locomotives_per_node: int | None = None) -> pl.DataFrame

Generate default locomotive pool

Arguments:

  • demand_file - Path to a file with origin-destination demand

  • shares - List of shares for each locomotive type in loco_info (implemented for two-way shares only) Outputs:

  • loco_pool - Locomotive pool containing all locomotives' information that are within the system

build_refuelers

def build_refuelers(node_list: pd.Series, loco_pool: pl.DataFrame,
                    refueler_info: pd.DataFrame,
                    refuelers_per_incoming_corridor: int) -> pl.DataFrame

Build the default set of refueling facilities.

Arguments:

  • node_list - List of origin or destination demand nodes
  • loco_pool - Locomotive pool
  • refueler_info - DataFrame with information for each type of refueling infrastructure to use

  • refuelers_per_incoming_corridor - Queue size per corridor arriving at each node. Outputs:

  • refuelers - Polars dataframe of facility county by node and type of fuel

altrios.train_planner.planner

dispatch

def dispatch(dispatch_time: int, origin: str, loco_pool: pl.DataFrame,
             train_tonnage: float, hp_required: float, total_cars: float,
             config: planner_config.TrainPlannerConfig) -> pl.Series

Identify and select locomotives to dispatch for a train, based on origin, requirements, and availability.

This function selects the optimal set of locomotives from the available pool at a specified origin based on horsepower requirements, tonnage needs, and configuration parameters. It implements locomotive selection logic including potential diesel requirements and ensures sufficient power for the given train.

Parameters

dispatch_time : int Time (in hours) when the train is scheduled to depart origin : str Origin node name where the train will depart from loco_pool : pl.DataFrame DataFrame containing all locomotives in the network with their statuses and properties train_tonnage : float Total tonnage of the train to be dispatched hp_required : float Horsepower required for this train type on this origin-destination corridor total_cars : float Total number of cars (loaded, empty, or otherwise) included on the train config : planner_config.TrainPlannerConfig Configuration object with dispatch settings and rules

Returns

pl.Series Boolean series with same length as loco_pool, with True values indicating selected locomotives

Raises

ValueError If no locomotives are available at the origin or if requirements cannot be met

update_refuel_queue

def update_refuel_queue(
        loco_pool: pl.DataFrame, refuelers: pl.DataFrame, current_time: float,
        event_tracker: pl.DataFrame) -> Tuple[pl.DataFrame, pl.DataFrame]

Update locomotive refueling status, manage service queues, and track events.

This function processes arrived locomotives, updates refueling and servicing status, and manages the refueling queue across all locations. It tracks when locomotives finish refueling or servicing, updates their status appropriately, and records these events.

Parameters

loco_pool : pl.DataFrame DataFrame containing all locomotives in the network with their current status refuelers : pl.DataFrame DataFrame containing all refueling ports in the network with capacity information current_time : float Current simulation time in hours event_tracker : pl.DataFrame DataFrame tracking locomotive events (arrivals, refueling, etc.)

Returns

Tuple[pl.DataFrame, pl.DataFrame] loco_pool : Updated locomotive pool DataFrame with new statuses event_tracker : Updated event tracker with new refueling/servicing events

run_train_planner

def run_train_planner(
    rail_vehicles: List[alt.RailVehicle],
    location_map: Dict[str, List[alt.Location]],
    network: List[alt.Link],
    loco_pool: Optional[pl.DataFrame],
    refuelers: Optional[pl.DataFrame],
    scenario_year: int,
    train_type: alt.TrainType = alt.TrainType.Freight,
    config: planner_config.TrainPlannerConfig = planner_config.
    TrainPlannerConfig(),
    demand_file: Union[pl.DataFrame, Path, str] = defaults.DEMAND_FILE,
    network_charging_guidelines: Optional[pl.DataFrame] = None
) -> Tuple[
        pl.DataFrame,
        pl.DataFrame,
        pl.DataFrame,
        List[alt.SpeedLimitTrainSim],
        List[alt.EstTimeNet],
]

Run the train planner to generate consist plans, refueling schedules, and simulations.

This function is the main entry point for train planning. It processes demand data, schedules trains, assigns locomotives, plans refueling, and generates train simulations based on the provided configuration. It handles both single-train mode and multi-train scheduling across a network.

Parameters

rail_vehicles : List[alt.RailVehicle] List of available rail vehicle types with their properties location_map : Dict[str, List[alt.Location]] Dictionary mapping location IDs to lists of Location objects network : List[alt.Link] List of links defining the rail network loco_pool : Optional[pl.DataFrame] DataFrame containing available locomotives with their properties. If None, will be generated based on demand. refuelers : Optional[pl.DataFrame] DataFrame containing refueling facilities with their capacities. If None, will be generated based on configuration. scenario_year : int The year for which to run the simulation (affects energy prices, etc.) train_type : alt.TrainType Type of train to simulate (default: Freight) config : planner_config.TrainPlannerConfig Configuration object with planning parameters demand_file : Union[pl.DataFrame, Path, str] Source of demand data, either as DataFrame or file path network_charging_guidelines : Optional[pl.DataFrame] Guidelines for charging infrastructure by location

Returns

Tuple[pl.DataFrame, pl.DataFrame, pl.DataFrame, List[alt.SpeedLimitTrainSim], List[alt.EstTimeNet]] train_consist_plan : DataFrame with planned train consists and schedules loco_pool : Updated DataFrame of all locomotives and their states refuelers : DataFrame of all refueling facilities speed_limit_train_sims : List of SpeedLimitTrainSim objects for each planned train est_time_nets : List of EstTimeNet objects with estimated timings

altrios.train_planner.planner_config

TrainPlannerConfig Objects

@dataclass
class TrainPlannerConfig()

Dataclass class for train planner configuration parameters.

Attributes:

  • single_train_mode: True to only run one round-trip train and schedule its charging; False to plan train consists
  • min_cars_per_train: Dict of the minimum length in number of cars to form a train for each train type
  • target_cars_per_train: Dict of the standard train length in number of cars for each train type
  • manifest_empty_return_ratio: Desired railcar reuse ratio to calculate the empty manifest car demand, (E_ij+E_ji)/(L_ij+L_ji)
  • cars_per_locomotive: Heuristic scaling factor used to size number of locomotives needed based on demand.
  • cars_per_locomotive_fixed: If True, cars_per_locomotive overrides hp_per_ton calculations used for dispatching decisions.
  • refuelers_per_incoming_corridor: Heuristic scaling factor used to scale number of refuelers needed at each node based on number of incoming corridors.
  • containers_per_car: Containers stacked on each car (applicable only for intermodal containers)
  • require_diesel: True to require each consist to have at least one diesel locomotive.
  • manifest_empty_return_ratio: Dict
  • drag_coeff_function: Dict
  • hp_required_per_ton: Dict
  • dispatch_scaling_dict: Dict
  • loco_info: Dict
  • refueler_info: Dict
  • return_demand_generators: Dict

return_demand_generators

default defined in train_demand_generators.py

altrios.train_planner.schedulers

calculate_waiting_time_single_dispatch

def calculate_waiting_time_single_dispatch(
        cumulative_demand_control: int, last_dispatch: int,
        demand_hourly: pl.DataFrame, dispatch_hour: int,
        remaining_demand_list: pl.DataFrame,
        remaining_demand_list_control: pl.DataFrame, search_range: int,
        od_pair_loop: str, min_num_cars_per_train: int,
        target_num_cars_per_train: int,
        config: planner_config.TrainPlannerConfig) -> tuple

Calculate the waiting time for a single dispatch using Polars DataFrames.

find_minimum_waiting_time

def find_minimum_waiting_time(
        num_iterations: int, demand_hourly: pl.DataFrame,
        border_time_list: list, min_num_cars_per_train: int,
        target_num_cars_per_train: int,
        config: planner_config.TrainPlannerConfig) -> pl.DataFrame

Find the minimum waiting time for dispatches using Polars DataFrame.

dispatch_hourly_demand_optimized_departure

def dispatch_hourly_demand_optimized_departure(
        demand_hourly: pl.DataFrame, rail_vehicles: List[alt.RailVehicle],
        freight_type_to_car_type: Dict[str, str],
        config: planner_config.TrainPlannerConfig) -> pl.DataFrame

Converts a table of demand into a dispatch plan where trains depart from each origin in uniformly spaced intervals.

Arguments:

  • demand - DataFrame or LazyFrame representing origin-destination demands (number of trains).
  • rail_vehicles - List of altrios.RailVehicle objects.

  • config - TrainPlannerConfig object. Outputs:

    Updated demand DataFrame or LazyFrame representing dispatches, each defined with an origin, destination, train type, number of (loaded and empty) cars, tonnage, and HP per ton requirement.

dispatch_uniform_demand_uniform_departure

def dispatch_uniform_demand_uniform_departure(
        demand: pl.DataFrame, rail_vehicles: List[alt.RailVehicle],
        freight_type_to_car_type: Dict[str, str],
        config: planner_config.TrainPlannerConfig) -> pl.DataFrame

Generate a tabulated demand pair to indicate the expected dispatching interval and actual dispatching timesteps after rounding, with departures from each terminal spaced as evenly as possible

Arguments:

  • demand - DataFrame or LazyFrame representing origin-destination demands (number of trains).
  • rail_vehicles - List of altrios.RailVehicle objects.

  • config - TrainPlannerConfig object. Outputs:

  • schedule - Tabulated dispatching time for each demand pair for each train type in hours

altrios.train_planner.train_demand_generators

initialize_reverse_empties

def initialize_reverse_empties(
    demand: Union[pl.LazyFrame, pl.DataFrame]
) -> Union[pl.LazyFrame, pl.DataFrame]

Swap Origin and Destination and append _Empty to Train_Type.

Arguments:

  • demand - DataFrame or LazyFrame representing origin-destination demand.

    Outputs:

    Updated demand DataFrame or LazyFrame.

generate_return_demand_unit

def generate_return_demand_unit(
    demand_subset: Union[pl.LazyFrame, pl.DataFrame],
    config: planner_config.TrainPlannerConfig
) -> Union[pl.LazyFrame, pl.DataFrame]

Given a set of Unit train demand for one or more origin-destination pairs, generate demand in the reverse direction(s).

Arguments:

  • demand - DataFrame or LazyFrame representing origin-destination demand for Unit trains.

    Outputs:

    Updated demand DataFrame or LazyFrame representing demand in the reverse direction(s) for each origin-destination pair.

generate_return_demand_manifest

def generate_return_demand_manifest(
    demand_subset: Union[pl.LazyFrame, pl.DataFrame],
    config: planner_config.TrainPlannerConfig
) -> Union[pl.LazyFrame, pl.DataFrame]

Given a set of Manifest train demand for one or more origin-destination pairs, generate demand in the reverse direction(s).

Arguments:

  • demand - DataFrame or LazyFrame representing origin-destination demand for Unit trains.

    Outputs:

    Updated demand DataFrame or LazyFrame representing demand in the reverse direction(s) for each origin-destination pair.

generate_return_demand_intermodal

def generate_return_demand_intermodal(
    demand_subset: Union[pl.LazyFrame, pl.DataFrame],
    config: planner_config.TrainPlannerConfig
) -> Union[pl.LazyFrame, pl.DataFrame]

Given a set of Intermodal train demand for one or more origin-destination pairs, generate demand in the reverse direction(s).

Arguments:

  • demand - DataFrame or LazyFrame representing origin-destination demand for Unit trains.

    Outputs:

    Updated demand DataFrame or LazyFrame representing demand in the reverse direction(s) for each origin-destination pair.

generate_return_demand

def generate_return_demand(
        demand: pl.DataFrame,
        config: planner_config.TrainPlannerConfig) -> pl.DataFrame

Create a dataframe for additional demand needed for empty cars of the return trains

Arguments:

  • df_annual_demand - The user_input file loaded by previous functions that contains loaded demand for each demand pair.

  • config - Object storing train planner configuration paramaters Outputs:

  • df_return_demand - The demand generated by the need of returning the empty cars to their original nodes

generate_manifest_rebalancing_demand

def generate_manifest_rebalancing_demand(
        demand: pl.DataFrame, node_list: List[str],
        config: planner_config.TrainPlannerConfig) -> pl.DataFrame

Create a dataframe for summarized view of all origins' manifest demand in number of cars and received cars, both with loaded and empty counts

Arguments:

  • demand - The user_input file loaded by previous functions that contains laoded demand for each demand pair.

  • node_list - A list containing all the names of nodes in the system

  • config - Object storing train planner configuration paramaters

    Outputs:

  • origin_manifest_demand - The dataframe that summarized all the manifest demand originated from each node by number of loaded and empty cars with additional columns for checking the unbalance quantity and serve as check columns for the manifest empty car rebalancing function

generate_demand_trains

def generate_demand_trains(
        demand: pl.DataFrame, demand_returns: pl.DataFrame,
        demand_rebalancing: pl.DataFrame, rail_vehicles: List[alt.RailVehicle],
        freight_type_to_car_type: Dict[str, str],
        config: planner_config.TrainPlannerConfig) -> pl.DataFrame

Generate a tabulated demand pair to indicate the final demand for each demand pair for each train type in number of trains

Arguments:

  • demand - Tabulated demand for each demand pair for each train type in number of cars

  • demand - The user_input file loaded and prepared by previous functions that contains loaded car demand for each demand pair.

  • demand_returns - The demand generated by the need of returning the empty cars to their original nodes

  • demand_rebalancing - Documented additional manifest demand pairs and corresponding quantity for rebalancing process

  • config - Object storing train planner configuration paramaters Outputs:

  • demand - Tabulated demand for each demand pair in terms of number of cars and number of trains

altrios.optimization

altrios.optimization.multi_obj_opt

Module for multi-objective optimization. This will likely use PyMOO extensively.

altrios.optimization.cal_and_val

Module for running train, locomotive, and/or consist models to calibrate and validate against test data.

get_delta_seconds

def get_delta_seconds(ds: pd.Series) -> pd.Series

Arugments:

  • ds: pandas.Series; data of the current segment previously passed to to_datetime_from_format returns:
  • out: pandas.Series; a pandas.Series data that shows the datetime deltas between rows of the segment. Row i has time elasped between i and row i-1. Row 0 has value 0.

Returns pd.Series of time delta [s]

get_error

def get_error(t: np.array, mod: np.array, exp: np.array)

Return error for model data, mod, w.r.t. experimental data, exp, over time, t

ModelError Objects

@dataclass
class ModelError(object)

Dataclass class for calculating model error of various ALTRIOS objects w.r.t. test data.

Attributes:

  • ser_model_dict: dict variable in which:

  • key: a str representing trip keyword string

  • value: a str converted from Rust locomotive models' serialization method

  • model_type: str that can only be 'ConsistSimulation', 'SetSpeedTrainSim' or 'LocomotiveSimulation'; indicates which model to instantiate during optimization process

  • dfs: a dict variable in which:

  • key: str representing trip keyword; will be the same keyword as in models

  • value: pandas.DataFrame variable with trip detailed data to be compared against. each df should have a time [s] column

  • objectives: a list of 2-element tuples. For each tuple, element 0 is the name of the reference test data signal in dfs; element 1 is a tuple of strings representing a hierarchical path to the corresponding model signal. This field is used for error calculation.

  • params: a tuple whose individual element is a str containing hierarchical paths to parameters to manipulate starting from one of the 3 possible Rust model structs

  • verbose: bool: if True, the verbose of error calculation will be printed

  • debug: bool: if True, prints more stuff

  • allow_partial: whether to allow partial runs, if True, errors out whenever a run can't be completed

get_errors

def get_errors(
    mod_dict,
    return_mods: Optional[bool] = False,
    pyplot: bool = False,
    plotly: bool = False,
    show_pyplot: bool = False,
    plot_save_dir: Optional[Path] = None,
    plot_perc_err: bool = False,
    font_size: float = 16,
    perc_err_target_for_plot: float = 1.5
) -> Tuple[
        Dict[str, Dict[str, float]],  # error dict
        # if return_mods is True, solved models
        Optional[Tuple[Dict[str, Dict[str, float]], Dict[
            str,
            Union[SetSpeedTrainSim, ConsistSimulation, LocomotiveSimulation],
        ]]]]

Calculate model errors w.r.t. test data for each element in dfs/models for each objective. Arugments:

  • mod_dict: the dict whose values are generated Rust ALTRIOS models
  • return_mods: bool; if true, also returns dict of solved models
  • pyplot: if true, plots objectives with matplotlib.pyplot
  • plotly: if true, plots with plotly. plot_save_dir must be provided.
  • show_pyplot: if true, shows pyplot plots
  • plot_save_dir: Path for saving plots.
  • plot_perc_err: Whether to include axes for plotting % error

Returns:

  • errors: dict whose values are dicts containing the errors wrt each objective
  • solved_mods Optional; dict whose values are the Rust locomotive models; only returned when return_mods is True

update_params

def update_params(
    xs: List[Any]
) -> Dict[str, Union[LocomotiveSimulation, SetSpeedTrainSim,
                     ConsistSimulation]]

Updates model parameters based on xs, which must match length of self.params

setup_plots

def setup_plots(
    key: str,
    mod: Any,
    time_seconds: List[float],
    bc: List[float],
    plots_per_key: int,
    pyplot: bool = False,
    plotly: bool = False,
    plot_save_dir: Optional[str] = None
) -> Tuple[Optional[Figure], Optional[plt.Axes], Optional[go.Figure]]

Arguments:

  • plot: ...
  • plotly: make and save plotly plots

CalibrationProblem Objects

class CalibrationProblem(ElementwiseProblem)

Problem for calibrating models to match test data

run_minimize

def run_minimize(problem: CalibrationProblem,
                 algorithm: GeneticAlgorithm,
                 termination: DMOT,
                 save_history: bool = False,
                 copy_algorithm: bool = False,
                 copy_termination: bool = False,
                 save_path: Optional[str] = "pymoo_res",
                 pickle_res_to_file: bool = False)

Arguments:

  • save_path: filename for results -- will save res_df separately by appending

min_error_selection

def min_error_selection(result_df: pd.DataFrame,
                        param_num: int,
                        norm_num: int = 2) -> np.ndarray

Arguments:

  • result_df - pd.DataFrame containing pymoo res.X and res.F concatenated
  • param_num - number of parameters
  • norm_num - norm number -- e.g. 2 would result in RMS error

altrios.fuel_grid

altrios.tests.test_locomotive

altrios.tests

altrios.tests.test_powertrain_generator

altrios.tests.test_fuel_grid

altrios.tests.test_metric_calculator

altrios.tests.test_consist_sim

altrios.tests.test_powertrain_edrive

altrios.tests.test_serde

altrios.tests.test_powertrain_res

altrios.tests.test_utilities

altrios.tests.test_locomotive_simulation

altrios.tests.test_multi_obj_opt

altrios.tests.test_powertrain_fuel_conv

altrios.tests.test_train_planner

TestTrainPlanner Objects

class TestTrainPlanner(unittest.TestCase)

populate_me

def populate_me()

to be populated with an actual test and renamed accordingly

altrios.tests.test_multi_obj_cal_and_val

altrios.tests.mock_resources

altrios.tests.test_objectives

altrios.tests.test_consist

altrios.resources

altrios.resources.powertrains

altrios.resources.powertrains.fuel_converters

altrios.resources.powertrains.reversible_energy_storages

altrios.resources.rolling_stock

altrios.resources.networks

altrios.resources.trains

altrios.demos.speed_limit_train_sim_demo_with_derating

altrios.demos

Module containing demo files. Be sure to check out https://natlabrockies.github.io/altrios/how-to-run-altrios.

altrios.demos.sim_manager_demo

altrios.demos.set_speed_simple_corr_demo

SetSpeedTrainSim over a simple, hypothetical corridor

altrios.demos.rollout_demo

altrios.demos.speed_limit_train_sim_demo

altrios.demos.test_demos

altrios.demos.plot_util

plot_locos_from_ts

def plot_locos_from_ts(ts: alt.SetSpeedTrainSim,
                       x: str,
                       show_plots: bool = False)

Can take in either SetSpeedTrainSim or SpeedLimitTrainSim Extracts first instance of each loco_type and plots representative plots Offers two plotting options to put on x axis ts: train sim x: ["time","offset"]

altrios.demos.hel_demo

altrios.demos.set_speed_train_sim_demo

altrios.demos.bel_demo

altrios.demos.conv_demo

altrios.demos.speed_limit_simple_corr_demo

SetSpeedTrainSim over a simple, hypothetical corridor

altrios.metric_calculator

ScenarioInfo Objects

@dataclass
class ScenarioInfo()

Dataclass class maintaining records of scenario parameters that influence metric calculations.

Fields:

  • sims: SpeedLimitTrainSim (single-train sim) or SpeedLimitTrainSimVec (multi-train sim) including simulation results

  • simulation_days: Number of days included in these results (after any warm-start or cool-down days were excluded)

  • annualize: Whether to scale up output metrics to a full year's equivalent.

  • scenario_year: Year that is being considered in this scenario.

  • loco_pool: polars.DataFrame defining the pool of locomotives that were available to potentially be dispatched, each having a Locomotive_ID,Locomotive_Type,Cost_USD,Lifespan_Years. Not required for single-train sim.

  • consist_plan: polars.DataFrame defining dispatched train consists, where each row includes a Locomotive_ID and a Train_ID. Not required for single-train sim.

  • refuel_facilities: polars.DataFrame defining refueling facilities, each with a Refueler_Type, Port_Count, and Cost_USD, and Lifespan_Years. Not required for single-train sim.

  • refuel_sessions: polars.DataFrame defining refueling sessions, each with a Locomotive_ID, Locomotive_Type, Fuel_Type, Node, and Refuel_Energy_J. Not required for single-train sim.

  • emissions_factors: polars.DataFrame with unit CO2eq_kg_per_MWh defined for each Node. Not required for single-train sim.

  • nodal_energy_prices: polars.DataFrame with unit Price defined for each Node and Fuel. Not required for single-train sim.

  • count_unused_locomotives: If True, fleet composition is defined using all locomotives in loco_pool; if False, fleet composition is defined using only the locomotives dispatched. Not required for single-train sim.

main

def main(scenario_infos: Union[ScenarioInfo, List[ScenarioInfo]],
         annual_metrics: Union[Tuple[str, str], List[Tuple[str, str]]] = [
             ('Freight_Moved', 'million tonne-mi'),
             ('Freight_Moved', 'million tonne-km'),
             ('Freight_Moved', 'car-miles'), ('Freight_Moved', 'cars'),
             ('Freight_Moved', 'detailed car counts'), ('GHG', 'tonne CO2-eq'),
             ('Count_Locomotives', 'assets'), ('Count_Refuelers', 'assets'),
             ('Energy_Costs', 'USD'),
             ('Energy_Per_Freight_Moved', 'kWh per car-mile')
         ],
         calculate_multiyear_metrics: bool = True) -> pl.DataFrame

Given a set of simulation results and the associated consist plans, computes economic and environmental metrics.

Arguments:

  • scenario_infos - List (with one entry per scenario year) of Scenario Info objects
  • metricsToCalc - List of metrics to calculate, each specified as a tuple consisting of a metric and the desired unit

  • calculate_multiyear_metrics - True if multi-year rollout costs (including levelized cost) are to be computed Outputs:

  • values - DataFrame of output and intermediate metrics (metric name, units, value, and scenario year)

calculate_annual_metric

def calculate_annual_metric(metric_name: str, units: str,
                            info: ScenarioInfo) -> MetricType

Given a years' worth of simulation results and the associated consist plan, computes the requested metric.

Arguments:

  • thisRow - DataFrame containing the requested metric and requested units

  • info - A scenario information object representing parameters and results for a single year Outputs:

  • values - DataFrame of requested output metric + any intermediate metrics (metric name, units, value, and scenario year)

calculate_rollout_lcotkm

def calculate_rollout_lcotkm(values: MetricType) -> MetricType

Given a DataFrame of each year's costs and gross freight deliveries, computes the multi-year levelized cost per gross tonne-km of freight delivered.

Arguments:

  • values - DataFrame containing total costs and gross freight deliveries for each modeled scenario year Outputs:

    DataFrame of LCOTKM result (metric name, units, value, and scenario year)

calculate_energy_per_freight

def calculate_energy_per_freight(info: ScenarioInfo, units: str) -> MetricType

Given a years' worth of simulation results, computes a single year energy usage per unit of freight moved.

Arguments:

  • info - A scenario information object representing parameters and results for a single year

  • units - Requested units Outputs:

    DataFrame of energy usage per freight moved (metric name, units, value, and scenario year)

calculate_energy_cost

def calculate_energy_cost(info: ScenarioInfo, units: str) -> MetricType

Given a years' worth of simulation results, computes a single year energy costs.

Arguments:

  • info - A scenario information object representing parameters and results for a single year

  • units - Requested units Outputs:

    DataFrame of energy costs + intermediate metrics (metric name, units, value, and scenario year)

calculate_diesel_use

def calculate_diesel_use(info: ScenarioInfo, units: str)

Given a years' worth of simulation results, computes a single year diesel fuel use.

Arguments:

  • info - A scenario information object representing parameters and results for a single year

  • units - Requested units. Outputs:

    DataFrame of diesel use (metric name, units, value, and scenario year)

calculate_electricity_use

def calculate_electricity_use(info: ScenarioInfo, units: str) -> MetricType

Given a years' worth of simulation results, computes a single year grid electricity use.

Arguments:

  • info - A scenario information object representing parameters and results for a single year

  • units - Requested units Outputs:

    DataFrame of grid electricity use (metric name, units, value, and scenario year)

calculate_freight_moved

def calculate_freight_moved(info: ScenarioInfo, units: str) -> MetricType

Given a years' worth of simulation results, computes a single year quantity of freight moved

Arguments:

  • info - A scenario information object representing parameters and results for a single year

  • units - Requested units Outputs:

    DataFrame of quantity of freight (metric name, units, value, and scenario year)

calculate_ghg

def calculate_ghg(info: ScenarioInfo, units: str) -> MetricType

Given a years' worth of simulation results, computes a single year GHG emissions from energy use

Arguments:

  • info - A scenario information object representing parameters and results for a single year

  • units - Requested units Outputs:

    DataFrame of GHG emissions from energy use (metric name, units, value, and scenario year)

calculate_locomotive_counts

def calculate_locomotive_counts(info: ScenarioInfo, _) -> MetricType

Given a single scenario year's locomotive consist plan, computes the year's locomotive fleet composition

Arguments:

  • info - A scenario information object representing parameters and results for a single year Outputs:

    DataFrame of locomotive fleet composition metrics (metric name, units, value, and scenario year)

calculate_refueler_counts

def calculate_refueler_counts(info: ScenarioInfo, _) -> MetricType

Given a single scenario year's results, counts how many refuelers were included in the simulation.

Arguments:

  • info - A scenario information object representing parameters and results for a single year Outputs:

    DataFrame of locomotive fleet composition metrics (metric name, units, value, and scenario year)

calculate_rollout_investments

def calculate_rollout_investments(values: MetricType) -> MetricType

Given multiple scenario years' locomotive fleet compositions, computes additional across-year metrics

Arguments:

  • values - DataFrame with multiple scenario years' locomotive fleet composition metrics Outputs:

    DataFrame of across-year locomotive fleet composition metrics (metric name, units, value, and scenario year)

calculate_rollout_total_costs

def calculate_rollout_total_costs(values: MetricType) -> MetricType

Given multiple scenario years' locomotive fleet compositions, computes total per-year costs

Arguments:

  • values - DataFrame with annual cost metrics Outputs:

    DataFrame of across-year locomotive fleet composition metrics (metric name, units, value, and scenario year)

altrios.user_interface

altrios.loaders

altrios.loaders.powertrain_components

altrios.objectives

altrios.stringline

Created on Wed Dec 15 09:59:15 2021

@author: groscoe2

warmupLength

hours

cooldownLength

hours

numDays

full simulation length

plotName

name of plot file

networkOverride

'C:/Users/MMP-S/Downloads/network.json' if ussing a different network file than one in results directory

checkSubfolders

walk through subfolders looking for results

overwriteExisting

overwrite existing plots

colors

line graph colors, if unspecified, will use default plotly colors

narrowWidth

narrow width of lines

wideWidth

wide width of lines

altrios.utilities

Module for general functions, classes, and unit conversion factors.

KWH_PER_MJ

https://www.eia.gov/energyexplained/units-and-calculators/energy-conversion-calculators.php

package_root

def package_root() -> Path

Returns the package root directory.

resources_root

def resources_root() -> Path

Returns the resources root directory.

cumutrapz

def cumutrapz(x, y)

Returns cumulative trapezoidal integral array for:

Arguments:

  • x - array of monotonically increasing values to integrate over
  • y - array of values being integrated

resample

def resample(df: pd.DataFrame,
             dt_new: Optional[float] = 1.0,
             time_col: Optional[str] = "Time[s]",
             rate_vars: Tuple[str] = [],
             hold_vars: Tuple[str] = []) -> pd.DataFrame

Resamples dataframe df.

Arguments:

  • df: dataframe to resample
  • dt_new: new time step size, default 1.0 s
  • time_col: column for time in s
  • rate_vars: list of variables that represent rates that need to be time averaged
  • hold_vars: vars that need zero-order hold from previous nearest time step (e.g. quantized variables like current gear)

smoothen

def smoothen(signal: npt.ArrayLike, period: int = 9) -> npt.ArrayLike

Apply smoothing to signal, assuming 1 Hz data collection.

copy_demo_files

def copy_demo_files(demo_path: Path = Path("demos"))

Copies demo files from package directory into local directory.

Arguments

  • demo_path: path (relative or absolute in )

Warning

Running this function will overwrite existing files so make sure any files with changes you'd like to keep are renamed.

show_plots

def show_plots() -> bool

Returns true if plots should be displayed based on SHOW_PLOTS environment variable. SHOW_PLOTS defaults to true, and to set it false, run SHOW_PLOTS=false python your_script.py

altrios.stringline_old

Created on Wed Dec 15 09:59:15 2021

@author: groscoe2

altrios.sim_manager

Module for getting the output of the Train Consist Planner and Meet Pass Planner to run 3 week simulation.

main

def main(
    rail_vehicles: List[alt.RailVehicle],
    location_map: Dict[str, List[alt.Location]],
    network: List[alt.Link],
    simulation_days: int = defaults.SIMULATION_DAYS,
    warm_start_days: int = defaults.WARM_START_DAYS,
    scenario_year: int = defaults.BASE_ANALYSIS_YEAR,
    debug: bool = False,
    loco_pool: Optional[pl.DataFrame] = None,
    refuelers: Optional[pl.DataFrame] = None,
    grid_emissions_factors: Optional[pl.DataFrame] = None,
    nodal_energy_prices: Optional[pl.DataFrame] = None,
    train_planner_config: planner_config.TrainPlannerConfig = planner_config.
    TrainPlannerConfig(),
    train_type: alt.TrainType = alt.TrainType.Freight,
    demand_file: Union[pl.DataFrame, Path, str] = str(defaults.DEMAND_FILE),
    network_charging_guidelines: Optional[pl.DataFrame] = None
) -> Tuple[
        pl.DataFrame,
        pl.DataFrame,
        pl.DataFrame,
        pl.DataFrame,
        alt.SpeedLimitTrainSimVec,
        List[List[alt.LinkIdxTime]],
]

Return

return (
    train_consist_plan,
    loco_pool,
    refuelers,
    grid_emissions_factors,
    nodal_energy_prices,
    train_sims,
    timed_paths,
)