agentlib_mpc.utils.plotting package

Submodules

agentlib_mpc.utils.plotting.admm_animation module

Modules that defines functions to be used for automatically creating animations of ADMM convergence

agentlib_mpc.utils.plotting.admm_animation.animate_full(i: int, lines: dict[str, matplotlib.lines.Line2D], annotation: Annotation, data: dict[str, pandas.core.frame.DataFrame], time_step: float)[source]

agentlib_mpc.utils.plotting.admm_animation.init_full(lines: dict[str, matplotlib.lines.Line2D], annotation: Annotation, ax: Axes)[source]

agentlib_mpc.utils.plotting.admm_animation.make_animation(data: dict[str, pandas.core.frame.DataFrame], time_step: float = 0, file_name: str = '', customize: Callable[[Figure, Axes, Style], Tuple[Figure, Axes]] = None, iteration=-1, interval: int = 300)[source]

agentlib_mpc.utils.plotting.admm_animation.make_image(data: dict[str, pandas.core.frame.DataFrame], time_step: float = 0, file_name: str = '', customize: Callable[[Figure, Axes, Style], Tuple[Figure, Axes]] = None, iteration=-1)[source]

agentlib_mpc.utils.plotting.admm_animation.make_lines(labels: list[str], ax: Axes, fig: Figure) → dict[str, matplotlib.lines.Line2D][source]

agentlib_mpc.utils.plotting.admm_consensus_shades module

agentlib_mpc.utils.plotting.admm_consensus_shades.plot_consensus_shades(results: dict[str, dict[str, pandas.core.frame.DataFrame]], data: dict[str, pandas.core.frame.DataFrame], time_step: float)[source]

Parameters:: series –

Returns:

agentlib_mpc.utils.plotting.admm_dashboard module

agentlib_mpc.utils.plotting.admm_dashboard.create_app(agent_data: Dict[str, DataFrame], residuals_df: DataFrame)[source]

Create and configure the Dash app.

Args: agent_data (Dict[str, pd.DataFrame]): Dictionary containing data for each agent. residuals_df (pd.DataFrame): DataFrame containing residuals data.

Returns: dash.Dash: Configured Dash app.

agentlib_mpc.utils.plotting.admm_dashboard.create_coupling_var_plot(plot_data: Dict[str, List[float]], prediction_grid: List[float], coupling_var: str) → Figure[source]

Create a plotly figure for the coupling variable plot.

Args: plot_data (Dict[str, List[float]]): Dictionary with agent names as keys and their values as lists. prediction_grid (List[float]): List of prediction grid values. coupling_var (str): Name of the coupling variable.

Returns: go.Figure: Plotly figure object.

agentlib_mpc.utils.plotting.admm_dashboard.create_residuals_plot(residuals_df: DataFrame, time_step: float) → Figure[source]

Create a plotly figure for the residuals plot.

Args: residuals_df (pd.DataFrame): DataFrame containing residuals data. time_step (float): Selected time step.

Returns: go.Figure: Plotly figure object.

agentlib_mpc.utils.plotting.admm_dashboard.get_coupling_variables(df: DataFrame) → List[str][source]

Identify coupling variables in the dataframe.

Args: df (pd.DataFrame): The MPC data for an agent.

Returns: List[str]: List of coupling variable names.

agentlib_mpc.utils.plotting.admm_dashboard.get_data_for_plot(agent_data: Dict[str, DataFrame], time_step: float, iteration: int, coupling_var: str) → Dict[str, List[float]][source]

Extract data for the coupling variable plot.

Args: agent_data (Dict[str, pd.DataFrame]): Dictionary containing data for each agent. time_step (float): Selected time step. iteration (int): Selected iteration number. coupling_var (str): Name of the selected coupling variable.

Returns: Dict[str, List[float]]: Dictionary with agent names as keys and their values as lists.

agentlib_mpc.utils.plotting.admm_dashboard.get_max_iterations_per_timestep(agent_data: Dict[str, DataFrame]) → Dict[float, int][source]

agentlib_mpc.utils.plotting.admm_dashboard.load_agent_data(directory: str) → Dict[str, DataFrame][source]

Load MPC data for multiple agents from files containing ‘admm’ in their name.

Args: directory (str): Directory path containing the data files.

Returns: Dict[str, pd.DataFrame]: Dictionary with agent names as keys and their data as values.

agentlib_mpc.utils.plotting.admm_dashboard.main()[source]

