Source code for idaes.core.util.tables

#################################################################################
# The Institute for the Design of Advanced Energy Systems Integrated Platform
# Framework (IDAES IP) was produced under the DOE Institute for the
# Design of Advanced Energy Systems (IDAES).
#
# Copyright (c) 2018-2024 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.md and LICENSE.md
# for full copyright and license information.
#################################################################################
# TODO: Missing doc strings
# pylint: disable=missing-module-docstring
# pylint: disable=missing-class-docstring

from collections import OrderedDict

from pandas import DataFrame

from pyomo.environ import value
from pyomo.network import Arc, Port
from pyomo.core.base.var import VarData, Var
from pyomo.core.base.param import Param
from pyomo.core.base.expression import Expression

import idaes.logger as idaeslog
from idaes.core.util.units_of_measurement import report_quantity

_log = idaeslog.getLogger(__name__)

__author__ = "John Eslick, Andrew Lee"


[docs]def arcs_to_stream_dict( blk, additional=None, descend_into=True, sort=False, prepend=None, s=None ): """ Creates a stream dictionary from the Arcs in a model, using the Arc names as keys. This can be used to automate the creation of the streams dictionary needed for the ``create_stream_table_dataframe()`` and ``stream_states_dict()`` functions. Args: blk (pyomo.environ._BlockData): Pyomo model to search for Arcs additional (dict): Additional states to add to the stream dictionary, which aren't represented by arcs in blk, for example feed or product streams without Arcs attached or states internal to a unit model. descend_into (bool): If True, search subblocks for Arcs as well. The default is True. sort (bool): If True sort keys and return an OrderedDict prepend (str): Prepend a string to the arc name joined with a '.'. This can be useful to prevent conflicting names when sub blocks contain Arcs that have the same names when used in combination with descend_into=False. s (dict): Add streams to an existing stream dict. Returns: Dictionary with Arc names as keys and the Arcs as values. """ if s is None: s = {} for c in blk.component_objects(Arc, descend_into=descend_into): key = c.getname() if prepend is not None: key = ".".join([prepend, key]) s[key] = c if additional is not None: s.update(additional) if sort: s = OrderedDict(sorted(s.items())) return s
[docs]def stream_states_dict(streams, time_point=0): """ Method to create a dictionary of state block representing stream states. This takes a dict with stream name keys and stream values. Args: streams : dict with name keys and stream values. Names will be used as display names for stream table, and streams may be Arcs, Ports or StateBlocks. time_point : point in the time domain at which to generate stream table (default = 0) Returns: A pandas DataFrame containing the stream table data. """ stream_dict = OrderedDict() def _stream_dict_add(sb, n, i=None): """add a line to the stream table""" if i is None: key = n else: key = "{}[{}]".format(n, i) stream_dict[key] = sb for n in streams.keys(): if isinstance(streams[n], Arc): for i, a in streams[n].items(): try: # if getting the StateBlock from the destination port # fails for any reason try the source port. This could # happen if a port does not have an associated # StateBlock. For example a surrogate model may not # use state blocks, unit models may handle physical # properties without state blocks, or the port could # be used to serve the purpose of a translator block. sb = _get_state_from_port(a.ports[1], time_point) except: # pylint: disable=W0702 sb = _get_state_from_port(a.ports[0], time_point) _stream_dict_add(sb, n, i) elif isinstance(streams[n], Port): sb = _get_state_from_port(streams[n], time_point) _stream_dict_add(sb, n) else: # _IndexedStateBlock is a private class, so cannot directly test # whether streams[n] is one or not. try: sb = streams[n][time_point] except KeyError as err: raise TypeError( f"Either component type of stream argument {streams[n]} " f"is unindexed or {time_point} is not a member of its " f"indexing set." ) from err _stream_dict_add(sb, n) return stream_dict
[docs]def create_stream_table_dataframe( streams, true_state=False, time_point=0, orient="columns" ): """ Method to create a stream table in the form of a pandas dataframe. Method takes a dict with name keys and stream values. Use an OrderedDict to list the streams in a specific order, otherwise the dataframe can be sorted later. Args: streams : dict with name keys and stream values. Names will be used as display names for stream table, and streams may be Arcs, Ports or StateBlocks. true_state : indicated whether the stream table should contain the display variables define in the StateBlock (False, default) or the state variables (True). time_point : point in the time domain at which to generate stream table (default = 0) orient : orientation of stream table. Accepted values are 'columns' (default) where streams are displayed as columns, or 'index' where stream are displayed as rows. Returns: A pandas DataFrame containing the stream table data. """ stream_attributes = OrderedDict() stream_states = stream_states_dict(streams=streams, time_point=time_point) full_keys = [] # List of all rows in dataframe to fill in missing data stream_attributes["Units"] = {} for key, sb in stream_states.items(): stream_attributes[key] = {} if true_state: disp_dict = sb.define_state_vars() else: disp_dict = sb.define_display_vars() for k in disp_dict: for row, i in enumerate(disp_dict[k]): stream_key = k if i is None else f"{k} {i}" quant = report_quantity(disp_dict[k][i]) stream_attributes[key][stream_key] = quant.m if row == 0 or stream_key not in stream_attributes["Units"]: stream_attributes["Units"][stream_key] = quant.u if stream_key not in full_keys: full_keys.append(stream_key) # Check for missing rows in any stream, and fill with "-" if needed for k, v in stream_attributes.items(): for r in full_keys: if r not in v.keys(): # Missing row, fill with placeholder v[r] = "-" return DataFrame.from_dict(stream_attributes, orient=orient)
[docs]def create_stream_table_ui( streams, true_state=False, time_point=0, orient="columns", precision=5 ): """ Method to create a stream table in the form of a pandas dataframe. Method takes a dict with name keys and stream values. Use an OrderedDict to list the streams in a specific order, otherwise the dataframe can be sorted later. Note: This function process each stream the same way `create_stream_table_dataframe` does. Args: streams : dict with name keys and stream values. Names will be used as display names for stream table, and streams may be Arcs, Ports or StateBlocks. true_state : indicated whether the stream table should contain the display variables define in the StateBlock (False, default) or the state variables (True). time_point : point in the time domain at which to generate stream table (default = 0) orient : orientation of stream table. Accepted values are 'columns' (default) where streams are displayed as columns, or 'index' where stream are displayed as rows. precision: rounding the floating numbers to the give precision. Default is 5 digits after the floating point. Returns: A pandas DataFrame containing the stream table data. """ # Variable Types: class VariableTypes: UNFIXED = "unfixed" FIXED = "fixed" PARAMETER = "parameter" EXPRESSION = "expression" stream_attributes = OrderedDict() stream_states = stream_states_dict(streams=streams, time_point=time_point) full_keys = [] # List of all rows in dataframe to fill in missing data stream_attributes["Units"] = {} for key, sb in stream_states.items(): stream_attributes[key] = {} if true_state: disp_dict = sb.define_state_vars() else: disp_dict = sb.define_display_vars() for k in disp_dict: for row, i in enumerate(disp_dict[k]): stream_key = k if i is None else f"{k} {i}" # Identifying value's variable type var_type = None if isinstance(disp_dict[k][i], (VarData, Var)): if disp_dict[k][i].fixed: var_type = VariableTypes.FIXED else: var_type = VariableTypes.UNFIXED elif isinstance(disp_dict[k][i], Param): var_type = VariableTypes.PARAMETER elif isinstance(disp_dict[k][i], Expression): var_type = VariableTypes.EXPRESSION quant = report_quantity(disp_dict[k][i]) stream_attributes[key][stream_key] = ( round(quant.m, precision), var_type, ) if row == 0 or stream_key not in stream_attributes["Units"]: stream_attributes["Units"][stream_key] = quant.u if stream_key not in full_keys: full_keys.append(stream_key) # Check for missing rows in any stream, and fill with "-" if needed for k, v in stream_attributes.items(): for r in full_keys: if r not in v.keys(): # Missing row, fill with placeholder v[r] = "-" return DataFrame.from_dict(stream_attributes, orient=orient)
[docs]def stream_table_dataframe_to_string(stream_table, **kwargs): """ Method to print a stream table from a dataframe. Method takes any argument understood by DataFrame.to_string """ # Set some default values for keyword arguments na_rep = kwargs.pop("na_rep", "-") justify = kwargs.pop("justify", "center") # the lambda here could be replaced by "{:#.5g}".format, # but arguably that's not as clearly identifiable as a function/callable # pylint: disable-next=unnecessary-lambda float_format = kwargs.pop("float_format", lambda x: "{:#.5g}".format(x)) # Print stream table return stream_table.to_string( na_rep=na_rep, justify=justify, float_format=float_format, **kwargs )
def _get_state_from_port(port, time_point): """ Attempt to find a StateBlock-like object connected to a Port. If the object is indexed both in space and time, assume that the time index comes first. If no components are assigned to the Port, raise a ValueError. If the first component's parent block has no index, raise an AttributeError. If different variables on the port appear to be connected to different state blocks, raise a RuntimeError. Args: port (pyomo.network.Port): a port with variables derived from some single StateBlock time_point : point in the time domain at which to index StateBlock (default = 0) Returns: (StateBlock-like) : an object containing all the components contained in the port. """ vlist = list(port.iter_vars()) states = [v.parent_block().parent_component() for v in vlist] if len(vlist) == 0: raise ValueError( f"No block could be retrieved from Port {port.name} " f"because it contains no components." ) # Check the number of indices of the parent property block. If its indexed # both in space and time, keep the second, spatial index and throw out the # first, temporal index. If that ordering is changed, this method will # need to be changed as well. try: idx = vlist[0].parent_block().index() except AttributeError as err: raise AttributeError( f"No block could be retrieved from Port {port.name} " f"because block {vlist[0].parent_block().name} has no index." ) from err # Assuming the time index is always first and the spatial indices are all # the same if isinstance(idx, tuple): idx = (time_point, vlist[0].parent_block().index()[1:]) else: idx = (time_point,) # This method also assumes that ports with different spatial indices won't # end up at the same port. Otherwise this check is insufficient. if all(states[0] is s for s in states): return states[0][idx] raise RuntimeError( f"No block could be retrieved from Port {port.name} " f"because components are derived from multiple blocks." )
[docs]def generate_table(blocks, attributes, heading=None, exception=True): """ Create a Pandas DataFrame that contains a list of user-defined attributes from a set of Blocks. Args: blocks (dict): A dictionary with name keys and BlockData objects for values. Any name can be associated with a block. Use an OrderedDict to show the blocks in a specific order, otherwise the dataframe can be sorted later. attributes (list or tuple of strings): Attributes to report from a Block, can be a Var, Param, or Expression. If an attribute doesn't exist or doesn't have a valid value, it will be treated as missing data. heading (list or tuple of strings): A list of strings that will be used as column headings. If None the attribute names will be used. exception (bool): If True, raise exceptions related to invalid or missing indexes. If false missing or bad indexes are ignored and None is used for the table value. Setting this to False allows tables where some state blocks have the same attributes with different indexing. (default is True) Returns: (DataFrame): A Pandas dataframe containing a data table """ if heading is None: heading = attributes st = DataFrame(columns=heading) row = [None] * len(attributes) # not a big deal but save time on realloc for key, s in blocks.items(): for i, a in enumerate(attributes): j = None if isinstance(a, (list, tuple)): # if a is list or tuple, assume index supplied try: assert len(a) > 1 except AssertionError: _log.error(f"An index must be supplided for attribute {a[0]}") raise AssertionError( f"An index must be supplided for attribute {a[0]}" ) j = a[1:] a = a[0] v = getattr(s, a, None) if j is not None and v is not None: try: v = v[j] except KeyError: if not exception: v = None else: _log.error(f"{j} is not a valid index of {a}") raise KeyError(f"{j} is not a valid index of {a}") try: v = value(v, exception=False) except TypeError: if not exception: v = None else: _log.error(f"Cannot calculate value of {a} (may be subscriptable)") raise TypeError( f"Cannot calculate value of {a} (may be subscriptable)" ) except ZeroDivisionError: v = None row[i] = v st.loc[key] = row return st