Source code for idaes.vis.plot

##############################################################################
# Institute for the Design of Advanced Energy Systems Process Systems
# Engineering Framework (IDAES PSE Framework) Copyright (c) 2018, by the
# software owners: The Regents of the University of California, through
# Lawrence Berkeley National Laboratory,  National Technology & Engineering
# Solutions of Sandia, LLC, Carnegie Mellon University, West Virginia
# University Research Corporation, et al. All rights reserved.
# 
# Please see the files COPYRIGHT.txt and LICENSE.txt for full copyright and
# license information, respectively. Both files are also available online
# at the URL "https://github.com/IDAES/idaes".
##############################################################################
from bokeh.models import ColumnDataSource,LabelSet
from bokeh.models import Arrow, OpenHead, NormalHead
from bokeh.models import Label, HoverTool, CircleCross, X
from bokeh.plotting import figure, output_file, show, ColumnDataSource
from bokeh.resources import CDN
from bokeh.embed import file_html
from bokeh.io import save
from idaes.vis.plot_utils import *
from idaes.vis.vis_exceptions import *
from operator import itemgetter
from statistics import median
import warnings


[docs]class Plot(object): def __init__(self, current_plot=None): """Set the current plot in this constructor. Args: current_plot: A bokeh plot object. """ active_dev_warning = '''The visualization library is still in active development and we hope to improve on it in future releases. Please use its functionality at your own discretion.''' warnings.warn(active_dev_warning, stacklevel=3) self.current_plot = current_plot self.html = '' self.filename = ''
[docs] def annotate(self, x, y, label): """Annotate a plot with a given point and a label. Args: x: Value of independent variable. y: Value of dependent variable. label: Text label. Returns: None Raises: None """ source = ColumnDataSource(data=dict(x=[x], y=[y], labels=[label])) self.current_plot.circle(x, y, line_color='red', fill_color='red') labels = LabelSet(x='x', y='y', text='labels', level='glyph', source=source, render_mode='canvas', text_font_size='9pt') self.current_plot.add_layout(labels)
[docs] def resize(self, height=-1, width=-1): """Resize a plot's height and width. Args: height: Height in screen units. width: Width in screen units. Returns: None Raises: None """ try: if height != -1: self.current_plot.plot_height = height if width != -1: self.current_plot.plot_width = width except AttributeError: raise MissingCurrentPlot("No plot to resize!")
[docs] def save(self, destination): """Save the current plot object to HTML in filepath provided by destination. Args: destination: Valid file path to save HTML to. Returns: filename where HTML is saved. Raises: None """ if not self.html: self.html = file_html(self.current_plot, CDN, None) with open(destination, 'w') as output_file: output_file.write(self.html) self.filename = destination return self.filename
[docs] def show(self, in_notebook=True): """Display plot in a Jupyter notebook. Args: self: Plot object. in_notebook: Display in Jupyter notebook or generate HTML file. Returns: None Raises: None """ if in_notebook: try: from IPython.core.display import display, HTML self.html = file_html(self.current_plot, CDN, None) display(HTML(self.html)) except Exception as e: raise ShowFailedException("Showing the plot failed.\ Encountered exception {0}".format(e)) else: show(self.current_plot)
[docs] @classmethod def heat_exchanger_network(cls, exchangers, stream_list, mark_temperatures_with_tooltips=False, mark_modules_with_tooltips=False, stage_width=2, y_stream_step=1): """Plot a heat exchanger network diagram. Args: exchangers: List of exchangers where each exchanger is a dict of the form: .. code-block:: python {'hot': 'H2', 'cold': 'C1', 'Q': 1400, 'A': 159, 'annual_cost': 28358, 'stg': 2} where `hot` is the hot stream name, `cold` is the cold stream name, `A` is the area (in m^2), `annual_cost` is the annual cost in $, `Q` is the amount of heat transferred from one stream to another in a given exchanger and `stg` is the stage the exchanger belongs to. The `utility_type`, if present, will specify if we plot the cold stream as water (:class:`idaes.vis.plot_utils.HENStreamType.cold_utility`) or the hot stream as steam (:class:`idaes.vis.plot_utils.HENStreamType.hot_utility`). Additionally, the exchanger could have the key `modules`, like this: .. code-block:: python {'hot': 'H1', 'cold': 'C1', 'Q': 667, 'A': 50, 'annual_cost': 10979, 'stg': 3, 'modules': {10: 1, 20: 2}} The value of this key is a dictionary where each key is a module area and each value is how many modules of that area are in the exchanger. It's indicated as a tooltip on the resulting diagram. If a stream is involved in multiple exchanges in teh same stage, the stream will split into multiple sub-streams with each sub-stream carrying one of the exchanges. stream_list: List of dicts representing streams where each item is a dict of the form: .. code-block:: python {'name':'H1', 'temps': [443, 435, 355, 333], 'type': HENStreamType.hot} mark_temperatures_with_tooltips: if True, we plot the stream temperatures and assign hover tooltips to them. Otherwise, we label them with text labels. mark_modules_with_tooltips: if True, we plot markers for modules (if present) and assign hover tooltips to them. Otherwise, we don't add module info. stage_width: How many units to use for each stage in the diagram (defaults to 2). y_stream_step: How many units to use to separate each stream/sub-stream from the next (defaults to 1). """ total_streams_tuple = (-1, len(stream_list)+1) total_cost = sum([exchanger['annual_cost'] for exchanger in exchangers]) REGULAR_FONT_SIZE = "10pt" SMALL_FONT_SIZE = "5pt" COLD_STREAM_COLOR = "blue" HOT_STREAM_COLOR = "red" STREAM_TEMP_MARKER_FILL_COLOR = "white" STREAM_TEMP_MARKER_LINE_COLOR = HOT_STREAM_COLOR STAGE_WIDTH = stage_width STAGE_SEPARATOR = 0.5 Y_STREAM_STEP = y_stream_step p = figure(x_range=total_streams_tuple, y_range=total_streams_tuple, title='Annualized Investment Cost = ${0}'.format(total_cost)) p.title.align = 'center' hot_streams = [{k['name']: k['temps']} for k in stream_list if k['type'] == HENStreamType.hot] cold_streams = [{k['name']: k['temps']} for k in stream_list if k['type'] == HENStreamType.cold] stage_set = set(sorted([exchanger['stg'] for exchanger in exchangers])) color_stage_dict = get_color_dictionary(stage_set) stream_y_values, hot_split_streams, cold_split_streams = get_stream_y_values(exchangers, hot_streams, cold_streams, y_stream_step=Y_STREAM_STEP) stage_x_limits = {} max_x = 0 for stage in stage_set: if stage not in stage_x_limits: stage_x_limits[stage] = {'x_start': max_x, 'x_exchanger_start': max_x + STAGE_SEPARATOR, 'x_exchanger_end': max_x + STAGE_WIDTH, 'x_true_end': max_x + STAGE_WIDTH + STAGE_SEPARATOR} max_x = max_x + STAGE_WIDTH hot_stream_names = [list(stream.keys())[0] for stream in hot_streams] cold_stream_names = [list(stream.keys())[0] for stream in cold_streams] for i, stage in enumerate(stage_set): stage_exchangers = [exchanger for exchanger in exchangers if exchanger['stg'] == stage and not is_hot_or_cold_utility(exchanger)] hot_split_streams_by_stage = [hot_split_stream[0] for hot_split_stream in hot_split_streams if hot_split_stream[2] == stage] cold_split_streams_by_stage = [cold_split_stream[0] for cold_split_stream in cold_split_streams if cold_split_stream[2] == stage] used_split_y_values_for_stage = [] is_first_stage = i == 0 is_last_stage = i == (len(stage_set) - 1) # Figure out distance between each exchanger for this stage leaving some leeway # for closing out split stream rectangles: start_x = stage_x_limits[stage]['x_exchanger_start'] end_x = stage_x_limits[stage]['x_exchanger_end']-STAGE_SEPARATOR if len(stage_exchangers) > 0: dist_between_exchanger = ( end_x - start_x)/len(stage_exchangers) max_x = start_x for exchanger in stage_exchangers: exchanger_x_start = max_x + dist_between_exchanger hot_stream = exchanger['hot'] cold_stream = exchanger['cold'] if hot_stream in hot_split_streams_by_stage: # Give me the 1st unused hot stream split y value for this stage: for y_value in stream_y_values[hot_stream]['split_y_values']: if y_value not in used_split_y_values_for_stage: used_split_y_values_for_stage.append(y_value) exchanger_y_start = y_value break else: # Just assign it to the default y value: exchanger_y_start = stream_y_values[hot_stream]['default_y_value'] if cold_stream in cold_split_streams_by_stage: # Give me the 1st unused cold stream split y value for this stage: for y_value in stream_y_values[cold_stream]['split_y_values']: if y_value not in used_split_y_values_for_stage: used_split_y_values_for_stage.append(y_value) exchanger_y_end = y_value break else: # Just assign it to the default y value: exchanger_y_end = stream_y_values[cold_stream]['default_y_value'] p = plot_line_segment(p, exchanger_x_start, exchanger_x_start, exchanger_y_start, exchanger_y_end, color=color_stage_dict[stage], legend="stage {0}".format(str(stage))) p = add_exchanger_labels(p, exchanger_x_start, exchanger_y_start, exchanger_y_end, SMALL_FONT_SIZE, exchanger, STREAM_TEMP_MARKER_LINE_COLOR, STREAM_TEMP_MARKER_FILL_COLOR, mark_modules_with_tooltips) max_x += dist_between_exchanger for stream in stream_y_values.keys(): color = HOT_STREAM_COLOR if stream in hot_stream_names else COLD_STREAM_COLOR split_streams_list = hot_split_streams if stream in hot_stream_names else cold_split_streams if stream not in hot_split_streams_by_stage and stream not in cold_split_streams_by_stage: # Default y stream: p = plot_line_segment(p, stage_x_limits[stage]['x_start'], stage_x_limits[stage]['x_true_end'], stream_y_values[stream]['default_y_value'], stream_y_values[stream]['default_y_value'], color=color) else: # Split y stream: split_count = [stream_record[3] for stream_record in split_streams_list if stream_record[2] == stage and stream_record[0] == stream][0] split_y_values = stream_y_values[stream]['split_y_values'][0:split_count] vertical_closer_y_start = max(stream_y_values[stream]['default_y_value'], max(split_y_values)) vertical_closer_y_end = min(stream_y_values[stream]['default_y_value'], min(split_y_values)) p = plot_line_segment(p, stage_x_limits[stage]['x_exchanger_start'], stage_x_limits[stage]['x_exchanger_start'], vertical_closer_y_start, vertical_closer_y_end, color=color) p = plot_line_segment(p, stage_x_limits[stage]['x_exchanger_end'], stage_x_limits[stage]['x_exchanger_end'], vertical_closer_y_start, vertical_closer_y_end, color=color) for y in sorted(split_y_values): p = plot_line_segment(p, stage_x_limits[stage]['x_exchanger_start'], stage_x_limits[stage]['x_exchanger_end'], y, y, color=color) # 1st stage: # For hot streams: # - Plot line segments, stream names and highest tempreature label. # For cold streams: # - Plot arrows and lowest temperature label. # Plot hot utility exchanges. if is_first_stage: for stream in stream_y_values.keys(): temp_list = [stream_dict['temps'] for stream_dict in stream_list if stream_dict['name'] == stream][0] if stream in hot_stream_names: # Plot line segments that start/end a hot/cold stream # and add relevant labels in the 1st stage: p = plot_line_segment(p, stage_x_limits[stage]['x_start'], stage_x_limits[stage]['x_exchanger_start'], stream_y_values[stream]['default_y_value'], stream_y_values[stream]['default_y_value'], color=HOT_STREAM_COLOR) temp_start_label = Label( x=stage_x_limits[stage]['x_start'], y=stream_y_values[stream]['default_y_value'], x_offset=-10, y_offset=2, text_font_size=REGULAR_FONT_SIZE, text=str(temp_list[0])) p.add_layout(temp_start_label) stream_name_label = Label( x=stage_x_limits[stage]['x_start'], y=stream_y_values[stream]['default_y_value'], x_offset=-10, y_offset=-12, text_font_size=REGULAR_FONT_SIZE, text=stream) p.add_layout(stream_name_label) if stream in cold_stream_names: p = plot_stream_arrow(p, COLD_STREAM_COLOR, temp_list[-1], REGULAR_FONT_SIZE, stage_x_limits[stage]['x_exchanger_start'], stage_x_limits[stage]['x_start'], stream_y_values[stream]['default_y_value'], stream_y_values[stream]['default_y_value']) # plot steam exchange: steam_exchanger = [exchanger for exchanger in exchangers if ( 'utility_type' in exchanger and exchanger['utility_type'] == HENStreamType.hot_utility)] if len(steam_exchanger) > 0: cold_stream_name = steam_exchanger[0]['cold'] y_steam = stream_y_values[cold_stream_name]['default_y_value'] p.add_layout(Arrow(line_color=color_stage_dict[stage], end=OpenHead(line_color=color_stage_dict[stage], line_width=2, size=10), x_start=0.25, y_start=y_steam-0.25, x_end=0.25, y_end=y_steam+0.25)) p = plot_line_segment(p, 0.25, 0.25, y_steam-0.25, y_steam+0.25, color=color_stage_dict[stage], legend="stage {0}".format(str(stage))) label_s = Label(x=0.25, y=y_steam-0.25, x_offset=-10, y_offset=-15, text_font_size=SMALL_FONT_SIZE, text=steam_exchanger[0]['hot']) p.add_layout(label_s) label_q = Label(x=0.25, y=y_steam+0.25, x_offset=-5, text_font_size=SMALL_FONT_SIZE, text='{0} kW'.format(steam_exchanger[0]['Q'])) p.add_layout(label_q) # Last stage: # For cold streams: # - Plot line segments, stream names and highest tempreature label. # For hot streams: # - Plot arrows and lowest temperature label. # Plot cold utility exchanges. elif is_last_stage: for stream in stream_y_values.keys(): temp_list = [stream_dict['temps'] for stream_dict in stream_list if stream_dict['name'] == stream][0] if stream in cold_stream_names: # Plot line segments that start/end a hot/cold stream # and add relevant labels in the 1st stage: p = plot_line_segment(p, stage_x_limits[stage]['x_exchanger_end'], stage_x_limits[stage]['x_true_end'], stream_y_values[stream]['default_y_value'], stream_y_values[stream]['default_y_value'], color=COLD_STREAM_COLOR) temp_start_label = Label( x=stage_x_limits[stage]['x_true_end'], y=stream_y_values[stream]['default_y_value'], x_offset=-10, y_offset=2, text_font_size=REGULAR_FONT_SIZE, text=str(temp_list[0])) p.add_layout(temp_start_label) stream_name_label = Label( x=stage_x_limits[stage]['x_true_end'], y=stream_y_values[stream]['default_y_value'], x_offset=-10, y_offset=-12, text_font_size=REGULAR_FONT_SIZE, text=stream) p.add_layout(stream_name_label) if stream in hot_stream_names: p = plot_stream_arrow(p, HOT_STREAM_COLOR, temp_list[-1], REGULAR_FONT_SIZE, stage_x_limits[stage]['x_exchanger_end'], stage_x_limits[stage]['x_true_end'], stream_y_values[stream]['default_y_value'], stream_y_values[stream]['default_y_value']) # plot water exchange: water_exchanger = [exchanger for exchanger in exchangers if ( 'utility_type' in exchanger and exchanger['utility_type'] == HENStreamType.cold_utility)] if len(water_exchanger) > 0: hot_stream_name = water_exchanger[0]['hot'] y_water = stream_y_values[hot_stream_name]['default_y_value'] x_water = stage_x_limits[stage]['x_exchanger_end'] \ + (stage_x_limits[stage]['x_true_end'] - stage_x_limits[stage]['x_exchanger_end'])/2 p.add_layout(Arrow(line_color=color_stage_dict[stage], end=OpenHead(line_color=color_stage_dict[stage], line_width=2, size=10), x_start=x_water, y_start=y_water+0.25, x_end=x_water, y_end=y_water-0.25)) p = plot_line_segment(p, x_water, x_water, y_water+0.25, y_water-0.25, color=color_stage_dict[stage], legend="stage {0}".format(str(stage))) label_s = Label(x=x_water, y=y_water-0.25, x_offset=-10, y_offset=-15, text_font_size=SMALL_FONT_SIZE, text=water_exchanger[0]['cold']) p.add_layout(label_s) label_q = Label(x=x_water, y=y_water+0.25, x_offset=-5, text_font_size=SMALL_FONT_SIZE, text='{0} kW'.format(water_exchanger[0]['Q'])) p.add_layout(label_q) # Otherwise, plot line segments after the exchangers to finish off a stage: else: for stream in stream_y_values.keys(): color = COLD_STREAM_COLOR if stream in cold_stream_names else HOT_STREAM_COLOR p = plot_line_segment(p, stage_x_limits[stage]['x_exchanger_end'], stage_x_limits[stage]['x_true_end'], stream_y_values[stream]['default_y_value'], stream_y_values[stream]['default_y_value'], color=color) # Plot stage temperatures for each stream for stream in stream_y_values.keys(): temp_list = [stream_dict['temps'] for stream_dict in stream_list if stream_dict['name'] == stream][0] if len(temp_list) > 1: temp_list = temp_list[1:-1] if stage <= len(temp_list): temp_x = stage_x_limits[stage]['x_exchanger_end'] \ + (stage_x_limits[stage]['x_true_end'] - stage_x_limits[stage]['x_exchanger_end'])/2 temp_to_mark = temp_list[stage - 1] marker_temp = CircleCross(x=temp_x, y=stream_y_values[stream]['default_y_value'], line_color=STREAM_TEMP_MARKER_LINE_COLOR, fill_color=STREAM_TEMP_MARKER_FILL_COLOR, size=10) if mark_temperatures_with_tooltips: source = ColumnDataSource(data=dict(x=[temp_x], y=[stream_y_values[stream] ['default_y_value']], temps=['{0}'.format(str(temp_to_mark))])) glyph_marker = p.add_glyph( source_or_glyph=source, glyph=marker_temp) hover_temps = HoverTool(renderers=[glyph_marker], tooltips=[('Temperature', '@temps')]) p.add_tools(hover_temps) else: glyph_marker = p.add_glyph(marker_temp) temp_label = Label( x=temp_x, y=stream_y_values[stream]['default_y_value'], text_font_size=SMALL_FONT_SIZE, text='{0}'.format(str(temp_to_mark))) p.add_layout(temp_label) p = turn_off_grid_and_axes_ticks(p) return Plot(current_plot=p)
[docs] @classmethod def profile(cls, data_frame, x='', y=[], title='', xlab='', ylab='', y_axis_type='auto', legend=[]): """ A profile plot includes 2 dependent variables and a single independent variable. Based on the Jupyter notebook `here <https://github.com/IDAES/model_contrib/blob/master/examples/mea_simple/mea_example_nb_01.ipynb>`_. Args: data_frame: a data frame with keys contained in x and y. x: Key in data-frame to use as x-axis. y: Keys in data-frame to use as y-axis. title: Title for a plot. xlab: Label for x-axis. ylab: Label for y-axis. y_axis_type: Specify "log" to pass logarithmic scale. legend : List of strings matching y. Returns: Plot object on success. Raises: MissingVariablesException: Dependent variable or their data not passed. BadDataFrameException: No data-frame was generated for the model object. """ # Check that some dependent variable names were passed: try: if validate(data_frame, x, y, legend=legend): # output to static HTML file (commented out for now) # output_file("profile_plot.html") p = figure( tools="pan,box_zoom,reset,save", title=title, x_axis_label=xlab, y_axis_label=ylab, y_axis_type=y_axis_type ) # Plotting both y1 and y2 against x: p.line(data_frame[x], data_frame[y[0]], legend=legend[0], line_color="green") p.line(data_frame[x], data_frame[y[1]], legend=legend[1], line_color="blue") else: raise BadDataFrameException("Invalid data frame passed") except Exception as e: raise e return Plot(current_plot=p)
[docs] @classmethod def property_model(cls, data_frame, x='', y=[], title='', xlab='', ylab='', y_axis_type='auto', legend=[]): """Draw pressure/enthalpy plots for different levels of temperature. Args: data_frame: a data frame with keys contained in x and y. x: Key in data-frame to plot on x-axis. y: Keys in data-frame to plot on y-axis. title: Title for a plot. xlab: Label for x-axis. ylab: Label for y-axis. y_axis_type: Specify "log" to pass logarithmic scale. legend : List of strings matching y. Returns: Plot object on success. Raises: MissingVariablesException: Dependent variable or their data not passed. BadDataFrameException: No data-frame was generated for the model object. """ pass
[docs] @classmethod def goodness_of_fit(cls, data_frame, x='', y=[], title='', xlab='', ylab='', y_axis_type='auto', legend=[]): """Draw y against predicted value (y^) and display (calculate?) value of R^2. Args: data_frame: a data frame with keys contained in x and y. x: Key in data-frame to use as x-axis. y: Keys in data-frame to plot on y-axis. title: Title for a plot. xlab: Label for x-axis. ylab: Label for y-axis. y_axis_type: Specify "log" to pass logarithmic scale. legend : List of strings matching y. Returns: Plot object on success. Raises: MissingVariablesException: Dependent variable or their data not passed. BadDataFrameException: No data-frame was generated for the model object. """ pass
[docs] @classmethod def tradeoff(cls, data_frame, x='', y=[], title='', xlab='', ylab='', y_axis_type='auto', legend=[]): """Draw some parameter varying and the result on the objective value. Args: data_frame: a data frame with keys contained in x and y. x: Key in data-frame to use as x-axis. y: Keys in data-frame to plot on y-axis. title: Title for a plot. xlab: Label for x-axis. ylab: Label for y-axis. y_axis_type: Specify "log" to pass logarithmic scale. legend : List of strings matching y. Returns: Plot object on success. Raises: MissingVariablesException: Dependent variable or their data not passed. BadDataFrameException: No data-frame was generated for the model object. """ pass
[docs] @classmethod def sensitivity(cls, data_frame, x='', y=[], title='', xlab='', ylab='', y_axis_type='auto', legend=[]): """Need more information. """ pass
[docs] @classmethod def residual(cls, data_frame, x='', y=[], title='', xlab='', ylab='', y_axis_type='auto', legend=[]): """Plot x, some continuous value (e.g: T, P), against Y (% residual value). Is this %-value calculated from variables in the idaes_model_object? Args: data_frame: a data frame with keys contained in x and y. x: Key in data-frame to use as x-axis. y: Keys in data-frame to plot on y-axis. title: Title for a plot. xlab: Label for x-axis. ylab: Label for y-axis. y_axis_type: Specify "log" to pass logarithmic scale. legend : List of strings matching y. Returns: Plot object on success. Raises: MissingVariablesException: Dependent variable or their data not passed. BadDataFrameException: No data-frame was generated for the model object. """ pass
[docs] @classmethod def isobar(cls, data_frame, x='', y=[], title='', xlab='', ylab='', y_axis_type='auto', legend=[]): """Need more information. """ pass
[docs] @classmethod def stream_table(cls, data_frame, title=''): """Display a table for all names in the idaes_model_object_names indexing rows according to row_start and row_stop. Args: data_frame: a data frame with keys contained in x and y. title: Title for a plot. Returns: Plot object on success. Raises: MissingVariablesException: Dependent variable or their data not passed. BadDataFrameException: No data-frame was generated for the model object. """ pass