"""Unified command line interface for datawork pipelines."""
import argparse
from ..api.graph import compute_dag, extract_config, extract_inputs, visualize
[docs]def command_line(*outputs):
"""Given output Data objects, create a standard command line interface and execute it."""
# compute the graph
g = compute_dag(outputs)
# get all configurable elements
conf = extract_config(g)
# Get undetermined nodes that will be positional file arguments
inputs = extract_inputs(g)
desc = "This is an automatically generated command line interface.\n"
parser = argparse.ArgumentParser(
description=desc, formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
# Add positional arguments for required inputs
for _, i in inputs:
parser.add_argument(i.name)
# Add option arguments for saving outputs
for o in outputs:
pass
for toolname, toolconf in conf.items():
tg = parser.add_argument_group(f"Options for {toolname} tool")
for c in toolconf:
c.prefix = toolname + "-"
c.add_argument(tg)
rg = parser.add_argument_group(f"Runtime options")
rg.add_argument(
"--visualize",
default=None,
type=str,
help="Visualize computation graph and exit "
"(provide a filename for PNG output).",
)
rg.add_argument(
"--dryrun",
action="store_true",
help="Only display computation steps instead of actually computing anything",
)
args = parser.parse_args()
for toolname, toolconf in conf.items():
for c in toolconf:
carg = args.__dict__[toolname + "_" + c.name] # noqa
# TODO: actually set the value for this config
if args.visualize is not None:
visualize(g, args.visualize, outputs)
if args.dryrun:
pass
else:
# compute and write outputs if requested
pass
return args