agentlib_mpc.utils package

Package containing utils for agentlib_mpc.

Subpackages

• agentlib_mpc.utils.plotting package
  • Submodules
  • agentlib_mpc.utils.plotting.admm_animation module
    • animate_full()
    • init_full()
    • make_animation()
    • make_image()
    • make_lines()
  • agentlib_mpc.utils.plotting.admm_consensus_shades module
    • plot_consensus_shades()
  • agentlib_mpc.utils.plotting.admm_dashboard module
    • create_app()
    • create_coupling_var_plot()
    • create_residuals_plot()
    • get_coupling_variables()
    • get_data_for_plot()
    • get_max_iterations_per_timestep()
    • load_agent_data()
    • main()
    • show_admm_dashboard()
  • agentlib_mpc.utils.plotting.admm_residuals module
    • load_residuals()
    • plot_over_time()
    • plot_single_time_step()
    • residuals_over_time()
  • agentlib_mpc.utils.plotting.basic module
    • EBCColors
      • EBCColors.blue
      • EBCColors.dark_grey
      • EBCColors.dark_red
      • EBCColors.ebc_palette_sort_1
      • EBCColors.ebc_palette_sort_2
      • EBCColors.green
      • EBCColors.grey
      • EBCColors.light_blue
      • EBCColors.light_grey
      • EBCColors.light_red
      • EBCColors.red
    • FontDict
      • FontDict.fontsize
    • Style
    • ValueRange
      • ValueRange.max
      • ValueRange.min
    • make_fig()
    • make_grid()
    • make_side_legend()
  • agentlib_mpc.utils.plotting.discretization_structure module
    • spy_structure()
  • agentlib_mpc.utils.plotting.interactive module
    • draggable_script()
    • get_port()
    • make_components()
    • make_figure_plotly()
    • obj_plot()
    • plot_admm_plotly()
    • plot_mpc_plotly()
    • show_dashboard()
    • solver_return()
  • agentlib_mpc.utils.plotting.ml_model_test module
    • calc_scores()
    • evaluate_model()
    • pairwise_sort()
    • predict_array()
  • agentlib_mpc.utils.plotting.mpc module
    • interpolate_colors()
    • plot_admm()
    • plot_mpc()

Submodules

agentlib_mpc.utils.analysis module

agentlib_mpc.utils.analysis.admm_at_time_step(data: DataFrame | Series, time_step: float = None, variable=None, iteration: float = -1, index_offset: float | Literal['auto'] | bool = True, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds') → DataFrame

Gets the results of an optimization at a time step. :param index_offset: Determines how the index will be updated when loading the data. :param The offset will be subtracted from the time-index. This is useful for results: :param of realtime systems: to cut the number down to something understandable. For example, if the time

index (level 0 of the input Dataframe) is [105, 115, 125] and we give an index_offset of 100, the data will be handled as if the index was [5, 15, 25].

If “auto” or True is provided as an argument, the index will be modified to

start at 0. If 0 or False are provided, no modifications will be made.

Parameters:

• want (where the time value with be a unix time stamp and we) –

  to cut the number down to something understandable. For example, if the time index (level 0 of the input Dataframe) is [105, 115, 125] and we give an index_offset of 100, the data will be handled as if the index was [5, 15, 25].

  If “auto” or True is provided as an argument, the index will be modified to

  start at 0. If 0 or False are provided, no modifications will be made.

• data – The multi-indexed results data frame from the mpc

• time_step – The time step from which results should be shown. If no exact match, shows closest.

• variable – If specified, only returns results with regard to a certain variable.

• iteration – Specifies, from which inner ADMM iteration data should be from. If negative, counts from last iteration. Default -1.

• convert_to – Whether the data should be converted to datetime, minutes etc.

Returns:

A single-indexed Dataframe of the optimization results at the specified time step. If variable is not specified, returns all variables with a double column index, if it is specified returns only values and/or bounds with single indexed columns.

agentlib_mpc.utils.analysis.convert_index(convert_to: Literal['seconds', 'minutes', 'hours', 'days'] | Literal['datetime'], index: Index)

Converts an index from seconds to datetime or another unit :param convert_to: unit, e.g. minutes, hours, datetime :param index: pandas index object

Returns:

agentlib_mpc.utils.analysis.convert_multi_index(data: DataFrame, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] | Literal['datetime'])

Converts an index of an MPC or ADMM results Dataframe to a different unit, assuming it is passed in seconds.

agentlib_mpc.utils.analysis.first_vals_at_trajectory_index(data: DataFrame | Series)

Gets the first values at each time step of a results trajectory.