agentlib_mpc.utils.plotting.admm_dashboard.show_admm_dashboard(data: dict[str, pandas.core.frame.DataFrame], residuals: DataFrame | None = None, scale: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds')[source]

agentlib_mpc.utils.plotting.admm_residuals module

agentlib_mpc.utils.plotting.admm_residuals.load_residuals(file: str | Path) → DataFrame[source]: Loads a residuals csv file in the correct format.

agentlib_mpc.utils.plotting.admm_residuals.plot_over_time(residuals: ~pandas.core.frame.DataFrame, primal_tol: float = None, dual_tol: float = None, max_iters: int = None) -> (<class 'matplotlib.figure.Figure'>, (<class 'matplotlib.axes._axes.Axes'>, <class 'matplotlib.axes._axes.Axes'>))[source]: Plots the final residuals over time.

agentlib_mpc.utils.plotting.admm_residuals.plot_single_time_step(residuals: ~pandas.core.frame.DataFrame, time_step: float = None, primal_tol: float = None, dual_tol: float = None) -> (<class 'matplotlib.figure.Figure'>, <class 'matplotlib.axes._axes.Axes'>)[source]: Plots the decrease of the residuals over iterations for a time step

agentlib_mpc.utils.plotting.admm_residuals.residuals_over_time(residuals: DataFrame) → DataFrame[source]

Evaluates the residuals over time. Takes a raw residuals DataFrame and returns a Dataframe, which has for each time step the number of iterations and the final primal and dual residuals.

Returns:: DataFrame with float index (time in seconds) and the columns (“primal_residual”, “dual_residual”, “iters”)

agentlib_mpc.utils.plotting.basic module

Some basic plotting utilities

class agentlib_mpc.utils.plotting.basic.EBCColors[source]

Bases: object

blue: tuple[float, float, float] = (0.0, 0.32941176470588235, 0.6235294117647059)

dark_grey: tuple[float, float, float] = (0.3058823529411765, 0.30980392156862746, 0.3137254901960784)

dark_red: tuple[float, float, float] = (0.6745098039215687, 0.16862745098039217, 0.10980392156862745)

ebc_palette_sort_1: list[tuple[float, float, float]] = [(0.6745098039215687, 0.16862745098039217, 0.10980392156862745), (0.8666666666666667, 0.25098039215686274, 0.17647058823529413), (0.9215686274509803, 0.5490196078431373, 0.5058823529411764), (0.3058823529411765, 0.30980392156862746, 0.3137254901960784), (0.615686274509804, 0.6196078431372549, 0.6274509803921569), (0.8509803921568627, 0.8509803921568627, 0.8509803921568627), (0.0, 0.32941176470588235, 0.6235294117647059), (0.615686274509804, 0.7647058823529411, 0.9019607843137255), (0.4392156862745098, 0.6784313725490196, 0.2784313725490196)]

ebc_palette_sort_2: list[tuple[float, float, float]] = [(0.8666666666666667, 0.25098039215686274, 0.17647058823529413), (0.0, 0.32941176470588235, 0.6235294117647059), (0.615686274509804, 0.6196078431372549, 0.6274509803921569), (0.4392156862745098, 0.6784313725490196, 0.2784313725490196), (0.6745098039215687, 0.16862745098039217, 0.10980392156862745), (0.3058823529411765, 0.30980392156862746, 0.3137254901960784), (0.9215686274509803, 0.5490196078431373, 0.5058823529411764), (0.615686274509804, 0.7647058823529411, 0.9019607843137255), (0.8509803921568627, 0.8509803921568627, 0.8509803921568627)]

green: tuple[float, float, float] = (0.4392156862745098, 0.6784313725490196, 0.2784313725490196)

grey: tuple[float, float, float] = (0.615686274509804, 0.6196078431372549, 0.6274509803921569)

light_blue: tuple[float, float, float] = (0.615686274509804, 0.7647058823529411, 0.9019607843137255)

light_grey: tuple[float, float, float] = (0.8509803921568627, 0.8509803921568627, 0.8509803921568627)

light_red: tuple[float, float, float] = (0.9215686274509803, 0.5490196078431373, 0.5058823529411764)

red: tuple[float, float, float] = (0.8666666666666667, 0.25098039215686274, 0.17647058823529413)

class agentlib_mpc.utils.plotting.basic.FontDict[source]

Bases: TypedDict

fontsize: float

class agentlib_mpc.utils.plotting.basic.Style(use_tex: bool = False)[source]: Bases: object

class agentlib_mpc.utils.plotting.basic.ValueRange(min: float, max: float)[source]

Bases: object

max: float

min: float

agentlib_mpc.utils.plotting.basic.make_fig(style: Style, customizer: Customizer = None, rows: int = 1) → tuple[plt.Figure, tuple[plt.Axes, ...]][source]
agentlib_mpc.utils.plotting.basic.make_fig(style: Style, customizer: MultiCustomizer = None) → tuple[plt.Figure, plt.Axes]
agentlib_mpc.utils.plotting.basic.make_fig(style: Style) → tuple[plt.Figure, plt.Axes]: Creates a figure and axes with an amount of rows. If rows is specified, return a tuple of axes, else only an ax

agentlib_mpc.utils.plotting.basic.make_grid(ax: Axes)[source]

agentlib_mpc.utils.plotting.basic.make_side_legend(ax: Axes, fig: Figure = None, right_position: float = 1)[source]

agentlib_mpc.utils.plotting.discretization_structure module

agentlib_mpc.utils.plotting.discretization_structure.spy_structure(df: DataFrame, customizer: Callable[[Figure, Axes, Style], Tuple[Figure, Axes]] = None, file: Path = '')[source]

agentlib_mpc.utils.plotting.interactive module

agentlib_mpc.utils.plotting.interactive.draggable_script()[source]

agentlib_mpc.utils.plotting.interactive.get_port()[source]

agentlib_mpc.utils.plotting.interactive.make_components(columns, data, convert_to, stats: DataFrame | None = None) → [<class 'dash.html.Div.Div'>][source]

agentlib_mpc.utils.plotting.interactive.make_figure_plotly() → Figure[source]

agentlib_mpc.utils.plotting.interactive.obj_plot(data, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds') → Graph[source]

agentlib_mpc.utils.plotting.interactive.plot_admm_plotly(series: Series, plot_actual_values: bool = True, plot_predictions: bool = False, step: bool = False, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds')[source]

Parameters:

series – A column of the MPC results Dataframe
fig – Plotly figure to plot on
plot_actual_values – whether the closed loop actual values at the start of each optimization should be plotted (default True)
plot_predictions – whether all predicted trajectories should be plotted
step – whether to use a step plot or a line plot
convert_to – Will convert the index of the returned series to the specified unit (seconds, minutes, hours, days)

Returns:

None

agentlib_mpc.utils.plotting.interactive.plot_mpc_plotly(series: Series, step: bool = False, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds', y_axis_label: str = '') → Figure[source]

Parameters:

title –
y_axis_label –
series – A column of the MPC results Dataframe
plot_actual_values – whether the closed loop actual values at the start of each optimization should be plotted (default True)
plot_predictions – whether all predicted trajectories should be plotted
step – whether to use a step plot or a line plot
convert_to – Will convert the index of the returned series to the specified unit (seconds, minutes, hours, days)

Returns:

Figure

agentlib_mpc.utils.plotting.interactive.show_dashboard(data: DataFrame, stats: DataFrame | None = None, scale: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds', port: int | None = None, variables_to_plot: list = None)[source]

agentlib_mpc.utils.plotting.interactive.solver_return(data, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds') → Graph[source]

agentlib_mpc.utils.plotting.ml_model_test module

agentlib_mpc.utils.plotting.ml_model_test.calc_scores(errors: ndarray, metric: Callable) → float[source]

agentlib_mpc.utils.plotting.ml_model_test.evaluate_model(training_data: TrainingData, model: CasadiPredictor | SerializedMLModel, metric: Callable = None, show_plot: bool = True, save_path: Path | None = None)[source]: Tests the Model on test data

agentlib_mpc.utils.plotting.ml_model_test.pairwise_sort(*arrays: tuple[numpy.ndarray, numpy.ndarray])[source]

agentlib_mpc.utils.plotting.ml_model_test.predict_array(df: DataFrame, ml_model: CasadiPredictor, outputs: Index) → DataFrame[source]

agentlib_mpc.utils.plotting.mpc module

agentlib_mpc.utils.plotting.mpc.interpolate_colors(progress: float, colors: list[tuple[float, float, float]]) → tuple[float, float, float][source]

Interpolates colors based on a sample number. To be used when plotting many mpc predictions in one plot, so a fade from old to new predictions can be seen.

Original credit to Max Berktold.

Parameters:

progress –
colors –

Returns:

agentlib_mpc.utils.plotting.mpc.plot_admm(series: Series, ax: Axes, plot_actual_values: bool = True, plot_predictions: bool = False, step: bool = False, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds')[source]

Parameters:

series – A column of the MPC results Dataframe
ax – which Axes to plot on
plot_actual_values – whether the closed loop actual values at the start of each optimization should be plotted (default True)
plot_predictions – whether all predicted trajectories should be plotted
step –
convert_to – Will convert the index of the returned series to the specified unit (seconds, minutes, hours, days)

Returns:

agentlib_mpc.utils.plotting.mpc.plot_mpc(series: Series, ax: Axes, plot_actual_values: bool = True, plot_predictions: bool = False, step: bool = False, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds')[source]

Parameters:

series – A column of the MPC results Dataframe
ax – which Axes to plot on
plot_actual_values – whether the closed loop actual values at the start of each optimization should be plotted (default True)
plot_predictions – whether all predicted trajectories should be plotted
step –
convert_to – Will convert the index of the returned series to the specified unit (seconds, minutes, hours, days)

Returns:

agentlib_mpc.utils.plotting.mpc_dashboard module

agentlib_mpc.utils.plotting.mpc_dashboard.detect_index_type(data: DataFrame) → Tuple[bool, bool][source]

Detect the type of index in the DataFrame.

Parameters:: data – DataFrame to check
Returns:: Tuple of (is_multi_index, is_datetime)

agentlib_mpc.utils.plotting.mpc_dashboard.is_mhe_data(series: Series) → bool[source]

Detect if the data represents MHE (Moving Horizon Estimator) results rather than MPC predictions.

Parameters:: series – Series of predictions with time steps as index
Returns:: True if the data appears to be MHE data, False otherwise
Return type:: bool

agentlib_mpc.utils.plotting.mpc_dashboard.launch_dashboard_from_results(results: Dict[str, Dict[str, Any]], scale: str = 'hours', step: bool = False) → bool[source]

Launch the multi-agent dashboard from results dictionary returned by mas.get_results().

Parameters:

results – Dictionary with agent results from mas.get_results()
scale – Time scale for plotting (“seconds”, “minutes”, “hours”, “days”)
step – Whether to use step plots

Returns:

True if dashboard was launched, False otherwise

Return type:

bool

agentlib_mpc.utils.plotting.mpc_dashboard.make_components(data: DataFrame, convert_to: str, stats: DataFrame | None = None, use_datetime: bool = False, step: bool = False) → Div[source]

Create dashboard components from MPC data and stats.

Parameters:

data – DataFrame with MPC data
convert_to – Time unit for plotting
stats – Optional DataFrame with MPC statistics
use_datetime – Whether to interpret timestamps as datetime
step – Whether to use step plots

Returns:

Dash HTML Div containing all components

agentlib_mpc.utils.plotting.mpc_dashboard.plot_mpc_plotly(series: Series, step: bool = False, convert_to: Literal['seconds', 'minutes', 'hours', 'days'] = 'seconds', y_axis_label: str = '', use_datetime: bool = False, max_predictions: int = 1000) → Figure[source]

Create a plotly figure from MPC prediction series.

Parameters:

series – Series of MPC predictions with time steps as index
step – Whether to display step plots (True) or continuous lines (False)
convert_to – Unit for time conversion
y_axis_label – Label for y-axis
use_datetime – Whether to interpret timestamps as datetime
max_predictions – Maximum number of predictions to show (for performance)

Returns:

Plotly figure object

agentlib_mpc.utils.plotting.mpc_dashboard.process_mas_results(results: Dict[str, Dict[str, Any]]) → Dict[str, Dict[str, Any]][source]

Process results from LocalMASAgency to prepare them for visualization.

Parameters:: results – Raw results from mas.get_results()
Returns:: Processed results ready for dashboard visualization

agentlib_mpc.utils.plotting.mpc_dashboard.reduce_triple_index(df: DataFrame) → DataFrame[source]

Reduce a triple-indexed DataFrame to a double index by keeping only the rows with the largest level 1 index for each unique level 0 index.

Parameters:: df – DataFrame with either double or triple index
Returns:: DataFrame with double index

agentlib_mpc.utils.plotting.mpc_dashboard.show_multi_room_dashboard(results: Dict[str, Dict[str, Any]], scale: str = 'hours', step: bool = False)[source]

Show a dashboard with dropdown selection for different agents/rooms.

Parameters:

results – Dictionary with agent results from mas.get_results()
scale – Time scale for plotting (“seconds”, “minutes”, “hours”, “days”)
step – Whether to use step plots