# -*- coding: utf-8 -*-
"""
QA/QC CLI utility functions.
"""
import click
import logging
import numpy as np
import os
from rex.utilities.cli_dtypes import STR, STRLIST, INT
from rex.utilities.loggers import init_logger
from gaps.cli import as_click_command, CLICommandFromFunction
from reV.utilities import ModuleName
from reV.qa_qc.qa_qc import QaQc, QaQcModule
from reV.qa_qc.summary import (SummarizeH5, SummarizeSupplyCurve,
SupplyCurvePlot, ExclusionsMask)
from reV import __version__
logger = logging.getLogger(__name__)
[docs]def cli_qa_qc(modules, out_dir, max_workers=None):
"""Run QA/QC on reV outputs
``reV`` QA/QC performs quality assurance checks on ``reV`` output
data. Users can specify the type of QA/QC that should be applied
to each ``reV`` module.
Parameters
----------
modules : dict
Dictionary of modules to QA/QC. Keys should be the names of the
modules to QA/QC. The values are dictionaries that represent the
config for the respective QA/QC step. Allowed config keys for
QA/QC are the "property" attributes of
:class:`~reV.qa_qc.qa_qc.QaQcModule`.
out_dir : str
Path to output directory.
max_workers : int, optional
Max number of workers to run for QA/QA. If ``None``, uses all
CPU cores. By default, ``None``.
Raises
------
ValueError
If fpath is not an H5 or CSV file.
"""
for module, mcf in modules.items():
module_config = QaQcModule(module, mcf, out_dir)
qa_dir = out_dir
if module_config.sub_dir is not None:
qa_dir = os.path.join(out_dir, module_config.sub_dir)
if module.lower() == 'exclusions':
QaQc.exclusions_mask(module_config.fpath, qa_dir,
layers_dict=module_config.excl_dict,
min_area=module_config.min_area,
kernel=module_config.area_filter_kernel,
plot_type=module_config.plot_type,
cmap=module_config.cmap,
plot_step=module_config.plot_step)
elif module_config.fpath.endswith('.h5'):
QaQc.h5(module_config.fpath, qa_dir, dsets=module_config.dsets,
group=module_config.group,
process_size=module_config.process_size,
max_workers=max_workers,
plot_type=module_config.plot_type, cmap=module_config.cmap)
elif module_config.fpath.endswith('.csv'):
QaQc.supply_curve(module_config.fpath, qa_dir,
columns=module_config.columns,
lcoe=module_config.lcoe,
plot_type=module_config.plot_type,
cmap=module_config.cmap)
else:
msg = ("Cannot run QA/QC for {}: 'fpath' must be a '*.h5' "
"or '*.csv' reV output file, but {} was given!"
.format(module, module_config.fpath))
logger.error(msg)
raise ValueError(msg)
qa_qc_command = CLICommandFromFunction(cli_qa_qc, name=str(ModuleName.QA_QC),
split_keys=None)
main = as_click_command(qa_qc_command)
@click.group()
@click.version_option(version=__version__)
@click.option('-v', '--verbose', is_flag=True,
help='Flag to turn on debug logging. Default is not verbose.')
@click.pass_context
def qa_qc_extra(ctx, verbose):
"""Execute extra QA/QC utility"""
ctx.ensure_object(dict)
ctx.obj['VERBOSE'] = verbose
@qa_qc_extra.group(chain=True)
@click.option('--out_dir', '-o', type=click.Path(), required=True,
help="Directory path to save summary tables and plots too")
@click.option('--log_file', '-log', type=click.Path(), default=None,
show_default=True,
help='File to log to, by default None')
@click.option('-v', '--verbose', is_flag=True,
help='Flag to turn on debug logging.')
@click.pass_context
def summarize(ctx, out_dir, log_file, verbose):
"""
Summarize reV data
"""
ctx.obj['OUT_DIR'] = out_dir
if any([verbose, ctx.obj['VERBOSE']]):
log_level = 'DEBUG'
else:
log_level = 'INFO'
init_logger('reV', log_file=log_file, log_level=log_level)
@summarize.command()
@click.option('--h5_file', '-h5', type=click.Path(exists=True), required=True,
help='Path to .h5 file to summarize')
@click.option('--dsets', '-ds', type=STRLIST, default=None,
show_default=True,
help='Datasets to summarize, by default None')
@click.option('--group', '-grp', type=STR, default=None,
show_default=True,
help=('Group within h5_file to summarize datasets for, by '
'default None'))
@click.option('--process_size', '-ps', type=INT, default=None,
show_default=True,
help='Number of sites to process at a time, by default None')
@click.option('--max_workers', '-w', type=INT, default=None,
show_default=True,
help=('Number of workers to use when summarizing 2D datasets,'
' by default None'))
@click.pass_context
def h5(ctx, h5_file, dsets, group, process_size, max_workers):
"""
Summarize datasets in .h5 file
"""
SummarizeH5.run(h5_file, ctx.obj['OUT_DIR'], group=group, dsets=dsets,
process_size=process_size, max_workers=max_workers)
@summarize.command()
@click.option('--plot_type', '-plt', default='plotly',
type=click.Choice(['plot', 'plotly'], case_sensitive=False),
show_default=True,
help=(" plot_type of plot to create 'plot' or 'plotly', by "
"default 'plot'"))
@click.option('--cmap', '-cmap', type=str, default='viridis',
show_default=True,
help="Colormap name, by default 'viridis'")
@click.pass_context
def scatter_plots(ctx, plot_type, cmap):
"""
create scatter plots from h5 summary tables
"""
QaQc.create_scatter_plots(ctx.obj['OUT_DIR'], plot_type, cmap)
@summarize.command()
@click.option('--sc_table', '-sct', type=click.Path(exists=True),
required=True, help='Path to .csv containing Supply Curve table')
@click.option('--columns', '-cols', type=STRLIST, default=None,
show_default=True,
help=('Column(s) to summarize, if None summarize all numeric '
'columns, by default None'))
@click.pass_context
def supply_curve_table(ctx, sc_table, columns):
"""
Summarize Supply Curve Table
"""
ctx.obj['SC_TABLE'] = sc_table
SummarizeSupplyCurve.run(sc_table, ctx.obj['OUT_DIR'], columns=columns)
@summarize.command()
@click.option('--sc_table', '-sct', type=click.Path(exists=True), default=None,
show_default=True,
help=("Path to .csv containing Supply Curve table, can be "
"supplied in 'supply-curve-table'"))
@click.option('--plot_type', '-plt', default='plotly',
type=click.Choice(['plot', 'plotly'], case_sensitive=False),
show_default=True,
help=(" plot_type of plot to create 'plot' or 'plotly', by "
"default 'plot'"))
@click.option('--lcoe', '-lcoe', type=STR, default='mean_lcoe',
help="LCOE value to plot, by default 'mean_lcoe'")
@click.pass_context
def supply_curve_plot(ctx, sc_table, plot_type, lcoe):
"""
Plot Supply Curve (cumulative capacity vs LCOE)
"""
if sc_table is None:
sc_table = ctx.obj['SC_TABLE']
SupplyCurvePlot.plot(sc_table, ctx.obj['OUT_DIR'],
plot_type=plot_type, lcoe=lcoe)
@summarize.command()
@click.option('--excl_mask', '-mask', type=click.Path(exists=True),
required=True,
help='Path to .npy file containing final exclusions mask')
@click.option('--plot_type', '-plt', default='plotly',
type=click.Choice(['plot', 'plotly'], case_sensitive=False),
show_default=True,
help=(" plot_type of plot to create 'plot' or 'plotly', by "
"default 'plot'"))
@click.option('--cmap', '-cmap', type=str, default='viridis',
show_default=True,
help="Colormap name, by default 'viridis'")
@click.option('--plot_step', '-step', type=int, default=100,
show_default=True,
help="Step between points to plot")
@click.pass_context
def exclusions_mask(ctx, excl_mask, plot_type, cmap, plot_step):
"""
create heat map of exclusions mask
"""
excl_mask = np.load(excl_mask)
ExclusionsMask.plot(excl_mask, ctx.obj['OUT_DIR'],
plot_type=plot_type, cmap=cmap,
plot_step=plot_step)
if __name__ == '__main__':
try:
main(obj={})
except Exception:
logger.exception('Error running reV QA/QC CLI.')
raise