agentlib_mpc.utils.analysis.get_number_of_iterations(data: DataFrame) → dict[agentlib_mpc.utils.analysis.SimulationTime, int]

Returns the number of iterations at each time instance of the ADMM simulation.

agentlib_mpc.utils.analysis.get_time_steps(data: DataFrame) → Iterable[float]

Returns the time steps at which an MPC step was performed.

agentlib_mpc.utils.analysis.last_vals_at_trajectory_index(data: DataFrame | Series)

Gets the last values at each time step of a results trajectory.

agentlib_mpc.utils.analysis.load_admm(file: Path | str) → DataFrame

agentlib_mpc.utils.analysis.load_mpc(file: Path | str) → DataFrame

agentlib_mpc.utils.analysis.load_mpc_stats(results_file: str | Path) → DataFrame | None

agentlib_mpc.utils.analysis.load_sim(file: Path, causality=None) → DataFrame

agentlib_mpc.utils.analysis.mpc_at_time_step(data: DataFrame, time_step: float, variable=None, variable_type='variable', index_offset: float | Literal['auto'] | bool = True) → DataFrame

Gets the results of an optimization at a time step.

Parameters:

• data – The multi-indexed results data frame from the mpc

• time_step – The time step from which results should be shown. If no exact match, shows closest.

• variable – If specified, only returns results with regard to a certain variable.

• variable_type – The type of the variable provided (parameter, variable, lower, …)

• index_offset – Determines how the index will be updated when loading the data.

• results (The offset will be subtracted from the time-index. This is useful for) –

• systems (of realtime) –

  to cut the number down to something understandable. For example, if the time index (level 0 of the input Dataframe) is [105, 115, 125] and we give an index_offset of 100, the data will be handled as if the index was [5, 15, 25].

  If “auto” or True is provided as an argument, the index will be modified to

  start at 0. If 0 or False are provided, no modifications will be made.

• want (where the time value with be a unix time stamp and we) –

  to cut the number down to something understandable. For example, if the time index (level 0 of the input Dataframe) is [105, 115, 125] and we give an index_offset of 100, the data will be handled as if the index was [5, 15, 25].

  If “auto” or True is provided as an argument, the index will be modified to

  start at 0. If 0 or False are provided, no modifications will be made.

Returns:

A single-indexed Dataframe of the optimization results

at the specified time step. If variable is not specified, returns all variables with a double column index, if it is specified returns only values and/or bounds with single indexed columns.

Return type:

pd.DataFrame

agentlib_mpc.utils.analysis.perform_index_update(data: DataFrame, offset: float | Literal['auto'] | bool, admm: bool = False) → DataFrame

Updates the index of a raw mpc/admm result dataframe, to be offset by a desired time value.

agentlib_mpc.utils.sampling module

agentlib_mpc.utils.sampling.earliest_index(time, arr, stop, start=0)

Helper function for interpolate_to_previous. Finds the current index to which we should forwardfill.

agentlib_mpc.utils.sampling.interpolate_to_previous(target_grid: Iterable[float], original_grid: Iterable[float], values: Sequence[float]) → List[float]

Interpolates to previous value of original grid, i.e. a forward fill.

Stand-in for the following scipy code: tck = interpolate.interp1d(list(original_grid), values, kind=”previous”) result = list(tck(target_grid))

agentlib_mpc.utils.sampling.pairwise(iterable: Iterable)

s -> (s0,s1), (s1,s2), (s2, s3), …

agentlib_mpc.utils.sampling.sample(trajectory: Real | Series | list[numbers.Real] | dict[numbers.Real, numbers.Real], grid: list | ndarray, current: float = 0, method: str = 'linear') → list

Obtain the specified portion of the trajectory.

Parameters:

• trajectory –

  The trajectory to be sampled. Scalars will be expanded onto the grid. Lists need to exactly match the provided grid. Otherwise, a pandas Series is accepted with the timestamp as index. A

  dict with the keys as time stamps is also accepted.

• current – start time of requested trajectory

• grid – target interpolation grid in seconds in relative terms (i.e. starting from 0 usually)

• method – interpolation method, currently accepted: ‘linear’, ‘spline’, ‘previous’

Returns:

Sampled list of values.

Takes a slice of the trajectory from the current time step with the specified length and interpolates it to match the requested sampling. If the requested horizon is longer than the available data, the last available value will be used for the remainder.

Raises:

• ValueError –

• TypeError –

agentlib_mpc.utils.sampling.sample_values_to_target_grid(values: Iterable[float], original_grid: Iterable[float], target_grid: Iterable[float], method: str | InterpolationMethods) → list[float]