experiments: migrate dvc repro -e to dvc exp run (#4677)

* experiments: migrate from `dvc repro -e` to `dvc exp run`

* tests: use experiments.run instead of reproduce

* update `repro` and `exp run` commands
This commit is contained in:
Peter Rowlands (변기호)
2020-10-09 00:26:31 +03:00
committed by GitHub
parent 3ceef7dbf1
commit c16d3fa43d
9 changed files with 158 additions and 318 deletions
+8 -125
View File
@@ -7,11 +7,10 @@ from datetime import date
from itertools import groupby
from typing import Iterable, Optional
from dvc.command import completion
from dvc.command.base import CmdBase, append_doc_link, fix_subparsers
from dvc.command.metrics import DEFAULT_PRECISION, _show_metrics
from dvc.command.status import CmdDataStatus
from dvc.dvcfile import PIPELINE_FILE
from dvc.command.metrics import DEFAULT_PRECISION
from dvc.command.repro import CmdRepro
from dvc.command.repro import add_arguments as add_repro_arguments
from dvc.exceptions import DvcException, InvalidArgumentError
from dvc.utils.flatten import flatten
@@ -388,7 +387,7 @@ class CmdExperimentsDiff(CmdBase):
return 0
class CmdExperimentsRun(CmdBase):
class CmdExperimentsRun(CmdRepro):
def run(self):
if not self.repo.experiments:
return 0
@@ -405,20 +404,8 @@ class CmdExperimentsRun(CmdBase):
ret = 0
for target in self.args.targets:
try:
stages = self.repo.reproduce(
self.repo.experiments.run(
target,
single_item=self.args.single_item,
force=self.args.force,
dry=self.args.dry,
interactive=self.args.interactive,
pipeline=self.args.pipeline,
all_pipelines=self.args.all_pipelines,
run_cache=not self.args.no_run_cache,
no_commit=self.args.no_commit,
downstream=self.args.downstream,
recursive=self.args.recursive,
force_downstream=self.args.force_downstream,
experiment=True,
queue=self.args.queue,
run_all=self.args.run_all,
jobs=self.args.jobs,
@@ -428,15 +415,8 @@ class CmdExperimentsRun(CmdBase):
or self.args.checkpoint_continue is not None
),
checkpoint_continue=self.args.checkpoint_continue,
**self._repro_kwargs,
)
if len(stages) == 0:
logger.info(CmdDataStatus.UP_TO_DATE_MSG)
if self.args.metrics:
metrics = self.repo.metrics.show()
logger.info(_show_metrics(metrics))
except DvcException:
logger.exception("")
ret = 1
@@ -636,105 +616,8 @@ def add_parser(subparsers, parent_parser):
help=EXPERIMENTS_RUN_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
experiments_run_parser.add_argument(
"targets",
nargs="*",
help=f"Stages to reproduce. '{PIPELINE_FILE}' by default.",
).complete = completion.DVC_FILE
experiments_run_parser.add_argument(
"-f",
"--force",
action="store_true",
default=False,
help="Reproduce even if dependencies were not changed.",
)
experiments_run_parser.add_argument(
"-s",
"--single-item",
action="store_true",
default=False,
help="Reproduce only single data item without recursive dependencies "
"check.",
)
experiments_run_parser.add_argument(
"-c",
"--cwd",
default=os.path.curdir,
help="Directory within your repo to reproduce from. Note: deprecated "
"by `dvc --cd <path>`.",
metavar="<path>",
)
experiments_run_parser.add_argument(
"-m",
"--metrics",
action="store_true",
default=False,
help="Show metrics after reproduction.",
)
experiments_run_parser.add_argument(
"--dry",
action="store_true",
default=False,
help="Only print the commands that would be executed without "
"actually executing.",
)
experiments_run_parser.add_argument(
"-i",
"--interactive",
action="store_true",
default=False,
help="Ask for confirmation before reproducing each stage.",
)
experiments_run_parser.add_argument(
"-p",
"--pipeline",
action="store_true",
default=False,
help="Reproduce the whole pipeline that the specified stage file "
"belongs to.",
)
experiments_run_parser.add_argument(
"-P",
"--all-pipelines",
action="store_true",
default=False,
help="Reproduce all pipelines in the repo.",
)
experiments_run_parser.add_argument(
"-R",
"--recursive",
action="store_true",
default=False,
help="Reproduce all stages in the specified directory.",
)
experiments_run_parser.add_argument(
"--no-run-cache",
action="store_true",
default=False,
help=(
"Execute stage commands even if they have already been run with "
"the same command/dependencies/outputs/etc before."
),
)
experiments_run_parser.add_argument(
"--force-downstream",
action="store_true",
default=False,
help="Reproduce all descendants of a changed stage even if their "
"direct dependencies didn't change.",
)
experiments_run_parser.add_argument(
"--no-commit",
action="store_true",
default=False,
help="Don't put files/directories into cache.",
)
experiments_run_parser.add_argument(
"--downstream",
action="store_true",
default=False,
help="Start from the specified stages when reproducing pipelines.",
)
# inherit arguments from `dvc repro`
add_repro_arguments(experiments_run_parser)
experiments_run_parser.add_argument(
"--params",
action="append",
+33 -57
View File
@@ -26,26 +26,7 @@ class CmdRepro(CmdBase):
ret = 0
for target in self.args.targets:
try:
stages = self.repo.reproduce(
target,
single_item=self.args.single_item,
force=self.args.force,
dry=self.args.dry,
interactive=self.args.interactive,
pipeline=self.args.pipeline,
all_pipelines=self.args.all_pipelines,
run_cache=not self.args.no_run_cache,
no_commit=self.args.no_commit,
downstream=self.args.downstream,
recursive=self.args.recursive,
force_downstream=self.args.force_downstream,
experiment=self.args.experiment,
queue=self.args.queue,
run_all=self.args.run_all,
jobs=self.args.jobs,
params=self.args.params,
pull=self.args.pull,
)
stages = self.repo.reproduce(target, **self._repro_kwargs)
if len(stages) == 0:
logger.info(CmdDataStatus.UP_TO_DATE_MSG)
@@ -62,18 +43,25 @@ class CmdRepro(CmdBase):
os.chdir(saved_dir)
return ret
@property
def _repro_kwargs(self):
return {
"single_item": self.args.single_item,
"force": self.args.force,
"dry": self.args.dry,
"interactive": self.args.interactive,
"pipeline": self.args.pipeline,
"all_pipelines": self.args.all_pipelines,
"run_cache": not self.args.no_run_cache,
"no_commit": self.args.no_commit,
"downstream": self.args.downstream,
"recursive": self.args.recursive,
"force_downstream": self.args.force_downstream,
"pull": self.args.pull,
}
def add_parser(subparsers, parent_parser):
REPRO_HELP = (
"Reproduce complete or partial pipelines by executing their stages."
)
repro_parser = subparsers.add_parser(
"repro",
parents=[parent_parser],
description=append_doc_link(REPRO_HELP, "repro"),
help=REPRO_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
def add_arguments(repro_parser):
repro_parser.add_argument(
"targets",
nargs="*",
@@ -173,32 +161,6 @@ def add_parser(subparsers, parent_parser):
default=False,
help="Start from the specified stages when reproducing pipelines.",
)
repro_parser.add_argument(
"-e",
"--experiment",
action="store_true",
default=False,
help=argparse.SUPPRESS,
)
repro_parser.add_argument(
"--params",
action="append",
default=[],
help=argparse.SUPPRESS,
metavar="[<filename>:]<params_list>",
)
repro_parser.add_argument(
"--queue", action="store_true", default=False, help=argparse.SUPPRESS
)
repro_parser.add_argument(
"--run-all",
action="store_true",
default=False,
help=argparse.SUPPRESS,
)
repro_parser.add_argument(
"-j", "--jobs", type=int, help=argparse.SUPPRESS, metavar="<number>"
)
repro_parser.add_argument(
"--pull",
action="store_true",
@@ -208,4 +170,18 @@ def add_parser(subparsers, parent_parser):
"from the run-cache."
),
)
def add_parser(subparsers, parent_parser):
REPRO_HELP = (
"Reproduce complete or partial pipelines by executing their stages."
)
repro_parser = subparsers.add_parser(
"repro",
parents=[parent_parser],
description=append_doc_link(REPRO_HELP, "repro"),
help=REPRO_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
add_arguments(repro_parser)
repro_parser.set_defaults(func=CmdRepro)
+5
View File
@@ -779,3 +779,8 @@ class Experiments:
from dvc.repo.experiments.show import show
return show(self.repo, *args, **kwargs)
def run(self, *args, **kwargs):
from dvc.repo.experiments.run import run
return run(self.repo, *args, **kwargs)
+67
View File
@@ -0,0 +1,67 @@
import logging
from typing import Iterable, Optional
from dvc.exceptions import InvalidArgumentError
from dvc.repo import locked
from dvc.repo.experiments import UnchangedExperimentError
logger = logging.getLogger(__name__)
def _parse_params(path_params: Iterable):
from ruamel.yaml import YAMLError
from dvc.dependency.param import ParamsDependency
from dvc.utils.flatten import unflatten
from dvc.utils.serialize import loads_yaml
ret = {}
for path_param in path_params:
path, _, params_str = path_param.rpartition(":")
# remove empty strings from params, on condition such as `-p "file1:"`
params = {}
for param_str in filter(bool, params_str.split(",")):
try:
# interpret value strings using YAML rules
key, value = param_str.split("=")
params[key] = loads_yaml(value)
except (ValueError, YAMLError):
raise InvalidArgumentError(
f"Invalid param/value pair '{param_str}'"
)
if not path:
path = ParamsDependency.DEFAULT_PARAMS_FILE
ret[path] = unflatten(params)
return ret
@locked
def run(
repo,
target: Optional[str] = None,
params: Optional[Iterable] = None,
run_all: Optional[bool] = False,
jobs: Optional[int] = 1,
**kwargs,
) -> dict:
"""Reproduce the specified target as an experiment.
Accepts the same additional kwargs as Repo.reproduce.
Returns a dict mapping new experiment SHAs to the results
of `repro` for that experiment.
"""
if run_all:
return repo.experiments.reproduce_queued(jobs=jobs)
if params:
params = _parse_params(params)
else:
params = []
try:
return repo.experiments.reproduce_one(
target=target, params=params, **kwargs
)
except UnchangedExperimentError:
# If experiment contains no changes, just run regular repro
return {None: repo.reproduce(target=target, **kwargs)}
-60
View File
@@ -2,7 +2,6 @@ import logging
from functools import partial
from dvc.exceptions import InvalidArgumentError, ReproductionError
from dvc.repo.experiments import UnchangedExperimentError
from dvc.repo.scm_context import scm_context
from . import locked
@@ -83,32 +82,6 @@ def reproduce(
"Neither `target` nor `--all-pipelines` are specified."
)
experiment = kwargs.pop("experiment", False)
params = _parse_params(kwargs.pop("params", []))
queue = kwargs.pop("queue", False)
run_all = kwargs.pop("run_all", False)
jobs = kwargs.pop("jobs", 1)
checkpoint = kwargs.pop("checkpoint", False)
checkpoint_continue = kwargs.pop("checkpoint_continue", None)
if (experiment or run_all) and self.experiments:
try:
return _reproduce_experiments(
self,
target=target,
recursive=recursive,
all_pipelines=all_pipelines,
params=params,
queue=queue,
run_all=run_all,
jobs=jobs,
checkpoint=checkpoint,
checkpoint_continue=checkpoint_continue,
**kwargs,
)
except UnchangedExperimentError:
# If experiment contains no changes, just run regular repro
pass
interactive = kwargs.get("interactive", False)
if not interactive:
kwargs["interactive"] = self.config["core"].get("interactive", False)
@@ -135,39 +108,6 @@ def reproduce(
return _reproduce_stages(active_graph, targets, **kwargs)
def _parse_params(path_params):
from ruamel.yaml import YAMLError
from dvc.dependency.param import ParamsDependency
from dvc.utils.flatten import unflatten
from dvc.utils.serialize import loads_yaml
ret = {}
for path_param in path_params:
path, _, params_str = path_param.rpartition(":")
# remove empty strings from params, on condition such as `-p "file1:"`
params = {}
for param_str in filter(bool, params_str.split(",")):
try:
# interpret value strings using YAML rules
key, value = param_str.split("=")
params[key] = loads_yaml(value)
except (ValueError, YAMLError):
raise InvalidArgumentError(
f"Invalid param/value pair '{param_str}'"
)
if not path:
path = ParamsDependency.DEFAULT_PARAMS_FILE
ret[path] = unflatten(params)
return ret
def _reproduce_experiments(repo, run_all=False, jobs=1, **kwargs):
if run_all:
return repo.experiments.reproduce_queued(jobs=jobs)
return repo.experiments.reproduce_one(**kwargs)
def _reproduce_stages(
G, stages, downstream=False, single_item=False, on_unchanged=None, **kwargs
):
+26 -40
View File
@@ -58,7 +58,7 @@ def test_new_simple(tmp_dir, scm, dvc, mocker):
tmp_dir.gen("params.yaml", "foo: 2")
new_mock = mocker.spy(dvc.experiments, "new")
dvc.reproduce(stage.addressing, experiment=True)
dvc.experiments.run(stage.addressing)
new_mock.assert_called_once()
assert (
@@ -80,13 +80,13 @@ def test_update_with_pull(tmp_dir, scm, dvc, mocker):
expected_revs = [scm.get_rev()]
tmp_dir.gen("params.yaml", "foo: 2")
dvc.reproduce(stage.addressing, experiment=True)
dvc.experiments.run(stage.addressing)
scm.add(["dvc.yaml", "dvc.lock", "params.yaml", "metrics.yaml"])
scm.commit("promote experiment")
expected_revs.append(scm.get_rev())
tmp_dir.gen("params.yaml", "foo: 3")
dvc.reproduce(stage.addressing, experiment=True)
dvc.experiments.run(stage.addressing)
exp_scm = dvc.experiments.scm
for rev in expected_revs:
@@ -105,11 +105,11 @@ def test_checkout(tmp_dir, scm, dvc):
scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"])
scm.commit("init")
dvc.reproduce(stage.addressing, experiment=True, params=["foo=2"])
exp_a = dvc.experiments.scm.get_rev()
results = dvc.experiments.run(stage.addressing, params=["foo=2"])
exp_a = first(results)
dvc.reproduce(stage.addressing, experiment=True, params=["foo=3"])
exp_b = dvc.experiments.scm.get_rev()
results = dvc.experiments.run(stage.addressing, params=["foo=3"])
exp_b = first(results)
dvc.experiments.checkout(exp_a)
assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 2"
@@ -134,13 +134,10 @@ def test_get_baseline(tmp_dir, scm, dvc):
expected = scm.get_rev()
assert dvc.experiments.get_baseline(expected) is None
dvc.reproduce(stage.addressing, experiment=True, params=["foo=2"])
rev = dvc.experiments.scm.get_rev()
assert dvc.experiments.get_baseline(rev) == expected
results = dvc.experiments.run(stage.addressing, params=["foo=2"])
assert dvc.experiments.get_baseline(first(results)) == expected
dvc.reproduce(
stage.addressing, experiment=True, params=["foo=3"], queue=True
)
dvc.experiments.run(stage.addressing, params=["foo=3"], queue=True)
assert dvc.experiments.get_baseline("stash@{0}") == expected
@@ -156,10 +153,8 @@ def test_update_py_params(tmp_dir, scm, dvc):
scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.py", "metrics.py"])
scm.commit("init")
dvc.reproduce(
stage.addressing, experiment=True, params=["params.py:INT=2"]
)
exp_a = dvc.experiments.scm.get_rev()
results = dvc.experiments.run(stage.addressing, params=["params.py:INT=2"])
exp_a = first(results)
dvc.experiments.checkout(exp_a)
assert (tmp_dir / "params.py").read_text().strip() == "INT = 2"
@@ -180,12 +175,11 @@ def test_update_py_params(tmp_dir, scm, dvc):
scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.py", "metrics.py"])
scm.commit("init")
dvc.reproduce(
results = dvc.experiments.run(
stage.addressing,
experiment=True,
params=["params.py:FLOAT=0.1,Train.seed=2121,Klass.a=222"],
)
exp_a = dvc.experiments.scm.get_rev()
exp_a = first(results)
result = (
"INT = 1\nFLOAT = 0.1\nDICT = {'a': 1}\n\n"
@@ -208,9 +202,7 @@ def test_update_py_params(tmp_dir, scm, dvc):
scm.commit("init")
with pytest.raises(PythonFileCorruptedError):
dvc.reproduce(
stage.addressing, experiment=True, params=["params.py:INT=2a"]
)
dvc.experiments.run(stage.addressing, params=["params.py:INT=2a"])
def test_extend_branch(tmp_dir, scm, dvc):
@@ -225,18 +217,17 @@ def test_extend_branch(tmp_dir, scm, dvc):
scm.add(["dvc.yaml", "dvc.lock", "copy.py", "params.yaml", "metrics.yaml"])
scm.commit("init")
dvc.reproduce(stage.addressing, experiment=True, params=["foo=2"])
exp_a = dvc.experiments.scm.get_rev()
results = dvc.experiments.run(stage.addressing, params=["foo=2"])
exp_a = first(results)
exp_branch = dvc.experiments._get_branch_containing(exp_a)
dvc.reproduce(
results = dvc.experiments.run(
stage.addressing,
experiment=True,
params=["foo=3"],
branch=exp_branch,
apply_workspace=False,
)
exp_b = dvc.experiments.scm.get_rev()
exp_b = first(results)
assert dvc.experiments._get_branch_containing(exp_b) == exp_branch
@@ -269,9 +260,9 @@ def test_detached_parent(tmp_dir, scm, dvc, mocker):
scm.checkout(detached_rev)
assert scm.repo.head.is_detached
dvc.reproduce(stage.addressing, experiment=True, params=["foo=3"])
results = dvc.experiments.run(stage.addressing, params=["foo=3"])
exp_rev = dvc.experiments.scm.get_rev()
exp_rev = first(results)
assert dvc.experiments.get_baseline(exp_rev) == detached_rev
assert (tmp_dir / "params.yaml").read_text().strip() == "foo: 3"
assert (tmp_dir / "metrics.yaml").read_text().strip() == "foo: 3"
@@ -300,9 +291,7 @@ def test_new_checkpoint(tmp_dir, scm, dvc, mocker):
scm.commit("init")
new_mock = mocker.spy(dvc.experiments, "new")
dvc.reproduce(
stage.addressing, experiment=True, checkpoint=True, params=["foo=2"]
)
dvc.experiments.run(stage.addressing, checkpoint=True, params=["foo=2"])
new_mock.assert_called_once()
assert (tmp_dir / "foo").read_text() == "5"
@@ -333,16 +322,13 @@ def test_continue_checkpoint(tmp_dir, scm, dvc, mocker):
)
scm.commit("init")
results = dvc.reproduce(
stage.addressing, experiment=True, checkpoint=True, params=["foo=2"]
results = dvc.experiments.run(
stage.addressing, checkpoint=True, params=["foo=2"]
)
exp_rev = first(results)
dvc.reproduce(
stage.addressing,
experiment=True,
checkpoint=True,
checkpoint_continue=exp_rev,
dvc.experiments.run(
stage.addressing, checkpoint=True, checkpoint_continue=exp_rev,
)
assert (tmp_dir / "foo").read_text() == "10"
+3 -7
View File
@@ -40,7 +40,7 @@ def test_show_experiment(tmp_dir, scm, dvc):
scm.repo.rev_parse(baseline_rev).committed_date
)
dvc.reproduce(PIPELINE_FILE, experiment=True, params=["foo=2"])
dvc.experiments.run(PIPELINE_FILE, params=["foo=2"])
results = dvc.experiments.show()
expected_baseline = {
@@ -76,9 +76,7 @@ def test_show_queued(tmp_dir, scm, dvc):
scm.commit("baseline")
baseline_rev = scm.get_rev()
dvc.reproduce(
stage.addressing, experiment=True, params=["foo=2"], queue=True
)
dvc.experiments.run(stage.addressing, params=["foo=2"], queue=True)
exp_rev = dvc.experiments.scm.resolve_rev("stash@{0}")
results = dvc.experiments.show()[baseline_rev]
@@ -93,9 +91,7 @@ def test_show_queued(tmp_dir, scm, dvc):
scm.commit("new commit")
new_rev = scm.get_rev()
dvc.reproduce(
stage.addressing, experiment=True, params=["foo=3"], queue=True
)
dvc.experiments.run(stage.addressing, params=["foo=3"], queue=True)
exp_rev = dvc.experiments.scm.resolve_rev("stash@{0}")
results = dvc.experiments.show()[new_rev]
+16 -24
View File
@@ -6,6 +6,8 @@ from dvc.command.experiments import (
)
from dvc.dvcfile import PIPELINE_FILE
from .test_repro import default_arguments as repro_arguments
def test_experiments_diff(dvc, mocker):
cli_args = parse_args(
@@ -61,32 +63,22 @@ def test_experiments_show(dvc, mocker):
)
default_run_arguments = {
"all_pipelines": False,
"downstream": False,
"dry": False,
"force": False,
"run_cache": True,
"interactive": False,
"no_commit": False,
"pipeline": False,
"single_item": False,
"recursive": False,
"force_downstream": False,
"params": [],
"queue": False,
"run_all": False,
"jobs": None,
"checkpoint": False,
"checkpoint_continue": None,
"experiment": True,
}
def test_experiments_run(dvc, mocker):
default_arguments = {
"params": [],
"queue": False,
"run_all": False,
"jobs": None,
"checkpoint": False,
"checkpoint_continue": None,
}
default_arguments.update(repro_arguments)
cmd = CmdExperimentsRun(parse_args(["exp", "run"]))
mocker.patch.object(cmd.repo, "reproduce")
mocker.patch.object(cmd.repo.experiments, "run")
cmd.run()
cmd.repo.reproduce.assert_called_with(
PIPELINE_FILE, **default_run_arguments
cmd.repo.experiments.run.assert_called_with(
PIPELINE_FILE, **default_arguments
)
-5
View File
@@ -14,11 +14,6 @@ default_arguments = {
"single_item": False,
"recursive": False,
"force_downstream": False,
"experiment": False,
"params": [],
"queue": False,
"run_all": False,
"jobs": None,
"pull": False,
}