##############################################################################
# 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".
##############################################################################
"""
Jupyter magics for the DMF.
"""
# stdlib
import inspect
import os
import webbrowser
# third-party
import pendulum
from IPython.core.magic import Magics, magics_class, line_magic
from IPython.display import display_markdown
import six
# local
from . import dmf, util, errors, help, workspace
__author__ = 'Dan Gunter <dkgunter@lbl.gov>'
_log = util.get_logger('magics')
[docs]@magics_class
class DmfMagics(Magics):
# Encoding which commands need 'init'.
# Key is command, value is #args: '*'=any, '+'=1 or more.
NEED_INIT_CMD = {'info': '*', 'help': '+'}
def __init__(self, shell):
super(DmfMagics, self).__init__(shell)
self._dmf = None
[docs] @line_magic
def dmf(self, line):
"""DMF outer command
"""
line = line.strip()
if line == '':
tokens = ['help']
else:
tokens = line.split()
subcmd = tokens[0]
if subcmd == 'workspaces':
pass
elif self._dmf is None:
required, why = self._init_required(subcmd, tokens)
if required:
return self._dmf_markdown(why)
submeth = getattr(self, 'dmf_' + subcmd, None)
if submeth is None:
txt = "Unrecognized command `{}`".format(subcmd)
return self._dmf_markdown(txt)
params = tokens[1:]
try:
return submeth(*params)
except Exception as err:
msg = 'Error in `%dmf {}`: {}'.format(subcmd, err)
return self._dmf_markdown(msg)
[docs] @line_magic
def idaes(self, line):
"""%idaes magic
"""
line = line.strip()
if line == '':
tokens = ['help']
else:
tokens = line.split()
subcmd = tokens[0]
if subcmd == 'help':
return self.idaes_help(*tokens[1:])
else:
txt = "Unrecognized command `{}`".format(subcmd)
return self._dmf_markdown(txt)
def _init_required(self, subcmd, tokens):
"""Return whether init is required before this particular
subcommand invocation and, if so, why.
"""
req, why = False, ''
if subcmd in self.NEED_INIT_CMD:
nargs_code = self.NEED_INIT_CMD[subcmd]
if nargs_code == '*' or (nargs_code == '+' and len(tokens) > 1):
req = True
nwhy = '' if nargs_code == '*' else ' with 1 or more args'
why = 'Must call `init` before command `{}`{}'.format(
subcmd, nwhy)
return req, why
[docs] def dmf_init(self, path, *extra):
"""Initialize DMF (do this before most other commands).
Args:
path (str): Full path to DMF home
"""
kwargs, create = {}, True
if len(extra) > 0:
if extra[0].lower() == 'create':
kwargs = {'create': True, 'add_defaults': True}
create = True
else:
_log.warn('Ignoring extra argument to "init"')
try:
self._dmf = dmf.DMF(path, **kwargs)
except errors.DMFWorkspaceNotFoundError:
if not create:
msg = 'Workspace not found at path "{}". ' \
'If you want to create a new workspace, add the word ' \
'"create", after the path, to the command.'.format(path)
return msg
else:
return 'Workspace could not be created at path "{}"'\
.format(path)
except errors.DMFBadWorkspaceError as err:
return 'Error initializing workspace: {}'.format(err)
self._dmf.set_meta({'name': os.path.basename(path)})
return None
[docs] def dmf_workspaces(self, *paths):
"""List DMF workspaces.
Args:
paths (List[str]): Paths to search, use "." by default
"""
wslist = []
if len(paths) == 0:
paths = ['.']
for root in paths:
wslist.extend(workspace.find_workspaces(root))
any_good_workspaces = False
output_table = []
for w in sorted(wslist):
try:
ws = workspace.Workspace(w)
output_table.append((w, ws.name, ws.description))
any_good_workspaces = True
except errors.WorkspaceError:
pass # XXX: Should we print a warning?
if not any_good_workspaces:
# either no paths, or all paths raised an error
return('ERROR: No valid workspaces found\n')
else:
lines = ['| Path | Name | Description |',
'| ---- | ---- | ----------- |']
for row in output_table:
rowstr = '| {} | {} | {} |'.format(row[0], row[1], row[2])
lines.append(rowstr)
listing = '\n'.join(lines)
self._dmf_markdown(listing)
[docs] def dmf_list(self):
"""List resources in the current workspace.
"""
lines = ['| ID | Name(s) | Type | Modified | Description | ',
'| -- | ------- | ---- | -------- | ----------- |']
for rsrc in self._dmf.find():
msince = pendulum.from_timestamp(rsrc.modified).diff_for_humans()
rowstr = '| {id} | {names} | {type} | {mdate} | {desc} |'.format(
id=rsrc.id_, names=','.join(rsrc.aliases), type=rsrc.type,
mdate=msince, desc=rsrc.desc)
lines.append(rowstr)
listing = '\n'.join(lines)
return self._dmf_markdown(listing)
[docs] def dmf_info(self, *topics):
"""Provide information about DMF current state for whatever
'topics' are provided. With no topic, provide general information
about the configuration.
Args:
(List[str]) topics: List of topics
Returns:
None
"""
if topics:
self._dmf_markdown('Sorry, no topic-specific info yet available')
return
# configuration info
text_lines = ['## Configuration']
for key, value in six.iteritems(self._dmf.meta):
hdr = ' * {}'.format(key)
if isinstance(value, list):
text_lines.append('{}:'.format(hdr))
for v in value:
text_lines.append(' - {}'.format(v))
elif isinstance(value, dict):
text_lines.append('{}:'.format(hdr))
for k2, v2 in six.iteritems(value):
text_lines.append(' - {}: {}'.format(k2, v2))
else:
text_lines.append('{}: {}'.format(hdr, value))
conf_info = '\n'.join(text_lines)
# all info
all_info = '\n'.join((conf_info, ))
self._dmf_markdown(all_info)
def _dmf_markdown(self, text):
display_markdown(text, raw=True)
[docs] def dmf_help(self, *names):
"""Provide help on IDAES objects and classes.
Invoking with no arguments gives general help.
Invoking with one or more arguments looks for help in the docs
on the given objects or classes.
"""
if len(names) == 0:
# give some general help for magics
return self._magics_help()
if len(names) > 1:
raise ValueError('Only one object or class at a time')
name = names[0]
# Check some special names first.
# To re-use the <module>.<class> mechanism, translate them into
# some pseudo-classes in the "dmf.help" pseudo-module.
if name.lower() in ('help', 'dmf'):
p_module = 'dmf.help'
p_class = name.title()
helpfiles = help.get_html_docs(self._dmf, p_module, p_class)
else:
helpfiles = self._find_help_for_object(name)
if helpfiles:
self._show_help_in_browser(helpfiles)
else:
return 'No Sphinx docs found for "{}"'.format(name)
[docs] def idaes_help(self, *names):
"""Provide help on IDAES objects and classes.
Invoking with no arguments gives general help.
Invoking with one or more arguments looks for help in the docs
on the given objects or classes.
"""
if len(names) == 0:
# give some general help for magics
return self._magics_help()
if len(names) > 1:
raise ValueError('Only one object or class at a time')
name = names[0]
# Check some special names first.
# To re-use the <module>.<class> mechanism, translate them into
# some pseudo-classes in the "idaes.help" pseudo-module.
if name.lower() in ('help', 'idaes'):
p_module = 'idaes.help'
p_class = name.title()
helpfiles = help.get_html_docs(self._dmf, p_module, p_class)
else:
helpfiles = self._find_help_for_object(name)
if helpfiles:
self._show_help_in_browser(helpfiles)
else:
return 'No Sphinx docs found for "{}"'.format(name)
def _find_help_for_object(self, name):
"""Get object by evaluating name as an expression."""
try:
obj = self.shell.ev(name)
except Exception as err:
raise errors.DmfError('Cannot evaluate object/class for help: '
'{}'.format(err))
_log.debug('Looking for HTML docs for object: {}'.format(obj))
return help.find_html_docs(self._dmf, obj)
@staticmethod
def _show_help_in_browser(helpfiles):
"""Open help docs in the browser."""
first_option = webbrowser._tryorder[0]
if first_option in ('xdg-open', 'chromium'):
# for these browsers, prefixing with file:// allows
# the anchors in the URL to work
url = 'file://' + helpfiles[0]
else:
url = helpfiles[0]
_log.debug('Opening URL "{}"'.format(url))
webbrowser.open_new(url)
def _magics_help(self):
"""Introspect to give a list of commands."""
# Build a dictionary of magic methods and their
# descriptions (from their docstrings).
help_dict = {'dmf': {}, 'idaes': {}}
for name, meth in inspect.getmembers(self, inspect.ismethod):
if name.startswith('dmf_'):
help_dict['dmf'][name] = self._extract_help_text(meth)
elif name.startswith('idaes_'):
help_dict['idaes'][name] = self._extract_help_text(meth)
# build help text from dict
txt = ''
for mname in 'idaes', 'dmf':
txt_list = []
for name in sorted(help_dict[mname].keys()):
cmd = name[len(mname) + 1:]
cmd_help = '* `%{} {}` - {}'.format(
mname, cmd, help_dict[mname][name])
txt_list.append(cmd_help)
section = '{}{} magic commands:'.format(
'\n\n' if txt else '', mname.upper())
txt += section + '\n' + '\n'.join(txt_list)
return self._dmf_markdown(txt)
@staticmethod
def _extract_help_text(meth):
doc = inspect.getdoc(meth)
# extract first sentence (strip off period)
sentence = doc[:doc.find('.')]
# take out line breaks from sentence
hdr = sentence.replace('\n', ' ')
return hdr
# def dmf_add(self, name, *params):
# """Add one resource
# """
# try:
# obj = self.shell.ev(name)
# except Exception as err:
# raise errors.DmfError('Cannot evaluate object/class for "add": {}'
# .format(err))
# if isinstance(obj, idaes_models.core.FlowsheetBlockData):
# rsrc = self._rfactory.create_flowsheet(obj)
# self._dmf.add(rsrc)
# else:
# return 'Unknown object type "{}". Cannot add as a DMF resource.'\
# .format(type(obj))
_registered = False
[docs]def register():
global _registered
if _registered:
return
try:
ip = get_ipython() # noqa: F821
_log.info('Registering DMF magics')
ip.register_magics(DmfMagics)
except: # noqa: E722
pass
_registered = True
register()