import cmd
try:
import simplejson as json
except ImportError:
import json
import logging
import os.path
from os import utime
import sys
import traceback
from setuptools.command import easy_install
from zope.interface import implementedBy
from openmdao.main.api import Assembly, Component, Driver, logger, \
set_as_top, get_available_types
from openmdao.main.variable import json_default
from openmdao.main.project import Project, ProjFinder, \
_clear_insts, _match_insts
from openmdao.main.publisher import publish, Publisher
from openmdao.main.mp_support import has_interface, is_instance
from openmdao.main.interfaces import IContainer, IComponent, IAssembly
from openmdao.main.factorymanager import register_class_factory, \
remove_class_factory, get_signature
from openmdao.main.repo import get_repo, find_vcs
from openmdao.main.releaseinfo import __version__, __date__
from openmdao.util.nameutil import isidentifier
from openmdao.util.fileutil import file_md5
from openmdao.util.dep import plugin_groups
from openmdao.gui.util import packagedict
from openmdao.gui.filemanager import FileManager
from openmdao.gui.projdirfactory import ProjDirFactory
[docs]def modifies_project(target):
''' Decorator for methods that update the project (or might do so)
via adding, removing or updating files in the project. This includes
the issuing of commands that end up in the default macro file.
(TODO: should be able to due this directly through the FileManager)
Updates the timestamp of the project _settings.cfg file so that
it can be used to determine when the project was 'last saved'.
'''
def wrapper(self, *args, **kargs):
result = target(self, *args, **kargs)
settings_path = self.files._get_abs_path("_settings.cfg")
with file(settings_path, 'a'):
utime(settings_path, None)
return result
return wrapper
[docs]def modifies_state(target):
''' Decorator for methods that modify the session state (or might do so).
Performs maintenance on root level containers/assemblies and
publishes the potentially updated components.
'''
def wrapper(self, *args, **kwargs):
result = target(self, *args, **kwargs)
self._update_roots()
# self._update_workflows()
if self.publish_updates:
self.publish_components()
return result
return wrapper
[docs]class ConsoleServer(cmd.Cmd):
''' Object which knows how to load an OpenMDAO project and provides a
command line interface and methods to interact with that project.
'''
def __init__(self, name='', host='', publish_updates=True):
cmd.Cmd.__init__(self)
self.intro = 'OpenMDAO ' + __version__ + ' (' + __date__ + ')'
self.prompt = 'OpenMDAO>> '
self._hist = []
self.host = host
self._projname = ''
self.proj = None
self.exc_info = None
self.publish_updates = publish_updates
self._publish_comps = {}
self._log_directory = os.getcwd()
self._log_handler = None
self._log_subscribers = 0
self._partial_cmd = None # for multi-line commands
self.projdirfactory = None
self.files = None
# make sure we have a ProjFinder in sys.path_hooks
if not ProjFinder in sys.path_hooks:
sys.path_hooks = [ProjFinder] + sys.path_hooks
[docs] def set_current_project(self, path):
""" Set current project name. """
# Called by ProjectHandler, since load_project() is too late to
# affect the rendering of the template.
self._projname = os.path.basename(path)
[docs] def get_current_project(self):
""" Get current project name. """
return self._projname
def _update_roots(self):
''' Ensure that all root containers in the project dictionary know
their own name and are set as top.
'''
for k, v in self.proj.items():
if has_interface(v, IContainer):
if v.name != k:
v.name = k
if v._call_cpath_updated:
set_as_top(v)
def _update_workflows(self):
''' Call :meth:`_update_workflow` on drivers to capture any workflow
updates now rather than waiting until they are run.
'''
for k, v in self.proj.items():
if has_interface(v, IContainer):
for driver in [obj for name, obj in v.items(recurse=True)
if is_instance(obj, Driver)]:
driver._update_workflow()
[docs] def publish_components(self):
''' Publish the current component tree and subscribed components.
'''
try:
publish('components', self.get_components())
publish('', {'Dataflow': self.get_dataflow('')})
publish('', {'Workflow': self.get_workflow('')})
except Exception as err:
self._error(err, sys.exc_info())
else:
comps = self._publish_comps.keys()
for pathname in comps:
comp, root = self.get_object(pathname, report=False)
if comp is None:
del self._publish_comps[pathname]
publish(pathname, {})
else:
publish(pathname, comp.get_attributes(io_only=False))
[docs] def send_pub_msg(self, msg, topic):
''' Publish the given message with the given topic.
'''
publish(topic, msg)
def _error(self, err, exc_info):
''' Publish error message and save stack trace if case it's requested.
'''
self._partial_cmd = None
self.exc_info = exc_info
msg = '%s: %s' % (err.__class__.__name__, err)
logger.error(msg)
self._print_error(msg)
def _print_error(self, msg):
''' Publish error message.
'''
try:
publish('console_errors', msg)
except:
logger.error('publishing of message failed')
[docs] def do_trace(self, arg):
''' Print remembered trace from last exception.
'''
if self.exc_info:
exc_type, exc_value, exc_traceback = self.exc_info
traceback.print_exception(exc_type, exc_value, exc_traceback)
else:
print "No trace available."
[docs] def precmd(self, line):
''' This method is called after the line has been input but before
it has been interpreted. If you want to modify the input line
before execution (for example, variable substitution), do it here.
'''
#self._hist += [line.strip()]
return line
@modifies_project
@modifies_state
[docs] def onecmd(self, line):
self._hist.append(line)
try:
cmd.Cmd.onecmd(self, line)
except Exception as err:
self._error(err, sys.exc_info())
[docs] def parseline(self, line):
"""Have to override this because base class version strips the lines,
making multi-line Python commands impossible.
"""
#line = line.strip()
if not line:
return None, None, line
elif line[0] == '?':
line = 'help ' + line[1:]
elif line[0] == '!':
if hasattr(self, 'do_shell'):
line = 'shell ' + line[1:]
else:
return None, None, line
i, n = 0, len(line)
while i < n and line[i] in self.identchars:
i = i + 1
cmd, arg = line[:i], line[i:].strip()
return cmd, arg, line
[docs] def emptyline(self):
# Default for empty line is to repeat last command - yuck
if self._partial_cmd:
self.default('')
[docs] def default(self, line):
''' Called on an input line when the command prefix is not recognized.
In this case we execute the line as Python code.
'''
line = line.rstrip()
if self._partial_cmd is None:
if line.endswith(':'):
self._partial_cmd = line
return
else:
if line:
self._partial_cmd = self._partial_cmd + '\n' + line
if line.startswith(' ') or line.startswith('\t'):
return
else:
line = self._partial_cmd
self._partial_cmd = None
try:
result = self.proj.command(line)
if result is not None:
print result
except Exception as err:
self._error(err, sys.exc_info())
@modifies_project
@modifies_state
[docs] def run(self, pathname, *args, **kwargs):
''' Run the component `pathname`.
If no pathname is specified, use `top`.
'''
pathname = pathname or 'top'
if pathname in self.proj:
print "Executing..."
try:
comp = self.proj.get(pathname)
comp.run(*args, **kwargs)
print "Execution complete."
except Exception as err:
self._error(err, sys.exc_info())
else:
self._print_error("Execution failed: No %r component was found." %
pathname)
@modifies_project
@modifies_state
[docs] def execfile(self, filename):
''' Execfile in server's globals.
'''
try:
self.proj.command("execfile('%s', '%s')" %
(filename, file_md5(filename)))
except Exception as err:
self._error(err, sys.exc_info())
[docs] def get_pid(self):
''' Return this server's :attr:`pid`.
'''
return os.getpid()
[docs] def get_project(self):
''' Return the current project.
'''
return self.proj
[docs] def get_history(self):
''' Return this server's :attr:`_hist`.
'''
return self._hist
[docs] def get_recorded_cmds(self):
''' Return this server's :attr:`_recorded_cmds`.
'''
return self._recorded_cmds[:]
[docs] def get_object(self, pathname, report=True):
''' Get the container with the specified pathname.
Returns the container and the name of the root object.
'''
cont = None
parts = pathname.split('.', 1)
root = parts[0]
if self.proj and root in self.proj:
if root == pathname:
cont = self.proj.get(root)
else:
try:
root_obj = self.proj.get(root)
except Exception as err:
self._error(err, sys.exc_info())
else:
try:
cont = root_obj.get(parts[1])
except AttributeError as err:
# When publishing, don't report remove as an error.
if report:
self._error(err, sys.exc_info())
except Exception as err:
self._error(err, sys.exc_info())
return cont, root
def _get_components(self, cont, pathname=None):
''' Get a heierarchical list of all the components in the given
container or dictionary. The name of the root container, if
specified, is prepended to all pathnames.
'''
comps = []
for k, v in cont.items():
if is_instance(v, Component):
comp = {}
if cont is self.proj._project_globals:
comp['pathname'] = k
else:
comp['pathname'] = '.'.join([pathname, k]) if pathname else k
children = self._get_components(v, comp['pathname'])
if len(children) > 0:
comp['children'] = children
comp['type'] = str(v.__class__.__name__)
inames = []
for klass in list(implementedBy(v.__class__)):
inames.append(klass.__name__)
comp['interfaces'] = inames
comps.append(comp)
return comps
[docs] def get_components(self):
''' Get hierarchical dictionary of openmdao objects.
'''
return json.dumps(self._get_components(self.proj._project_globals),
default=json_default)
[docs] def get_connectivity(self, pathname):
''' Get the connectivity data for the assembly with the given pathname
'''
connectivity = {}
asm, root = self.get_object(pathname)
if asm:
try:
connectivity = asm.get_connectivity()
except Exception as err:
self._error(err, sys.exc_info())
return json.dumps(connectivity, default=json_default)
[docs] def get_dataflow(self, pathname):
''' Get the structure of the specified assembly or of the global
namespace if no pathname is specified; consists of the list of
components and the connections between them (i.e., the dataflow).
'''
dataflow = {}
if pathname and len(pathname) > 0:
try:
asm, root = self.get_object(pathname)
if has_interface(asm, IAssembly):
dataflow = asm.get_dataflow()
except Exception as err:
self._error(err, sys.exc_info())
else:
components = []
for k, v in self.proj.items():
if is_instance(v, Component):
inames = [cls.__name__
for cls in list(implementedBy(v.__class__))]
components.append({
'name': k,
'pathname': k,
'type': type(v).__name__,
'valid': v.is_valid(),
'interfaces': inames,
'python_id': id(v)
})
dataflow['components'] = components
dataflow['connections'] = []
dataflow['parameters'] = []
dataflow['constraints'] = []
dataflow['objectives'] = []
return json.dumps(dataflow, default=json_default)
[docs] def get_available_events(self, pathname):
''' Serve a list of events that are available to a driver.
'''
events = []
if pathname:
drvr, root = self.get_object(pathname)
events = drvr.list_available_events()
return json.dumps(events, default=json_default)
[docs] def get_workflow(self, pathname):
''' Get the workflow for the specified driver or assembly.
If no driver or assembly is specified, get the workflows for
all of the top-level assemblies.
'''
flows = []
if pathname:
drvr, root = self.get_object(pathname)
# allow for request on the parent assembly
if is_instance(drvr, Assembly):
drvr = drvr.get('driver')
pathname = pathname + '.driver'
if drvr:
try:
flow = drvr.get_workflow()
except Exception as err:
self._error(err, sys.exc_info())
flows.append(flow)
else:
for k, v in self.proj.items():
if is_instance(v, Assembly):
v = v.get('driver')
if is_instance(v, Driver):
flow = v.get_workflow()
flows.append(flow)
return json.dumps(flows, default=json_default)
[docs] def get_attributes(self, pathname):
''' Get the attributes of the specified object.
'''
attr = {}
comp, root = self.get_object(pathname)
try:
if comp:
attr = comp.get_attributes(io_only=False)
return json.dumps(attr, default=json_default)
except Exception as err:
self._error(err, sys.exc_info())
[docs] def get_passthroughs(self, pathname):
''' Get the inputs and outputs of the assembly's child components
and indicate for each whether or not it is a passthrough variable.
'''
asm, root = self.get_object(pathname)
passthroughs = asm.get_passthroughs()
return json.dumps(passthroughs, default=json_default)
[docs] def get_value(self, pathname):
''' Get the value of the object with the given pathname.
'''
try:
val, root = self.get_object(pathname)
return val
except Exception as err:
self._print_error("error getting value: %s" % err)
[docs] def get_types(self):
''' Get a dictionary of types available for creation.
'''
#Don't want to get variable types showing up, so we exclude
#'openmdao.variable' from this list.
keyset = set(plugin_groups.keys())
exclset = set(['openmdao.variable'])
groups = list(keyset - exclset)
return packagedict(get_available_types(groups))
@modifies_state
[docs] def load_project(self, projdir):
''' Activate the project in the specified directory;
instantiate a file manager and projdirfactory.
'''
_clear_insts()
self.cleanup()
try:
# Start a new log file.
logging.getLogger().handlers[0].doRollover()
self.files = FileManager('files', path=projdir,
publish_updates=self.publish_updates)
self.projdirfactory = ProjDirFactory(projdir,
observer=self.files.observer)
register_class_factory(self.projdirfactory)
self.proj = Project(projdir)
repo = get_repo(projdir)
if repo is None:
find_vcs()[0](projdir).init_repo()
self.proj.activate()
except Exception as err:
self._error(err, sys.exc_info())
@modifies_project
[docs] def commit_project(self, comment=''):
''' Save the current project macro and commit to the project repo.
'''
if self.proj:
try:
repo = get_repo(self.proj.path)
repo.commit(comment)
print 'Committed project in directory ', self.proj.path
except Exception as err:
self._error(err, sys.exc_info())
else:
self._print_error('No Project to commit')
@modifies_project
[docs] def revert_project(self, commit_id=None):
''' Revert to the most recent commit of the project.
'''
if self.proj:
try:
repo = get_repo(self.proj.path)
repo.revert(commit_id)
if commit_id is None:
commit_id = 'latest'
print "Reverted project %s to commit '%s'" % (self.proj.name, commit_id)
except Exception as err:
self._error(err, sys.exc_info())
return err # give the caller an indication that something went wrong so he can
# give the proper error response to the http call if desired. Raising
# an exception here doesn't work
else:
msg = 'No Project to revert'
self._print_error(msg)
return Exception(msg)
[docs] def get_signature(self, classname):
''' Get constructor argument signature for `classname`.
'''
try:
return get_signature(str(classname))
except Exception as err:
self._error(err, sys.exc_info())
[docs] def put_object(self, pathname, classname, args=None):
''' Create or replace object with the given pathname with a new object
of the specified type.
'''
obj, root = self.get_object(pathname, report=False)
if obj:
self.replace_object(pathname, classname, args)
else:
self.add_object(pathname, classname, args)
@modifies_project
@modifies_state
[docs] def add_object(self, pathname, classname, args):
''' Add a new object of the given type to the specified parent.
'''
parentname, dot, name = pathname.rpartition('.')
if isidentifier(name):
name = name.encode('utf8')
if args is None:
args = ''
cmd = 'create("%s"%s)' % (classname, args)
if parentname:
cmd = '%s.add("%s", %s)' % (parentname, name, cmd)
else:
cmd = '%s = set_as_top(%s)' % (name, cmd)
try:
self.proj.command(cmd)
except Exception as err:
self._error(err, sys.exc_info())
else:
self._print_error('Error adding object:'
' "%s" is not a valid identifier' % name)
@modifies_project
@modifies_state
[docs] def replace_object(self, pathname, classname, args=None):
''' Replace existing object with object of the given type.
'''
pathname = pathname.encode('utf8')
parentname, dot, name = pathname.rpartition('.')
if parentname:
try:
self.proj.command('%s.replace("%s", create("%s"))'
% (parentname, name, classname))
except Exception as err:
self._error(err, sys.exc_info())
else:
self._print_error('Error replacing component, no parent: "%s"'
% pathname)
[docs] def cleanup(self):
''' Cleanup various resources.
'''
if self.proj:
self.proj.deactivate()
if self.projdirfactory:
self.projdirfactory.cleanup()
remove_class_factory(self.projdirfactory)
if self.files:
self.files.cleanup()
[docs] def get_files(self):
''' Get a nested dictionary of files.
'''
try:
return self.files.get_files(root=self.proj.path)
except AttributeError:
return {}
[docs] def get_file(self, filename):
''' Get contents of a file.
Returns a tuple of (file contents, mimetype, encoding).
Tuple values will be None if file was not found.
'''
return self.files.get_file(filename)
@modifies_project
[docs] def ensure_dir(self, dirname):
''' Create directory
(does nothing if directory already exists).
'''
return self.files.ensure_dir(dirname)
@modifies_project
[docs] def write_file(self, filename, contents):
''' Write contents to file.
'''
ret = self.files.write_file(filename, contents)
if not ret is True:
return ret
@modifies_project
[docs] def add_file(self, filename, contents):
''' Add file.
'''
return self.files.add_file(filename, contents)
@modifies_project
[docs] def delete_file(self, filename):
''' Delete file from project.
Returns False if file was not found; otherwise returns True.
'''
return self.files.delete_file(filename)
@modifies_project
[docs] def rename_file(self, oldpath, newname):
''' Rename file.
'''
return self.files.rename_file(oldpath, newname)
[docs] def install_addon(self, url, distribution):
print "Installing", distribution, "from", url
easy_install.main(["-U", "-f", url, distribution])
[docs] def add_subscriber(self, pathname, publish):
''' Publish the specified topic.
'''
if pathname in ['', 'components', 'files', 'types',
'console_errors', 'file_errors']:
# these topics are published automatically
return
elif pathname == 'log_msgs':
if publish:
self._start_log_msgs(pathname)
else:
self._stop_log_msgs()
elif pathname.startswith('/'): # treat it as a filename
if publish:
Publisher.register(pathname, pathname[1:])
else:
Publisher.unregister(pathname)
else:
parts = pathname.split('.', 1)
if len(parts) > 1:
root = self.proj.get(parts[0])
if root:
rest = parts[1]
root.register_published_vars(rest, publish)
cont, root = self.get_object(pathname)
if has_interface(cont, IComponent):
if publish:
if pathname in self._publish_comps:
self._publish_comps[pathname] += 1
else:
self._publish_comps[pathname] = 1
else:
if pathname in self._publish_comps:
self._publish_comps[pathname] -= 1
if self._publish_comps[pathname] < 1:
del self._publish_comps[pathname]
def _start_log_msgs(self, topic):
""" Start sending log messages. """
# Need to lock access while we capture state.
logging._acquireLock()
try:
# Flush output.
for handler in logging.getLogger().handlers:
handler.flush()
# Grab previously logged messages.
log_path = os.path.join(self._log_directory, 'openmdao_log.txt')
with open(log_path, 'r') as inp:
line = True # Just to get things started.
while line:
lines = []
for i in range(100): # Process in chunks.
line = inp.readline()
if line:
lines.append(line)
else:
break
if lines:
publish('log_msgs',
dict(active=False, text=''.join(lines)))
# End of historical messages.
publish('log_msgs', dict(active=False, text=''))
# Add handler to get any new messages.
if self._log_handler is None:
self._log_handler = _LogHandler()
logging.getLogger().addHandler(self._log_handler)
except Exception:
print "Can't initiate logging:"
traceback.print_exc()
finally:
logging._releaseLock()
self._log_subscribers += 1
def _stop_log_msgs(self):
""" Stop sending log messages. """
self._log_subscribers -= 1
if self._log_subscribers <= 0:
if self._log_handler is not None:
logging.getLogger().removeHandler(self._log_handler)
self._log_handler = None
self._log_subscribers = 0
[docs] def is_macro(self, filename):
return filename.lstrip('/') == os.path.join(os.path.basename(self.proj.macrodir),
self.proj.macro)
[docs] def file_forces_reload(self, filename):
"""Returns True if the given file (assumed to be a file in the
project) has classes that have been instantiated in the current
process or if the file is a macro file. Note that this doesn't keep
track of removes/deletions, so if an instance was created earlier and
then deleted, it will still be reported.
"""
pdf = self.projdirfactory
if pdf:
if self.is_macro(filename):
return True
if filename.endswith('.py'):
filename = filename.lstrip('/')
filename = os.path.join(self.proj.path, filename)
info = pdf._files.get(filename)
if info and _match_insts(info.classes.keys()):
return True
return False
class _LogHandler(logging.StreamHandler):
""" Logging handler that publishes messages. """
def __init__(self):
# Python < 2.7 doesn't like super() here.
logging.StreamHandler.__init__(self, _LogStream())
# Formatting set to match format of file.
msg_fmt = '%(asctime)s %(levelname)s %(name)s: %(message)s'
date_fmt = '%b %d %H:%M:%S'
self.setFormatter(logging.Formatter(msg_fmt, date_fmt))
class _LogStream(object):
""" Provides stream interface to publisher. """
def write(self, msg):
publish('log_msgs', dict(active=True, text=msg))
def flush(self):
pass
