"""
Utility functions related to the reporting system which generates reports by default for all runs.
"""
import os
import inspect
from openmdao.core.constants import _UNDEFINED
from openmdao.utils.hooks import _register_hook, _unregister_hook
from openmdao.utils.om_warnings import issue_warning
from openmdao.utils.file_utils import _iter_entry_points, _find_openmdao_output_dirs
from openmdao.utils.webview import webview
from openmdao.utils.general_utils import env_truthy, is_truthy
from openmdao.visualization.tables.table_builder import generate_table
_reports_registry = {}
_default_reports = ['scaling', 'total_coloring', 'n2', 'optimizer', 'inputs']
_active_reports = set() # these reports will actually run (assuming their hook funcs are triggered)
_plugins_loaded = False # use this to ensure plugins only loaded once
[docs]class Report(object):
r"""
A class to manage the registration of hooks associated with a particular named report.
Parameters
----------
name : str
The name of the corresponding report.
description : str
The description displayed when reports are listed using the `list_reports` command.
Attributes
----------
name : str
The name of the corresponding report.
description : str
The description displayed when reports are listed using the `list_reports` command.
hooks : list
List of (args, kwargs) to be passed to the `register_hook` function when the report
corresponding to this instance is activated.
"""
[docs] def __init__(self, name, description=''):
"""
Initialize attributes.
"""
self.name = name
self.description = description
self.hooks = []
[docs] def register_hook_args(self, *args, **kwargs):
r"""
Store positional and named args to be passed to the `register_hook` function.
This will only happen if the report corresponding to this instance is activated.
Parameters
----------
*args : list
The positional args to be passed to `register_hook`.
**kwargs : dict
The named args to be passed to `register_hook`.
"""
self.hooks.append((args, kwargs))
[docs] def register_hooks(self, instance):
"""
Register the hook(s) associated with the report corresponding with this instance.
Parameters
----------
instance : object
The instance where hooks may be registered.
"""
if hasattr(instance, '_has_active_report') and not instance._has_active_report(self.name):
return
for hook_args, hook_kwargs in self.hooks:
# always register hook with a specific instance id
kw = hook_kwargs.copy()
kw['inst_id'] = None if instance is None else instance._get_inst_id()
_register_hook(*hook_args, **kw)
[docs] def unregister_hooks(self, instance=None):
"""
Unregister hooks associated with this report.
Parameters
----------
instance : object or None
If not None, only unregister reports for this instance.
"""
keep = {'fname', 'class_name', 'inst_id', 'pre', 'post'}
for args, kw in self.hooks:
kwargs = {k: v for k, v in kw.items() if k in keep}
if instance is not None:
kwargs['inst_id'] = instance
_unregister_hook(*args, **kwargs)
def __getattr__(self, name):
"""
Return the named attribute from our stored hook args.
Parameters
----------
name : str
The name of the attribute.
"""
for args, kwargs in self.hooks:
if name in kwargs:
return kwargs[name]
elif name == 'fname' and len(args) > 0:
return args[0]
elif name == 'class_name' and len(args) > 1:
return args[1]
else:
raise AttributeError(f"Attribute '{name}' not found.")
[docs]def reports_active():
"""
Return True if reports are active globally.
Returns
-------
bool
Return True if reports are active.
"""
return not env_truthy('TESTFLO_RUNNING')
[docs]def register_report(name, func, desc, class_name, method, pre_or_post, inst_id=None, predicate=None,
**kwargs):
"""
Register a report with the reporting system.
Parameters
----------
name : str
Name of report. Report names must be unique across all reports.
func : function
A function to do the reporting. Expects the first argument to be an instance of class_name.
desc : str
A description of the report.
class_name : str
The name of the class owning the method where the report will be run.
method : str
In which method of class_name should this be run.
pre_or_post : str
Valid values are 'pre' and 'post'. Indicates when to run the report in the method.
inst_id : str or None
Either the instance ID of an OpenMDAO object (e.g. Problem, Driver) or None.
If None, then this report will be run for all objects of type class_name.
predicate : function or None
If not None, this function will be called to determine if the report's hook function
should run. The predicate function should take the class instance as its only argument and
return True if the report should run. Note that returning False does not disable the hook,
it just prevents the hook from running at that time.
**kwargs : dict
Keyword args passed to the report function.
Returns
-------
Report
The registered report object.
"""
global _reports_registry
if name in _reports_registry:
raise ValueError(f"A report with the name {name} was already registered.")
if pre_or_post not in ('pre', 'post'):
raise ValueError("The argument 'pre_or_post' can only have values of 'pre' or 'post', "
f"but {pre_or_post} was given")
if pre_or_post == 'pre' and method == 'setup':
raise ValueError('Reports cannot be registered to execute pre-setup.')
_reports_registry[name] = report = Report(name, desc)
pre = func if pre_or_post == 'pre' else None
post = func if pre_or_post == 'post' else None
report.register_hook_args(fname=method, class_name=class_name, inst_id=inst_id, pre=pre,
post=post, ncalls=1, predicate=predicate, **kwargs)
return report
[docs]def unregister_report(name):
"""
Unregister a report with the reporting system.
Parameters
----------
name : str
Name of report. Report names must be unique across all reports.
"""
global _reports_registry
del _reports_registry[name]
[docs]def register_report_hook(name, fname, class_name, inst_id=None, pre=None, post=None, description='',
**kwargs):
"""
Register a hook with a specific report name in the reporting system.
By calling this multiple times, multiple hooks can be registered for the same report. This
is sometimes necessary to get the correct behavior when setup errors occur prior to report
creation.
Parameters
----------
name : str
Name of report. Report names must be unique across all reports.
fname : str
The name of the function where the pre and/or post hook will be applied.
class_name : str
The name of the class owning the method where the hook will be applied.
inst_id : str or None
The name of the instance owning the method where the hook will be applied.
pre : function (None)
If not None, this hook will run before the function named by fname runs.
post : function (None)
If not None, this hook will run after the function named by fname runs.
description : str
A description of the report.
**kwargs : dict of keyword arguments
Keyword arguments that will be passed to the hook function.
"""
global _reports_registry
if name not in _reports_registry:
_reports_registry[name] = report = Report(name, description)
else:
report = _reports_registry[name]
if description is not None:
report.description = description
report.register_hook_args(fname, class_name, inst_id=inst_id, pre=pre, post=post, ncalls=1,
**kwargs)
[docs]def activate_report(name, instance=None):
"""
Activate a report that has been registered with the reporting system.
Parameters
----------
name : str
Name of report. Report names must be unique across all reports.
instance : object or None
If not None, report will be activated only for this instance, assuming the instance
in the reports_registry is either None or matching.
"""
global _reports_registry, _active_reports
if name not in _reports_registry:
issue_warning(f"No report with the name '{name}' is registered.")
return
if not reports_active():
return
inst_id = None if instance is None else instance._get_inst_id()
report = _reports_registry[name]
for _, hook_kwargs in report.hooks:
_inst_id = hook_kwargs['inst_id']
break
# handle case where report was registered for a specific inst_id
if _inst_id is not None:
if inst_id is None:
inst_id = _inst_id
elif inst_id != _inst_id: # registered inst_id doesn't match current instance
return
if (name, inst_id) in _active_reports:
issue_warning(f"A report with the name '{name}' for instance '{inst_id}' is already "
"active.")
return
report.register_hooks(instance)
_active_reports.add((name, inst_id))
[docs]def activate_reports(reports, instance):
"""
Activate any matching reports.
Parameters
----------
reports : list of str
List of report names that should be active. These names come down from the Problem
and some may refer to Problem reports while others may refer to Driver reports.
instance : object
The reports will be activated for this instance.
"""
cnames = [c.__name__ for c in inspect.getmro(instance.__class__)][:-1]
for name in reports:
try:
report = _reports_registry[name]
if report.class_name in cnames: # report corresponds to our class
activate_report(name, instance)
except KeyError:
issue_warning(f"Report with name '{name}' not found in reports registry.")
continue
[docs]def list_reports(default=False, outfile=None, max_width=80):
"""
Write table of information about reports currently registered in the reporting system.
Parameters
----------
default : bool
If True, list only the default reports.
outfile : str or None
Where to send report info. None will result in output to stdout.
max_width : int
Maximum width of the table. Defaults to 80.
"""
global _reports_registry
# if we haven't created any Problem instances, the registry could still be uninitialized.
# if it *has* already been initialized, this call will do nothing.
_load_report_plugins()
headers = ['name', 'description', 'class name', 'method', 'pre or post']
if default:
reg = {r: _reports_registry[r] for r in _default_reports}
else:
reg = _reports_registry
rows = []
# Now for the values
for name, report in sorted(reg.items()):
rows.append([])
rows[-1].append(name)
for attr in ('description', 'class_name', 'fname'):
rows[-1].append(getattr(report, attr))
pre = report.pre
rows[-1].append('pre' if pre else 'post')
generate_table(rows, tablefmt='box_grid', headers=headers, max_width=max_width).display(outfile)
def _list_reports_setup_parser(parser):
"""
Set up the openmdao subparser for the 'openmdao list_reports' command.
Parameters
----------
parser : argparse subparser
The parser we're adding options to.
"""
parser.add_argument('-d', '--default', action='store_true', dest='dflt',
help="List only the default reports.")
parser.add_argument('-o', '--outfile', action='store', dest='outfile',
help='Send list of reports to this file.')
def _list_reports_cmd(options, user_args):
"""
Return the post_setup hook function for 'openmdao list_reports'.
Parameters
----------
options : argparse Namespace
Command line options.
user_args : list of str
Args to be passed to the user script.
"""
list_reports(outfile=options.outfile, default=options.dflt)
[docs]def view_reports(probnames=None, level=2):
"""
Pop up a browser to view specified reports.
Parameters
----------
probnames : str, iter of str, or None
If not None, view only reports for the specified Problem(s), else view all reports.
level : int
Expand the reports directory tree to this level. Default is 2.
"""
tdir = os.getcwd()
om_out_dirs = set(str(p) for p in _find_openmdao_output_dirs(tdir, recurse=True))
if probnames:
matches = set()
if isinstance(probnames, str):
probnames = (probnames,)
for probname in probnames:
subdir = os.path.join(tdir, f'{probname}_out')
if subdir in om_out_dirs:
match = subdir
else:
pdir = os.path.join(tdir, probname)
if pdir in om_out_dirs:
match = pdir
else:
print(f"Can't find problem dir '{subdir}'.")
continue
matches.add(match)
om_out_dirs = matches
if om_out_dirs:
gen_reports_index_file(tdir, level, om_out_dirs)
webview(os.path.join(tdir, 'index.html'))
else:
print("No matching report dirs found.")
return
def _view_reports_setup_parser(parser):
"""
Set up the openmdao subparser for the 'openmdao view_reports' command.
Parameters
----------
parser : argparse subparser
The parser we're adding options to.
"""
parser.add_argument('problem', metavar='problem', nargs='*',
help='View reports only for the specified Problem(s).')
parser.add_argument('-l', '--level', action='store', dest='level', type=int, default=1,
help='Expand the reports directory tree to this level. Default is 1.')
def _view_reports_cmd(options, user_args):
"""
View completed reports via the 'openmdao view_reports' command.
Parameters
----------
options : argparse Namespace
Command line options.
user_args : list of str
Args to be passed to the user script.
"""
view_reports(options.problem, level=options.level)
[docs]def set_reports_dir(reports_dir_path):
"""
Set the path to the top level reports directory. Defaults to './reports'.
Parameters
----------
reports_dir_path : str
Path to the top level reports directory.
"""
raise RuntimeError('The set_reports_dir function has been removed. Reports are now stored in '
'the reports subdirectory under the enclosing problem\'s output directory')
# -----------------------------------------
# TODO: remove these once dymos has been updated to current reports system API
def _run_n2_report(x):
pass
def _setup_default_reports():
pass
report_function = _run_n2_report
run_n2_report = _run_n2_report
setup_default_reports = _setup_default_reports
set_default_reports_dir = set_reports_dir
# -----------------------------------------
def _reset_reports_dir():
"""
Reset the path to the top level reports directory from the environment or to './reports'.
This is used during testing, where environment variables are sometimes modified during
the test.
"""
raise RuntimeError('The _reset_reports_dir function in openmdao.api has been removed. '
'Reports are now placed in the reports sub-directory of the problem.')
[docs]def get_reports_dir():
"""
Get the path to the top level reports directory. Defaults to './reports'.
Returns
-------
str
Path to the top level reports directory.
"""
raise RuntimeError('The get_reports_dir function in openmdao.api has '
'been replaced by the get_reports_dir method on Problem and System.')
def _reports2list(reports, defaults):
"""
Return a list of reports based on the value of the reports var and current default report list.
Parameters
----------
reports : str, list, or _UNDEFINED
Variable indicating which reports should be active based on the current 'defaults' list.
defaults : list
List of current default reports. This could be either the global report defaults or
a newer list of defaults based on previous processing.
Returns
-------
list
The list of reports that should be active.
"""
if reports in [True, _UNDEFINED]:
return defaults
if not reports:
return []
if isinstance(reports, str):
low = reports.lower()
if low == 'all':
return list(_reports_registry) # activate all registered reports
if is_truthy(low):
if ',' in low:
return [s.strip() for s in reports.split(',') if s.strip()]
elif reports in _reports_registry:
return [reports]
else:
return defaults
else:
return []
return list(reports)
[docs]def get_reports_to_activate(reports=_UNDEFINED):
"""
Get the list of names of reports that should be activated.
Parameters
----------
reports : str, list of str, _UNDEFINED, or None
Indicates which report(s) are to be activated.
Returns
-------
list of str
List of report names.
"""
if not reports or not reports_active():
return []
reps_env = os.environ.get('OPENMDAO_REPORTS', 'true')
env_list = _reports2list(reps_env, _default_reports[:])
return _reports2list(reports, env_list)
[docs]def clear_report_registry():
"""
Clear all of the reports from the registry.
"""
global _reports_registry
_reports_registry = {}
[docs]def clear_reports(instance=None):
"""
Clear all of the currently active reports.
Parameters
----------
instance : object or None
If not None, only clear reports for this instance.
"""
global _active_reports, _reports_registry
inst_id = None if instance is None else instance._get_inst_id()
to_remove = set()
# need to remove the hooks
for name, active_inst_id in _active_reports:
if instance is None:
inst_id = active_inst_id
elif inst_id != active_inst_id:
continue
if name in _reports_registry:
_reports_registry[name].unregister_hooks(inst_id)
else:
issue_warning(f"No report with the name '{name}' is registered.")
to_remove.add((name, active_inst_id))
_active_reports -= to_remove
def _load_report_plugins():
"""
Load all 'openmdao_report' entry points and run them to register their plugins.
"""
global _plugins_loaded
if _plugins_loaded: # make sure we only run this once
return
_plugins_loaded = True
for ep in _iter_entry_points('openmdao_report'):
register_func = ep.load()
register_func() # this runs the function that calls register_report
def _get_reports_dir_files_html(reports_dir, explevel, level):
lines = []
files = os.listdir(reports_dir)
if files:
lines.append('<ul>')
for f in files:
if f.endswith('.html') and f != 'index.html':
path = os.path.join(reports_dir, f)
lines.append(f'<li> <a href="file:///{path}">{f}</a> </li>')
lines.append('</ul>')
return lines
def _add_dir_to_tree(startdir, explevel, level):
"""
Create nested lists of directories with links to files.
Parameters
----------
startdir : str
Starting directory.
explevel : int
Expand the tree to this level.
level : int
The current level of the tree.
Returns
-------
list of str
List of lines in the final html.
"""
repdir = os.path.join(startdir, 'reports')
if not os.path.isdir(repdir):
return []
flines = _get_reports_dir_files_html(repdir, explevel, level)
subdirs = [str(d) for d in _find_openmdao_output_dirs(startdir, recurse=False)
if str(d) != startdir]
sublines = []
for sub in subdirs:
sublines.extend(_add_dir_to_tree(sub, explevel, level + 1))
lines = []
if flines or sublines:
op = 'open' if level < explevel else ''
lines.append(f'<li><details {op}><summary>{os.path.basename(startdir)}</summary>')
lines.extend(flines)
lines.extend(sublines)
lines.append('</details></li>')
return lines
def _get_pre_body():
return """
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="utf-8">
<style>
.tree {
--spacing : 1.5rem;
--radius : 8px;
}
.tree li {
display : block;
position : relative;
padding-left : calc(2 * var(--spacing) - var(--radius) - 2px);
}
.tree ul {
margin-left : calc(var(--radius) - var(--spacing));
padding-left : 0;
}
.tree ul li {
border-left : 2px solid #ddd;
}
.tree ul li:last-child {
border-color : transparent;
}
.tree ul li::before {
content : '';
display : block;
position : absolute;
top : calc(var(--spacing) / -2);
left : -2px;
width : calc(var(--spacing) + 2px);
height : calc(var(--spacing) + 1px);
border : solid #ddd;
border-width : 0 0 2px 2px;
}
.tree summary {
display : block;
cursor : pointer;
}
.tree summary::marker,
.tree summary::-webkit-details-marker {
display : none;
}
.tree summary:focus {
outline : none;
}
.tree summary:focus-visible {
outline : 1px dotted #000;
}
.tree li::after,
.tree summary::before {
content : '';
display : block;
position : absolute;
top : calc(var(--spacing) / 2 - var(--radius));
left : calc(var(--spacing) - var(--radius) - 1px);
width : calc(2 * var(--radius));
height : calc(2 * var(--radius));
border-radius : 50%;
background : #ddd;
}
.tree summary::before {
content : '+';
z-index : 1;
background : #696;
color : #fff;
line-height : calc(2 * var(--radius) - 2px);
text-align : center;
}
.tree details[open] > summary::before {
content : '−';
}
</style>
<script>
</script>
</head>
<body>
"""
[docs]def gen_reports_index_file(start_dir, level, to_match):
"""
Generate an index.html file that will have links to all of the reports.
Parameters
----------
start_dir : str
The top directory containing the reports.
level : int
Expand the reports directory tree to this level.
to_match : set
Set of subdirectory names to show.
"""
start_dir = os.path.abspath(start_dir)
# tree view courtesy of: https://iamkate.com/code/tree-views/
parts = [_get_pre_body()]
lines = ['<ul class="tree">']
for outdir in sorted(to_match):
lines.extend(_add_dir_to_tree(outdir, explevel=level, level=0))
parts.append('\n'.join(lines))
parts.append('</ul></body>\n</html>')
with open(os.path.join(start_dir, 'index.html'), 'w') as f:
f.write('\n'.join(parts))
