Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-10 15:28:34 +01:00
Code Issues Packages Projects Releases Wiki Activity

exp: use new collection in show/cli commands

Browse Source 
- Updates 'dvc exp show --json' to use new data format
This commit is contained in:
Peter Rowlands
2023-04-20 16:58:32 +09:00
committed by Peter Rowlands (변기호)
parent 58f74ae543
commit ec090c5680
8 changed files with 673 additions and 1788 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/commands/experiments/show.py
+18 -301
View File
@@ -2,9 +2,8 @@ import argparse
import logging
import os
import re
from collections import Counter, OrderedDict, defaultdict
from datetime import date, datetime
from typing import TYPE_CHECKING, Any, Dict, Set
from typing import TYPE_CHECKING, Dict, Iterable
from funcy import lmap
@@ -12,9 +11,8 @@ from dvc.cli import completion
from dvc.cli.command import CmdBase
from dvc.cli.utils import append_doc_link
from dvc.commands.metrics import DEFAULT_PRECISION
from dvc.exceptions import DvcException, InvalidArgumentError
from dvc.exceptions import DvcException
from dvc.ui import ui
from dvc.utils.flatten import flatten
from dvc.utils.serialize import encode_exception
if TYPE_CHECKING:
@@ -28,30 +26,6 @@ FILL_VALUE_ERRORED = "!"
logger = logging.getLogger(__name__)
def _update_names(names, items):
for name, item in items:
item = item.get("data", {})
if isinstance(item, dict):
item = flatten(item)
names[name].update({key: None for key in item})
def _collect_names(all_experiments, **kwargs):
metric_names: Dict[str, Dict[str, Any]] = defaultdict(dict)
param_names: Dict[str, Dict[str, Any]] = defaultdict(dict)
deps_names: Set[str] = set()
for _, experiments in all_experiments.items():
for exp_data in experiments.values():
exp = exp_data.get("data", {})
_update_names(metric_names, exp.get("metrics", {}).items())
_update_names(param_names, exp.get("params", {}).items())
for dep_name in exp.get("deps", {}):
deps_names.add(dep_name)
return metric_names, param_names, sorted(deps_names)
experiment_types = {
"checkpoint_tip": "│ ╓",
"checkpoint_commit": "│ ╟",
@@ -62,229 +36,6 @@ experiment_types = {
}
def _collect_rows( # noqa: C901, PLR0912, PLR0913, PLR0915
base_rev,
experiments,
all_headers,
metric_headers,
param_headers,
metric_names,
param_names,
deps_names,
precision=DEFAULT_PRECISION,
sort_by=None,
sort_order=None,
fill_value=FILL_VALUE,
iso=False,
):
from dvc.scm import Git
if sort_by:
sort_path, sort_name, sort_type = _sort_column(
sort_by, metric_names, param_names
)
reverse = sort_order == "desc"
experiments = _sort_exp(experiments, sort_path, sort_name, sort_type, reverse)
new_checkpoint = True
for i, (rev, results) in enumerate(experiments.items()):
fill_value = FILL_VALUE_ERRORED if results.get("error") else fill_value
row_dict = {k: fill_value for k in all_headers}
exp = results.get("data", {})
state = exp.get("status")
if state == "Success":
state = fill_value
is_baseline = rev == "baseline"
if is_baseline:
name_rev = base_rev[:7] if Git.is_sha(base_rev) else base_rev
else:
name_rev = rev[:7]
tip = exp.get("checkpoint_tip")
parent_rev = exp.get("checkpoint_parent", "")
parent_exp = experiments.get(parent_rev, {}).get("data", {})
parent_tip = parent_exp.get("checkpoint_tip")
parent = ""
if is_baseline:
typ = "baseline"
elif tip:
if tip == parent_tip:
typ = "checkpoint_tip" if new_checkpoint else "checkpoint_commit"
elif parent_rev == base_rev:
typ = "checkpoint_base"
else:
typ = "checkpoint_commit"
parent = parent_rev[:7]
elif i < len(experiments) - 1:
typ = "branch_commit"
else:
typ = "branch_base"
if not is_baseline:
new_checkpoint = not (tip and tip == parent_tip)
row_dict["Experiment"] = exp.get("name", "")
row_dict["rev"] = name_rev
row_dict["typ"] = typ
row_dict["Created"] = format_time(exp.get("timestamp"), fill_value, iso)
row_dict["parent"] = parent
row_dict["State"] = state
row_dict["Executor"] = exp.get("executor", fill_value)
_extend_row(
row_dict,
metric_names,
metric_headers,
exp.get("metrics", {}).items(),
precision,
fill_value=fill_value,
)
_extend_row(
row_dict,
param_names,
param_headers,
exp.get("params", {}).items(),
precision,
fill_value=fill_value,
)
for dep in deps_names:
hash_info = exp.get("deps", {}).get(dep, {}).get("hash")
if hash_info is not None:
hash_info = hash_info[:7]
row_dict[dep] = hash_info
yield list(row_dict.values())
def _sort_column(sort_by, metric_names, param_names):
path, _, sort_name = sort_by.rpartition(":")
matches = set()
if path:
if path in metric_names and sort_name in metric_names[path]:
matches.add((path, sort_name, "metrics"))
if path in param_names and sort_name in param_names[path]:
matches.add((path, sort_name, "params"))
else:
for path in metric_names:
if sort_name in metric_names[path]:
matches.add((path, sort_name, "metrics"))
for path in param_names:
if sort_name in param_names[path]:
matches.add((path, sort_name, "params"))
if len(matches) == 1:
return matches.pop()
if len(matches) > 1:
raise InvalidArgumentError(
"Ambiguous sort column '{}' matched '{}'".format(
sort_by,
", ".join([f"{path}:{name}" for path, name, _ in matches]),
)
)
raise InvalidArgumentError(f"Unknown sort column '{sort_by}'")
def _sort_exp(experiments, sort_path, sort_name, typ, reverse):
def _sort(item):
rev, exp = item
exp_data = exp.get("data", {})
tip = exp_data.get("checkpoint_tip")
if tip and tip != rev:
# Sort checkpoint experiments by tip commit
return _sort((tip, experiments[tip]))
data = exp_data.get(typ, {}).get(sort_path, {}).get("data", {})
val = flatten(data).get(sort_name)
return val is None, val
ret = OrderedDict()
if "baseline" in experiments:
ret["baseline"] = experiments.pop("baseline")
ret.update(sorted(experiments.items(), key=_sort, reverse=reverse))
return ret
def format_time(datetime_obj, fill_value=FILL_VALUE, iso=False):
if datetime_obj is None:
return fill_value
if iso:
return datetime_obj.isoformat()
if datetime_obj.date() == date.today():
fmt = "%I:%M %p"
else:
fmt = "%b %d, %Y"
return datetime_obj.strftime(fmt)
def _extend_row(row, names, headers, items, precision, fill_value=FILL_VALUE):
from dvc.compare import _format_field, with_value
for fname, data in items:
item = data.get("data", {})
item = flatten(item) if isinstance(item, dict) else {fname: item}
for name in names[fname]:
value = with_value(
item.get(name),
FILL_VALUE_ERRORED if data.get("error", None) else fill_value,
)
# wrap field data in ui.rich_text, otherwise rich may
# interpret unescaped braces from list/dict types as rich
# markup tags
value = ui.rich_text(str(_format_field(value, precision)))
if name in headers:
row[name] = value
else:
row[f"{fname}:{name}"] = value
def experiments_table( # noqa: PLR0913
all_experiments,
headers,
metric_headers,
metric_names,
param_headers,
param_names,
deps_names,
sort_by=None,
sort_order=None,
precision=DEFAULT_PRECISION,
fill_value=FILL_VALUE,
iso=False,
) -> "TabularData":
from funcy import lconcat
from dvc.compare import TabularData
all_headers = lconcat(headers, metric_headers, param_headers, deps_names)
td = TabularData(all_headers, fill_value=fill_value)
for base_rev, experiments in all_experiments.items():
rows = _collect_rows(
base_rev,
experiments,
all_headers,
metric_headers,
param_headers,
metric_names,
param_names,
deps_names,
sort_by=sort_by,
sort_order=sort_order,
precision=precision,
fill_value=fill_value,
iso=iso,
)
td.extend(rows)
return td
def prepare_exp_id(kwargs) -> "RichText":
exp_name = kwargs["Experiment"]
rev = kwargs["rev"]
@@ -309,7 +60,8 @@ def baseline_styler(typ):
def show_experiments( # noqa: C901, PLR0912
all_experiments,
td: "TabularData",
headers: Dict[str, Iterable[str]],
keep=None,
drop=None,
pager=True,
@@ -318,40 +70,6 @@ def show_experiments( # noqa: C901, PLR0912
pcp=False,
**kwargs,
):
from funcy.seqs import flatten as flatten_list
metric_names, param_names, deps_names = _collect_names(all_experiments)
headers_list = [
"Experiment",
"rev",
"typ",
"Created",
"parent",
"State",
"Executor",
]
names = {**metric_names, **param_names}
counter = Counter(flatten_list([list(a.keys()) for a in names.values()]))
counter.update(headers_list)
metric_headers = _normalize_headers(metric_names, counter)
param_headers = _normalize_headers(param_names, counter)
td = experiments_table(
all_experiments,
headers_list,
metric_headers,
metric_names,
param_headers,
param_names,
deps_names,
kwargs.get("sort_by"),
kwargs.get("sort_order"),
kwargs.get("precision"),
kwargs.get("fill_value"),
kwargs.get("iso"),
)
if keep:
for col in td.keys(): # noqa: SIM118
if re.match(keep, col):
@@ -368,11 +86,6 @@ def show_experiments( # noqa: C901, PLR0912
td.column("Experiment")[:] = map(prepare_exp_id, td.as_dict(merge_headers))
td.drop(*merge_headers[1:])
headers = {
"metrics": metric_headers,
"params": param_headers,
"deps": deps_names,
}
styles = {
"Experiment": {"no_wrap": True, "header_style": "black on grey93"},
"Created": {"header_style": "black on grey93"},
@@ -453,19 +166,17 @@ def _normalize_headers(names, count):
def _format_json(item):
from dvc.repo.experiments.show import _CachedError
if isinstance(item, (date, datetime)):
return item.isoformat()
if isinstance(item, _CachedError):
return {"type": getattr(item, "typ", "_CachedError"), "msg": str(item)}
return encode_exception(item)
class CmdExperimentsShow(CmdBase):
def run(self):
from dvc.repo.experiments.show import tabulate
try:
all_experiments = self.repo.experiments.show(
exps = self.repo.experiments.show(
all_branches=self.args.all_branches,
all_tags=self.args.all_tags,
all_commits=self.args.all_commits,
@@ -483,23 +194,29 @@ class CmdExperimentsShow(CmdBase):
return 1
if self.args.json:
ui.write_json(all_experiments, default=_format_json)
ui.write_json([exp.dumpd() for exp in exps], default=_format_json)
else:
precision = (
self.args.precision or None if self.args.csv else DEFAULT_PRECISION
)
fill_value = "" if self.args.csv else FILL_VALUE
iso = self.args.csv
td, headers = tabulate(
exps,
precision=precision,
fill_value=fill_value,
iso=iso,
sort_by=self.args.sort_by,
sort_order=self.args.sort_order,
)
show_experiments(
all_experiments,
td,
headers,
keep=self.args.keep,
drop=self.args.drop,
sort_by=self.args.sort_by,
sort_order=self.args.sort_order,
precision=precision,
fill_value=fill_value,
iso=iso,
pager=not self.args.no_pager,
csv=self.args.csv,
markdown=self.args.markdown,
dvc/commands/queue/status.py
+3 -5
View File
@@ -1,10 +1,8 @@
import argparse
import logging
from typing import Dict, List, Optional
from dvc.cli.command import CmdBase
from dvc.cli.utils import append_doc_link
from dvc.commands.experiments.show import format_time
from dvc.compare import TabularData
from dvc.ui import ui
@@ -15,9 +13,9 @@ class CmdQueueStatus(CmdBase):
"""Show queue task and worker status."""
def run(self) -> int:
result: List[
Dict[str, Optional[str]]
] = self.repo.experiments.celery_queue.status()
from dvc.repo.experiments.show import format_time
result = self.repo.experiments.celery_queue.status()
if result:
all_headers = ["Task", "Name", "Created", "Status"]
td = TabularData(all_headers)
dvc/repo/experiments/diff.py
+7 -8
View File
@@ -7,7 +7,7 @@ logger = logging.getLogger(__name__)
def diff(repo, *args, a_rev=None, b_rev=None, param_deps=False, **kwargs):
from dvc.repo.experiments.show import collect_experiment_commit
from dvc.repo.experiments.collect import collect_rev
from dvc.scm import resolve_rev
if repo.scm.no_commits:
@@ -15,22 +15,21 @@ def diff(repo, *args, a_rev=None, b_rev=None, param_deps=False, **kwargs):
if a_rev:
rev = resolve_rev(repo.scm, a_rev)
old = collect_experiment_commit(repo, rev, param_deps=param_deps)
else:
old = collect_experiment_commit(repo, "HEAD", param_deps=param_deps)
rev = resolve_rev(repo.scm, "HEAD")
old = collect_rev(repo, rev, param_deps=param_deps)
if b_rev:
rev = resolve_rev(repo.scm, b_rev)
new = collect_experiment_commit(repo, rev, param_deps=param_deps)
else:
new = collect_experiment_commit(repo, "workspace", param_deps=param_deps)
rev = "workspace"
new = collect_rev(repo, rev, param_deps=param_deps)
with_unchanged = kwargs.pop("all", False)
return {
key: _diff(
format_dict(old.get("data", {}).get(key, {})),
format_dict(new.get("data", {}).get(key, {})),
format_dict(getattr(old.data, key, {})),
format_dict(getattr(new.data, key, {})),
with_unchanged=with_unchanged,
)
for key in ["metrics", "params"]
dvc/repo/experiments/show.py
+352 -479
View File
@@ -1,517 +1,390 @@
import logging
from collections import OrderedDict, defaultdict
from datetime import datetime
from enum import Enum
from itertools import chain
from collections import Counter, defaultdict
from datetime import date, datetime
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterable,
Iterator,
List,
Literal,
Mapping,
NamedTuple,
Optional,
Set,
Tuple,
Union,
)
from scmrepo.exceptions import SCMError as InnerScmError
from dvc.exceptions import InvalidArgumentError
from dvc.scm import Git
from dvc.ui import ui
from dvc.utils.flatten import flatten
from dvc.exceptions import DvcException
from dvc.scm import Git, SCMError, iter_revs, resolve_rev
from .refs import ExpRefInfo
from .serialize import SerializableError, SerializableExp
from .collect import collect
if TYPE_CHECKING:
from scmrepo.git.objects import GitCommit
from dvc.compare import TabularData
from dvc.repo import Repo
from dvc.ui.table import CellT
from .serialize import ExpRange, ExpState
logger = logging.getLogger(__name__)
class ExpStatus(Enum):
Success = 0
Queued = 1
Running = 2
Failed = 3
class _CachedError(DvcException):
def __init__(self, msg, typ, *args):
super().__init__(msg, *args)
self.typ = typ
def _is_scm_error(collected_exp: Dict[str, Any]) -> bool:
if "error" in collected_exp and (
isinstance(collected_exp["error"], (_CachedError, SCMError, InnerScmError))
):
return True
return False
def _format_exp(exp: SerializableExp) -> Dict[str, Any]:
# SerializableExp always includes error but we need to strip it from show
# output when it is false-y to maintain compatibility w/tools that consume
# json output and assume that "error" key presence means there was an error
exp_dict = exp.dumpd()
if "error" in exp_dict and not exp_dict["error"]:
del exp_dict["error"]
return {"data": exp_dict}
def _format_error(error: SerializableError):
msg = error.msg or "None"
return {"data": {}, "error": _CachedError(msg, error.type)}
def collect_experiment_commit(
def show(
repo: "Repo",
exp_rev: str,
status: ExpStatus = ExpStatus.Success,
param_deps: bool = False,
force: bool = False,
revs: Union[List[str], str, None] = None,
all_branches: bool = False,
all_tags: bool = False,
all_commits: bool = False,
num: int = 1,
hide_queued: bool = False,
hide_failed: bool = False,
sha_only: bool = False,
**kwargs,
) -> Dict[str, Any]:
cache = repo.experiments.cache
# TODO: support filtering serialized exp when param_deps is set
if exp_rev != "workspace" and not (force or param_deps):
cached_exp = cache.get(exp_rev)
if cached_exp:
if status == ExpStatus.Running or (
isinstance(cached_exp, SerializableExp)
and cached_exp.status == ExpStatus.Running.name
):
# expire cached queued exp entry once we start running it
cache.delete(exp_rev)
elif isinstance(cached_exp, SerializableError):
return _format_error(cached_exp)
else:
return _format_exp(cached_exp)
try:
exp = _collect_from_repo(
repo,
exp_rev,
status=status,
param_deps=param_deps,
force=force,
**kwargs,
)
if not (
exp_rev == "workspace"
or status == ExpStatus.Running
or param_deps
or exp.contains_error
):
cache.put(exp, force=True)
return _format_exp(exp)
except Exception as exc: # noqa: BLE001, pylint: disable=broad-except
logger.debug("", exc_info=True)
error = SerializableError(str(exc), type(exc).__name__)
return _format_error(error)
def _collect_from_repo(
repo: "Repo",
exp_rev: str,
status: ExpStatus = ExpStatus.Success,
running: Optional[Dict[str, Any]] = None,
**kwargs,
) -> "SerializableExp":
running = running or {}
with repo.switch(exp_rev) as rev:
if rev == "workspace":
timestamp: Optional[datetime] = None
else:
commit = repo.scm.resolve_commit(rev)
timestamp = datetime.fromtimestamp(commit.commit_time)
if status == ExpStatus.Running:
executor: Optional[str] = running.get(exp_rev, {}).get("location", None)
else:
executor = None
if status == ExpStatus.Failed:
error: Optional["SerializableError"] = SerializableError(
"Experiment run failed."
)
else:
error = None
return SerializableExp.from_repo(
repo,
rev=exp_rev,
timestamp=timestamp,
status=status.name,
executor=executor,
error=error,
)
def _collect_complete_experiment(
repo: "Repo",
baseline: str,
exp_rev: str,
running: Dict[str, Any],
revs: List[str],
**kwargs,
) -> Dict[str, Dict[str, Any]]:
results: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
checkpoint = len(revs) > 1
prev = ""
for rev in revs:
status = ExpStatus.Running if rev in running else ExpStatus.Success
collected_exp = collect_experiment_commit(
repo,
rev,
status=status,
running=running,
**kwargs,
)
if _is_scm_error(collected_exp):
return {}
if checkpoint:
exp = {"checkpoint_tip": exp_rev}
if prev:
results[prev]["data"][ # type: ignore[unreachable]
"checkpoint_parent"
] = rev
if rev in results:
results[rev]["data"].update(exp)
results.move_to_end(rev)
else:
exp.update(collected_exp["data"])
else:
exp = collected_exp["data"]
if rev not in results:
results[rev] = {"data": exp}
prev = rev
if checkpoint and prev:
results[prev]["data"]["checkpoint_parent"] = baseline
return results
def _collect_branch(
repo: "Repo",
baseline: str,
running: Dict[str, Any],
refs: Optional[Iterable[str]] = None,
**kwargs,
) -> Dict[str, Dict[str, Any]]:
results: Dict[str, Dict[str, Any]] = OrderedDict()
status = ExpStatus.Running if baseline in running else ExpStatus.Success
results["baseline"] = collect_experiment_commit(
) -> List["ExpState"]:
return collect(
repo,
baseline,
status=status,
running=running,
revs=revs,
all_branches=all_branches,
all_tags=all_tags,
all_commits=all_commits,
num=num,
hide_queued=hide_queued,
hide_failed=hide_failed,
sha_only=sha_only,
**kwargs,
)
if baseline == "workspace" or _is_scm_error(results["baseline"]):
return results
ref_info = ExpRefInfo(baseline_sha=baseline)
commits: List[Tuple[str, "GitCommit", str, List[str]]] = []
assert isinstance(repo.scm, Git)
if refs:
ref_it = (ref for ref in iter(refs) if ref.startswith(str(ref_info)))
else:
ref_it = repo.scm.iter_refs(base=str(ref_info))
for ref in ref_it:
try:
commit = repo.scm.resolve_commit(ref)
exp_rev = resolve_rev(repo.scm, ref)
revs = list(repo.scm.branch_revs(exp_rev, baseline))
except (SCMError, InnerScmError):
continue
commits.append((ref, commit, exp_rev, revs))
def tabulate(
baseline_states: Iterable["ExpState"],
fill_value: str = "-",
error_value: str = "!",
**kwargs,
) -> Tuple["TabularData", Dict[str, Iterable[str]]]:
"""Return table data for experiments.
for exp_ref, _, exp_rev, revs in sorted(
commits, key=lambda x: x[1].commit_time, reverse=True
):
ref_info = ExpRefInfo.from_ref(exp_ref)
assert ref_info.baseline_sha == baseline
collected_exp = _collect_complete_experiment(
repo,
baseline=baseline,
exp_rev=exp_rev,
running=running,
revs=revs,
Returns:
Tuple of (table_data, data_headers)
"""
from funcy import lconcat
from funcy.seqs import flatten as flatten_list
from dvc.compare import TabularData
data_names = _collect_names(baseline_states)
metrics_names = data_names.metrics
params_names = data_names.params
deps_names = data_names.sorted_deps
headers = [
"Experiment",
"rev",
"typ",
"Created",
"parent",
"State",
"Executor",
]
names = {**metrics_names, **params_names}
counter = Counter(flatten_list([list(a.keys()) for a in names.values()]))
counter.update(headers)
metrics_headers = _normalize_headers(metrics_names, counter)
params_headers = _normalize_headers(params_names, counter)
all_headers = lconcat(headers, metrics_headers, params_headers, deps_names)
td = TabularData(all_headers, fill_value=fill_value)
td.extend(
_build_rows(
baseline_states,
all_headers=all_headers,
metrics_headers=metrics_headers,
params_headers=params_headers,
metrics_names=metrics_names,
params_names=params_names,
deps_names=deps_names,
fill_value=fill_value,
error_value=error_value,
**kwargs,
)
if _is_scm_error(collected_exp):
continue
results.update(collected_exp)
return results
)
data_headers: Dict[str, Iterable[str]] = {
"metrics": metrics_headers,
"params": params_headers,
"deps": deps_names,
}
return td, data_headers
def get_branch_names(
scm: "Git", baseline_set: Iterable[str], refs: Optional[Iterable[str]] = None
) -> Dict[str, Optional[str]]:
names: Dict[str, Optional[str]] = {}
bases = [
f"refs/exps/{baseline[:2]}/{baseline[2:]}/" for baseline in baseline_set
] + ["refs/heads/", "refs/tags/"]
ref_it = iter(refs) if refs else scm.iter_refs()
for ref in ref_it:
for base in bases:
if ref.startswith(base):
try:
rev = scm.get_ref(ref)
names[rev] = ref[len(base) :]
except KeyError:
logger.debug("unresolved ref %s", ref)
break
logger.debug("found refs for revs: %s", names)
return names
def update_names( # noqa: C901
repo: "Repo",
branch_names: Dict[str, Optional[str]],
result: Dict[str, Dict[str, Any]],
):
rev_set = set()
for baseline in result:
for rev in result[baseline]:
if rev == "baseline":
rev = baseline
if rev != "workspace":
rev_set.add(rev)
if rev_set:
rev_set.difference_update(branch_names.keys())
exact_name = repo.experiments.get_exact_name(rev_set)
for baseline, baseline_results in result.items():
name_set: Set[str] = set()
for rev, rev_result in baseline_results.items():
name: Optional[str] = None
if rev == "baseline":
rev = baseline
if rev == "workspace":
continue
name = branch_names.get(rev, None) or exact_name[rev]
if name and name not in name_set:
name_set.add(name)
rev_result["data"]["name"] = name
def _collect_active_experiment(
repo: "Repo",
found_revs: Dict[str, List[str]],
running: Dict[str, Any],
def _build_rows(
baseline_states: Iterable["ExpState"],
*,
all_headers: Iterable[str],
fill_value: str,
sort_by: Optional[str] = None,
sort_order: Optional[Literal["asc", "desc"]] = None,
**kwargs,
) -> Dict[str, Dict[str, Any]]:
result: Dict[str, Dict] = defaultdict(OrderedDict)
for entry in chain(
repo.experiments.tempdir_queue.iter_active(),
repo.experiments.celery_queue.iter_active(),
):
stash_rev = entry.stash_rev
if entry.baseline_rev in found_revs and (
stash_rev not in running or not running[stash_rev].get("last")
) -> Iterator[Tuple["CellT", ...]]:
for baseline in baseline_states:
row: Dict[str, "CellT"] = {k: fill_value for k in all_headers}
row["Experiment"] = ""
if baseline.name:
row["rev"] = baseline.name
elif Git.is_sha(baseline.rev):
row["rev"] = baseline.rev[:7]
else:
row["rev"] = baseline.rev
row["typ"] = "baseline"
row["parent"] = ""
if baseline.data:
row["Created"] = format_time(
baseline.data.timestamp, fill_value=fill_value, **kwargs
)
row.update(
_data_cells( # pylint: disable=missing-kwoa
baseline, fill_value=fill_value, **kwargs
)
)
yield tuple(row.values())
if baseline.experiments:
if sort_by:
metrics_names: Mapping[str, Iterable[str]] = kwargs.get(
"metrics_names", {}
)
params_names: Mapping[str, Iterable[str]] = kwargs.get(
"params_names", {}
)
sort_path, sort_name, sort_type = _sort_column(
sort_by, metrics_names, params_names
)
reverse = sort_order == "desc"
experiments = _sort_exp(
baseline.experiments, sort_path, sort_name, sort_type, reverse
)
else:
experiments = baseline.experiments
for i, child in enumerate(experiments):
yield from _exp_range_rows(
child,
all_headers=all_headers,
fill_value=fill_value,
is_base=i == len(baseline.experiments) - 1,
**kwargs,
)
def _sort_column(
sort_by: str,
metric_names: Mapping[str, Iterable[str]],
param_names: Mapping[str, Iterable[str]],
) -> Tuple[str, str, str]:
path, _, sort_name = sort_by.rpartition(":")
matches: Set[Tuple[str, str, str]] = set()
if path:
if path in metric_names and sort_name in metric_names[path]:
matches.add((path, sort_name, "metrics"))
if path in param_names and sort_name in param_names[path]:
matches.add((path, sort_name, "params"))
else:
for path in metric_names:
if sort_name in metric_names[path]:
matches.add((path, sort_name, "metrics"))
for path in param_names:
if sort_name in param_names[path]:
matches.add((path, sort_name, "params"))
if len(matches) == 1:
return matches.pop()
if len(matches) > 1:
raise InvalidArgumentError(
"Ambiguous sort column '{}' matched '{}'".format(
sort_by,
", ".join([f"{path}:{name}" for path, name, _ in matches]),
)
)
raise InvalidArgumentError(f"Unknown sort column '{sort_by}'")
def _sort_exp(
experiments: Iterable["ExpRange"],
sort_path: str,
sort_name: str,
typ: str,
reverse: bool,
) -> List["ExpRange"]:
from funcy import first
def _sort(exp_range: "ExpRange"):
exp = first(exp_range.revs)
if not exp:
return True
data = exp.data.dumpd().get(typ, {}).get(sort_path, {}).get("data", {})
val = flatten(data).get(sort_name)
return val is None, val
return sorted(experiments, key=_sort, reverse=reverse)
def _exp_range_rows(
exp_range: "ExpRange",
*,
all_headers: Iterable[str],
fill_value: str,
is_base: bool = False,
**kwargs,
) -> Iterator[Tuple["CellT", ...]]:
for i, exp in enumerate(exp_range.revs):
row: Dict[str, "CellT"] = {k: fill_value for k in all_headers}
row["Experiment"] = exp.name or ""
row["rev"] = exp.rev[:7] if Git.is_sha(exp.rev) else exp.rev
if len(exp_range.revs) > 1:
if i == 0:
row["typ"] = "checkpoint_tip"
elif i == len(exp_range.revs) - 1:
row["typ"] = "checkpoint_base"
else:
row["typ"] = "checkpoint_commit"
else:
row["typ"] = "branch_base" if is_base else "branch_commit"
row["parent"] = ""
if exp_range.executor:
row["State"] = exp_range.executor.state.capitalize()
if exp_range.executor.name:
row["Executor"] = exp_range.executor.name.capitalize()
if exp.data:
row["Created"] = format_time(
exp.data.timestamp, fill_value=fill_value, **kwargs
)
row.update(
_data_cells( # pylint: disable=missing-kwoa
exp, fill_value=fill_value, **kwargs
)
)
yield tuple(row.values())
def _data_cells(
exp: "ExpState",
*,
metrics_headers: Iterable[str],
params_headers: Iterable[str],
metrics_names: Mapping[str, Iterable[str]],
params_names: Mapping[str, Iterable[str]],
deps_names: Iterable[str],
fill_value: str = "-",
error_value: str = "!",
precision: Optional[int] = None,
**kwargs,
) -> Iterator[Tuple[str, "CellT"]]:
def _d_cells(
d: Mapping[str, Any],
names: Mapping[str, Iterable[str]],
headers: Iterable[str],
) -> Iterator[Tuple[str, "CellT"]]:
from dvc.compare import _format_field, with_value
for fname, data in d.items():
item = data.get("data", {})
item = flatten(item) if isinstance(item, dict) else {fname: item}
for name in names[fname]:
value = with_value(
item.get(name),
error_value if data.get("error") else fill_value,
)
# wrap field data in ui.rich_text, otherwise rich may
# interpret unescaped braces from list/dict types as rich
# markup tags
value = ui.rich_text(str(_format_field(value, precision)))
if name in headers:
yield name, value
else:
yield f"{fname}:{name}", value
if not exp.data:
return
yield from _d_cells(exp.data.metrics, metrics_names, metrics_headers)
yield from _d_cells(exp.data.params, params_names, params_headers)
for name in deps_names:
dep = exp.data.deps.get(name)
if dep:
yield name, dep.hash or fill_value
def format_time(
timestamp: Optional[datetime],
fill_value: str = "-",
iso: bool = False,
**kwargs,
) -> str:
if not timestamp:
return fill_value
if iso:
return timestamp.isoformat()
if timestamp.date() == date.today():
fmt = "%I:%M %p"
else:
fmt = "%b %d, %Y"
return timestamp.strftime(fmt)
class _DataNames(NamedTuple):
# NOTE: we use nested dict instead of set for metrics/params names to
# preserve key ordering
metrics: Dict[str, Dict[str, Any]]
params: Dict[str, Dict[str, Any]]
deps: Set[str]
@property
def sorted_deps(self):
return sorted(self.deps)
def update(self, other: "_DataNames"):
def _update_d(
d: Dict[str, Dict[str, Any]], other_d: Mapping[str, Mapping[str, Any]]
):
collected_exp = collect_experiment_commit(
repo,
stash_rev,
status=ExpStatus.Running,
running=running,
**kwargs,
)
if _is_scm_error(collected_exp):
continue
result[entry.baseline_rev][stash_rev] = collected_exp
for k, v in other_d.items():
if k in d:
d[k].update(v)
else:
d[k] = dict(v)
_update_d(self.metrics, other.metrics)
_update_d(self.params, other.params)
self.deps.update(other.deps)
def _collect_names(exp_states: Iterable["ExpState"]) -> _DataNames:
result = _DataNames(defaultdict(dict), defaultdict(dict), set())
def _collect_d(result_d: Dict[str, Dict[str, Any]], data_d: Dict[str, Any]):
for path, item in data_d.items():
item = item.get("data", {})
if isinstance(item, dict):
item = flatten(item)
result_d[path].update((key, None) for key in item)
for exp in exp_states:
if exp.data:
_collect_d(result.metrics, exp.data.metrics)
_collect_d(result.params, exp.data.params)
result.deps.update(exp.data.deps)
if exp.experiments:
for child in exp.experiments:
result.update(_collect_names(child.revs))
return result
def _collect_queued_experiment(
repo: "Repo",
found_revs: Dict[str, List[str]],
running: Dict[str, Any],
**kwargs,
) -> Dict[str, Dict[str, Any]]:
result: Dict[str, Dict] = defaultdict(OrderedDict)
for entry in repo.experiments.celery_queue.iter_queued():
stash_rev = entry.stash_rev
if entry.baseline_rev in found_revs:
collected_exp = collect_experiment_commit(
repo,
stash_rev,
status=ExpStatus.Queued,
running=running,
**kwargs,
)
if _is_scm_error(collected_exp):
continue
result[entry.baseline_rev][stash_rev] = collected_exp
return result
def _collect_failed_experiment(
repo: "Repo",
found_revs: Dict[str, List[str]],
running: Dict[str, Any],
**kwargs,
) -> Dict[str, Dict[str, Any]]:
result: Dict[str, Dict] = defaultdict(OrderedDict)
for queue_done_result in repo.experiments.celery_queue.iter_failed():
entry = queue_done_result.entry
stash_rev = entry.stash_rev
if entry.baseline_rev in found_revs:
collected_exp = collect_experiment_commit(
repo,
stash_rev,
status=ExpStatus.Failed,
running=running,
**kwargs,
)
if _is_scm_error(collected_exp):
continue
result[entry.baseline_rev][stash_rev] = collected_exp
return result
def update_new(
to_dict: Dict[str, Dict[str, Any]], from_dict: Dict[str, Dict[str, Any]]
):
for baseline, experiments in from_dict.items():
for rev, experiment in experiments.items():
to_dict[baseline][rev] = to_dict[baseline].get(rev, experiment)
def move_properties_to_head(result: Dict[str, Dict[str, Dict[str, Any]]]):
for _, baseline_results in result.items():
checkpoint: bool = False
head: Dict[str, Any] = {}
for rev, rev_data in baseline_results.items():
if (
"data" not in rev_data
or rev_data["data"].get("checkpoint_tip", None) is None
):
checkpoint = False
head = {}
continue
rev_result: Dict[str, Any] = rev_data["data"]
if (
checkpoint is True
and rev_result["checkpoint_tip"] == head["checkpoint_tip"]
):
if "name" in rev_result and "name" not in head:
head["name"] = rev_result["name"]
del rev_result["name"]
if rev_result["executor"]:
if not head["executor"]:
head["executor"] = rev_result["executor"]
rev_result["executor"] = None
if rev_result["status"] == ExpStatus.Running.name:
head["status"] = ExpStatus.Running.name
rev_result["status"] = ExpStatus.Success.name
elif rev_result["checkpoint_tip"] == rev:
head = rev_result
checkpoint = True
def show( # noqa: PLR0913
repo: "Repo",
all_branches=False,
all_tags=False,
revs: Union[List[str], str, None] = None,
all_commits=False,
hide_queued=False,
hide_failed=False,
sha_only=False,
num=1,
param_deps=False,
onerror: Optional[Callable] = None,
fetch_running: bool = True,
force: bool = False,
):
if repo.scm.no_commits:
return {}
res: Dict[str, Dict] = defaultdict(OrderedDict)
if not any([revs, all_branches, all_tags, all_commits]):
revs = ["HEAD"]
if isinstance(revs, str):
revs = [revs]
assert isinstance(repo.scm, Git)
found_revs: Dict[str, List[str]] = {"workspace": []}
found_revs.update(
iter_revs(repo.scm, revs, num, all_branches, all_tags, all_commits)
)
cached_refs = list(repo.scm.iter_refs())
branch_names = get_branch_names(repo.scm, found_revs, refs=cached_refs)
running: Dict[str, Dict] = repo.experiments.get_running_exps(
fetch_refs=fetch_running
)
queued_experiment = (
_collect_queued_experiment(
repo,
found_revs,
running,
param_deps=param_deps,
force=force,
)
if not hide_queued
else {}
)
active_experiment = _collect_active_experiment(
repo,
found_revs,
running,
param_deps=param_deps,
onerror=onerror,
force=force,
)
failed_experiments = (
_collect_failed_experiment(
repo,
found_revs,
running,
param_deps=param_deps,
onerror=onerror,
force=force,
)
if not hide_failed
else {}
)
for baseline in found_revs:
res[baseline] = _collect_branch(
repo,
baseline,
running=running,
param_deps=param_deps,
onerror=onerror,
force=force,
refs=cached_refs,
)
update_new(res, failed_experiments)
update_new(res, active_experiment)
update_new(res, queued_experiment)
if not sha_only:
update_names(repo, branch_names, res)
move_properties_to_head(res)
return res
def _normalize_headers(
names: Mapping[str, Mapping[str, Any]], count: Mapping[str, int]
) -> List[str]:
return [
name if count[name] == 1 else f"{path}:{name}"
for path in names
for name in names[path]
]
tests/func/experiments/test_checkpoints.py
+5 -8
View File
@@ -255,17 +255,14 @@ def test_auto_push_self_remote(
def test_tmp_dir_failed_checkpoint(tmp_dir, scm, dvc, failed_checkpoint_stage, caplog):
baseline = scm.get_rev()
results = dvc.experiments.run(
dvc.experiments.run(
failed_checkpoint_stage.addressing, params=["foo=2"], tmp_dir=True
)
exp = first(results)
result = dvc.experiments.show()[baseline]
# Assert 4 rows: baseline, 2 checkpoints, and final commit
assert len(result) == 4
assert exp in result
result = dvc.experiments.show()[1]
assert len(result.experiments) == 1
# Assert 2 checkpoints and final commit
assert len(result.experiments[0]) == 3
assert (
"Checkpoint stage 'failed-checkpoint-file' was interrupted remaining "
"stages in pipeline will not be reproduced." in caplog.text
tests/func/experiments/test_show.py
+287 -464
View File
@@ -1,21 +1,17 @@
import logging
import os
import random
from datetime import datetime
from unittest.mock import ANY
import pytest
from funcy import first, get_in
from funcy import first
from scmrepo.exceptions import SCMError
from dvc.cli import main
from dvc.repo.experiments.executor.base import BaseExecutor, ExecutorInfo, TaskStatus
from dvc.repo.experiments.queue.base import QueueEntry
from dvc.repo.experiments.refs import CELERY_STASH, ExpRefInfo
from dvc.repo.experiments.show import _CachedError
from dvc.repo.experiments.refs import CELERY_STASH
from dvc.repo.experiments.utils import EXEC_PID_DIR, EXEC_TMP_DIR, exp_refs_by_rev
from dvc.utils import relpath
from dvc.utils.serialize import YAMLFileCorruptedError
LOCK_CONTENTS = {
"read": {
@@ -41,42 +37,73 @@ def make_executor_info(**kwargs):
return ExecutorInfo(**kwargs)
def make_executor(local=None, **kwargs):
if local:
local_executor = {
"root": ANY,
"log": ANY,
"pid": ANY,
"returncode": ANY,
"task_id": ANY,
}
local_executor.update(local)
else:
local_executor = ANY
data = {
"state": ANY,
"local": local_executor,
"name": ANY,
}
data.update(kwargs)
return data
def make_data(params=None, **kwargs):
params = {"data": params or {"foo": 1}}
data = {
"rev": ANY,
"deps": {
"copy.py": {
"hash": ANY,
"size": ANY,
"nfiles": None,
}
},
"metrics": {"metrics.yaml": params},
"outs": {},
"params": {"params.yaml": params},
"timestamp": ANY,
"meta": ANY,
}
data.update(kwargs)
return data
@pytest.mark.vscode
def test_show_branch_and_tag_name(tmp_dir, scm, dvc, exp_stage):
with tmp_dir.branch("new/branch", new=True):
tmp_dir.scm_gen("branch", "branch", "commit")
branch_rev = scm.get_rev()
result = dvc.experiments.show(all_branches=True)
assert result[branch_rev]["baseline"]["data"]["name"] == "new/branch"
expected = [None, "master", "new/branch"]
assert [exp.name for exp in result] == expected
scm.tag("new/tag")
tag_rev = scm.get_rev()
result = dvc.experiments.show(all_tags=True)
assert result[tag_rev]["baseline"]["data"]["name"] == "new/tag"
with scm.detach_head(tag_rev):
result = dvc.experiments.show(all_tags=True)
expected = [None, "new/tag"]
assert [exp.name for exp in result] == expected
@pytest.mark.vscode
def test_show_simple(tmp_dir, scm, dvc, exp_stage):
assert dvc.experiments.show()["workspace"] == {
"baseline": {
"data": {
"rev": "workspace",
"deps": {
"copy.py": {
"hash": ANY,
"size": ANY,
"nfiles": None,
}
},
"metrics": {"metrics.yaml": {"data": {"foo": 1}}},
"outs": {},
"params": {"params.yaml": {"data": {"foo": 1}}},
"status": "Success",
"executor": None,
"timestamp": None,
}
}
assert dvc.experiments.show()[0].dumpd() == {
"rev": "workspace",
"name": None,
"data": make_data(rev="workspace"),
"error": None,
"experiments": None,
}
@@ -88,39 +115,33 @@ def test_show_experiment(tmp_dir, scm, dvc, exp_stage, workspace):
scm.gitpython.repo.rev_parse(baseline_rev).committed_date
)
dvc.experiments.run(exp_stage.addressing, params=["foo=2"], tmp_dir=not workspace)
exp_rev = first(
dvc.experiments.run(
exp_stage.addressing, params=["foo=2"], tmp_dir=not workspace
)
)
results = dvc.experiments.show()
expected_baseline = {
"data": {
"rev": ANY,
"deps": {
"copy.py": {
"hash": ANY,
"size": ANY,
"nfiles": None,
}
},
"metrics": {"metrics.yaml": {"data": {"foo": 1}}},
"outs": {},
"params": {"params.yaml": {"data": {"foo": 1}}},
"status": "Success",
"executor": None,
"timestamp": timestamp,
"name": "master",
}
assert results[1].dumpd() == {
"rev": baseline_rev,
"name": "master",
"data": make_data(rev=baseline_rev, timestamp=timestamp),
"error": None,
"experiments": [
{
"revs": [
{
"rev": exp_rev,
"name": ANY,
"data": make_data(rev=exp_rev, params={"foo": 2}),
"error": None,
"experiments": None,
}
],
"executor": None,
"name": ANY,
}
],
}
expected_params = {"data": {"foo": 2}}
assert set(results.keys()) == {"workspace", baseline_rev}
experiments = results[baseline_rev]
assert len(experiments) == 2
for rev, exp in experiments.items():
if rev == "baseline":
assert exp == expected_baseline
else:
assert exp["data"]["metrics"]["metrics.yaml"] == expected_params
assert exp["data"]["params"]["params.yaml"] == expected_params
@pytest.mark.vscode
@@ -132,12 +153,28 @@ def test_show_queued(tmp_dir, scm, dvc, exp_stage):
)
exp_rev = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
results = dvc.experiments.show()[baseline_rev]
assert len(results) == 2
exp = results[exp_rev]["data"]
assert exp["name"] == "test_name"
assert exp["status"] == "Queued"
assert exp["params"]["params.yaml"] == {"data": {"foo": 2}}
results = dvc.experiments.show()
assert results[1].dumpd() == {
"rev": baseline_rev,
"name": "master",
"data": make_data(rev=baseline_rev),
"error": None,
"experiments": [
{
"revs": [
{
"rev": exp_rev,
"name": "test_name",
"data": make_data(rev=exp_rev, params={"foo": 2}, metrics=ANY),
"error": None,
"experiments": None,
}
],
"executor": make_executor(state="queued"),
"name": "test_name",
}
],
}
# test that only queued experiments for the current baseline are returned
tmp_dir.gen("foo", "foo")
@@ -148,99 +185,85 @@ def test_show_queued(tmp_dir, scm, dvc, exp_stage):
dvc.experiments.run(exp_stage.addressing, params=["foo=3"], queue=True)
exp_rev = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
results = dvc.experiments.show()[new_rev]
assert len(results) == 2
exp = results[exp_rev]["data"]
assert exp["status"] == "Queued"
assert exp["params"]["params.yaml"] == {"data": {"foo": 3}}
results = dvc.experiments.show()
assert results[1].dumpd() == {
"rev": new_rev,
"name": "master",
"data": make_data(rev=new_rev),
"error": None,
"experiments": [
{
"revs": [
{
"rev": exp_rev,
"name": ANY,
"data": make_data(rev=exp_rev, params={"foo": 3}, metrics=ANY),
"error": None,
"experiments": None,
}
],
"executor": make_executor(state="queued"),
"name": ANY,
}
],
}
@pytest.mark.vscode
def test_show_failed_experiment(tmp_dir, scm, dvc, failed_exp_stage, test_queue):
baseline_rev = scm.get_rev()
timestamp = datetime.fromtimestamp(
scm.gitpython.repo.rev_parse(baseline_rev).committed_date
)
dvc.experiments.run(
failed_exp_stage.addressing, params=["foo=2"], queue=True, name="show-failed"
)
dvc.experiments.run(failed_exp_stage.addressing, params=["foo=2"], queue=True)
exp_rev = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
dvc.experiments.run(run_all=True)
test_queue.wait(["show-failed"])
experiments = dvc.experiments.show()[baseline_rev]
expected_baseline = {
"data": {
"rev": ANY,
"deps": {
"copy.py": {
"hash": ANY,
"size": ANY,
"nfiles": None,
}
},
"metrics": {},
"outs": {},
"params": {"params.yaml": {"data": {"foo": 1}}},
"status": "Success",
"executor": None,
"timestamp": timestamp,
"name": "master",
}
results = dvc.experiments.show()
assert results[1].dumpd() == {
"rev": baseline_rev,
"name": "master",
"data": make_data(rev=baseline_rev, metrics=ANY),
"error": None,
"experiments": [
{
"revs": [
{
"rev": exp_rev,
"name": ANY,
"data": make_data(rev=exp_rev, params={"foo": 2}, metrics=ANY),
"error": {"msg": "Experiment run failed", "type": ANY},
"experiments": None,
}
],
"executor": make_executor(
state="failed",
local={"returncode": 255},
),
"name": ANY,
}
],
}
expected_failed = {
"data": {
"rev": ANY,
"timestamp": ANY,
"params": {"params.yaml": {"data": {"foo": 2}}},
"deps": {"copy.py": {"hash": None, "size": None, "nfiles": None}},
"metrics": {},
"outs": {},
"status": "Failed",
"executor": None,
"error": {
"msg": "Experiment run failed.",
"type": "",
},
"name": ANY,
}
}
assert len(experiments) == 2
for rev, exp in experiments.items():
if rev == "baseline":
assert exp == expected_baseline
else:
assert rev == exp_rev
assert exp == expected_failed
@pytest.mark.vscode
@pytest.mark.parametrize("workspace", [True, False])
def test_show_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, capsys, workspace):
baseline_rev = scm.get_rev()
results = dvc.experiments.run(
checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace
)
exp_rev = first(results)
results = dvc.experiments.show()[baseline_rev]
# Assert 4 rows: baseline, 2 checkpoints, and final commit
assert len(results) == checkpoint_stage.iterations + 2
checkpoints = []
for rev, exp in results.items():
if rev != "baseline":
checkpoints.append(rev)
assert exp["data"]["checkpoint_tip"] == exp_rev
baseline = dvc.experiments.show()[1]
assert baseline.data.meta.get("has_checkpoints")
checkpoints = first(baseline.experiments)
# 2 checkpoints + final commit
assert len(checkpoints) == checkpoint_stage.iterations + 1
assert first(checkpoints).rev == exp_rev
capsys.readouterr()
assert main(["exp", "show", "--no-pager"]) == 0
cap = capsys.readouterr()
for i, rev in enumerate(checkpoints):
for i, exp in enumerate(checkpoints):
rev = exp.rev
if i == 0:
name = dvc.experiments.get_exact_name([rev])[rev]
name = f"{rev[:7]} [{name}]"
@@ -254,44 +277,6 @@ def test_show_checkpoint(tmp_dir, scm, dvc, checkpoint_stage, capsys, workspace)
assert f"{fs} {name}" in cap.out
@pytest.mark.vscode
@pytest.mark.parametrize("workspace", [True, False])
def test_show_checkpoint_branch(tmp_dir, scm, dvc, checkpoint_stage, capsys, workspace):
results = dvc.experiments.run(
checkpoint_stage.addressing, params=["foo=2"], tmp_dir=not workspace
)
branch_rev = first(results)
if not workspace:
dvc.experiments.apply(branch_rev)
results = dvc.experiments.run(
checkpoint_stage.addressing,
checkpoint_resume=branch_rev,
tmp_dir=not workspace,
)
checkpoint_a = first(results)
dvc.experiments.apply(branch_rev)
results = dvc.experiments.run(
checkpoint_stage.addressing,
checkpoint_resume=branch_rev,
params=["foo=100"],
tmp_dir=not workspace,
)
checkpoint_b = first(results)
capsys.readouterr()
assert main(["exp", "show", "--no-pager"]) == 0
cap = capsys.readouterr()
for rev in (checkpoint_a, checkpoint_b):
ref = dvc.experiments.get_branch_by_rev(rev)
ref_info = ExpRefInfo.from_ref(ref)
name = f"{rev[:7]} [{ref_info.name}]"
assert f"╓ {name}" in cap.out
assert f"({branch_rev[:7]})" in cap.out
def test_show_filter(
tmp_dir,
scm,
@@ -373,19 +358,21 @@ def test_show_multiple_commits(tmp_dir, scm, dvc, exp_stage):
dvc.experiments.show(num=-2)
expected = {"workspace", init_rev, next_rev}
expected = ["workspace", next_rev, init_rev]
results = dvc.experiments.show(num=2)
assert set(results.keys()) == expected
assert [exp.rev for exp in results] == expected
expected = {"workspace"} | set(scm.branch_revs("master"))
expected = ["workspace", *scm.branch_revs("master")]
results = dvc.experiments.show(all_commits=True)
assert set(results.keys()) == expected
assert [exp.rev for exp in results] == expected
results = dvc.experiments.show(num=100)
assert set(results.keys()) == expected
assert [exp.rev for exp in results] == expected
def test_show_sort(tmp_dir, scm, dvc, exp_stage, caplog):
dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
with caplog.at_level(logging.ERROR):
assert main(["exp", "show", "--no-pager", "--sort-by=bar"]) != 0
assert "Unknown sort column" in caplog.text
@@ -408,14 +395,20 @@ def test_show_sort(tmp_dir, scm, dvc, exp_stage, caplog):
(TaskStatus.FAILED, False),
],
)
def test_show_running_workspace(
def test_show_running(
tmp_dir, scm, dvc, exp_stage, capsys, caplog, status, pid_exists, mocker
):
from dvc.rwlock import RWLOCK_FILE
from dvc_task.proc.process import ProcessInfo
baseline_rev = scm.get_rev()
pid_dir = os.path.join(dvc.tmp_dir, EXEC_TMP_DIR, EXEC_PID_DIR)
lock_file = relpath(os.path.join(dvc.tmp_dir, RWLOCK_FILE), str(tmp_dir))
info = make_executor_info(location=BaseExecutor.DEFAULT_LOCATION, status=status)
info = make_executor_info(
location=BaseExecutor.DEFAULT_LOCATION,
status=status,
baseline_rev=baseline_rev,
)
pidfile = os.path.join(
pid_dir,
"workspace",
@@ -425,195 +418,42 @@ def test_show_running_workspace(
(tmp_dir / pidfile).dump_json(info.asdict())
(tmp_dir / lock_file).dump_json(LOCK_CONTENTS)
mocker.patch.object(ProcessInfo, "load", return_value=mocker.Mock(pid=123))
mocker.patch("psutil.pid_exists", return_value=pid_exists)
assert dvc.experiments.show().get("workspace") == {
"baseline": {
"data": {
"rev": "workspace",
"deps": {
"copy.py": {
"hash": ANY,
"size": ANY,
"nfiles": None,
}
},
"metrics": {"metrics.yaml": {"data": {"foo": 1}}},
"params": {"params.yaml": {"data": {"foo": 1}}},
"outs": {},
"status": (
"Running"
if status == TaskStatus.RUNNING and pid_exists
else "Success"
),
"executor": (
info.location
if status == TaskStatus.RUNNING and pid_exists
else None
),
"timestamp": None,
tempdir_active = mocker.spy(dvc.experiments.tempdir_queue, "collect_active_data")
celery_active = mocker.spy(dvc.experiments.celery_queue, "collect_active_data")
results = dvc.experiments.show()
assert results[1].dumpd() == {
"rev": ANY,
"name": "master",
"data": make_data(),
"error": None,
"experiments": [
{
"revs": ANY,
"executor": make_executor(state="running"),
"name": ANY,
}
}
]
if pid_exists
else None,
}
capsys.readouterr()
assert main(["exp", "show", "--csv"]) == 0
cap = capsys.readouterr()
if status == TaskStatus.RUNNING:
if pid_exists:
assert "Running" in cap.out
assert info.location in cap.out
else:
cmd = LOCK_CONTENTS["read"]["data/MNIST"][0]["cmd"]
pid = LOCK_CONTENTS["read"]["data/MNIST"][0]["pid"]
assert (
f"Process '{cmd}' with (Pid {pid}), in RWLock-file "
f"'{lock_file}' had been killed." in caplog.text
)
assert f"Delete corrupted RWLock-file '{lock_file}'" in caplog.text
def test_show_running_tempdir(tmp_dir, scm, dvc, exp_stage, mocker):
baseline_rev = scm.get_rev()
run_results = dvc.experiments.run(
exp_stage.addressing, params=["foo=2"], tmp_dir=True
)
exp_rev = first(run_results)
exp_ref = first(exp_refs_by_rev(scm, exp_rev))
queue = dvc.experiments.tempdir_queue
stash_rev = "abc123"
entries = [
QueueEntry(
str(tmp_dir / ".dvc" / "tmp" / "foo"),
str(tmp_dir / ".dvc" / "tmp" / "foo"),
str(exp_ref),
stash_rev,
exp_ref.baseline_sha,
None,
exp_ref.name,
None,
)
]
mocker.patch.object(
dvc.experiments.tempdir_queue,
"iter_active",
return_value=entries,
)
info = make_executor_info(location=BaseExecutor.DEFAULT_LOCATION)
pidfile = queue.get_infofile_path(stash_rev)
os.makedirs(os.path.dirname(pidfile), exist_ok=True)
(tmp_dir / pidfile).dump_json(info.asdict())
mock_fetch = mocker.patch.object(
dvc.experiments.tempdir_queue,
"get_running_exps",
return_value={exp_rev: info.asdict()},
)
results = dvc.experiments.show()
mock_fetch.assert_has_calls(
[mocker.call(True)],
)
exp_data = get_in(results, [baseline_rev, exp_rev, "data"])
assert exp_data["status"] == "Running"
assert exp_data["executor"] == info.location
assert results["workspace"]["baseline"]["data"]["status"] == "Success"
def test_show_running_celery(tmp_dir, scm, dvc, exp_stage, mocker):
baseline_rev = scm.get_rev()
dvc.experiments.run(exp_stage.addressing, params=["foo=2"], queue=True)
exp_rev = dvc.experiments.scm.resolve_rev(f"{CELERY_STASH}@{{0}}")
queue = dvc.experiments.celery_queue
entries = list(queue.iter_queued())
mocker.patch.object(
dvc.experiments.celery_queue,
"iter_active",
return_value=entries,
)
info = make_executor_info(location=BaseExecutor.DEFAULT_LOCATION)
pidfile = queue.get_infofile_path(entries[0].stash_rev)
os.makedirs(os.path.dirname(pidfile), exist_ok=True)
(tmp_dir / pidfile).dump_json(info.asdict())
results = dvc.experiments.show()
exp_data = get_in(results, [baseline_rev, exp_rev, "data"])
assert exp_data["status"] == "Running"
assert exp_data["executor"] == info.location
assert results["workspace"]["baseline"]["data"]["status"] == "Success"
def test_show_running_checkpoint(
tmp_dir, scm, dvc, checkpoint_stage, mocker, test_queue
):
from dvc.repo.experiments.executor.local import TempDirExecutor
baseline_rev = scm.get_rev()
dvc.experiments.run(
checkpoint_stage.addressing, params=["foo=2"], queue=True, name="foo"
)
queue = dvc.experiments.celery_queue
entries = list(queue.iter_queued())
run_results = dvc.experiments.run(run_all=True)
test_queue.wait(["foo"])
checkpoint_rev = first(run_results)
exp_ref = first(exp_refs_by_rev(scm, checkpoint_rev))
mocker.patch.object(
dvc.experiments.celery_queue,
"iter_active",
return_value=entries,
)
mocker.patch.object(
dvc.experiments.celery_queue,
"iter_failed",
return_value=[],
)
pidfile = queue.get_infofile_path(entries[0].stash_rev)
info = make_executor_info(
git_url="foo.git",
baseline_rev=baseline_rev,
location=TempDirExecutor.DEFAULT_LOCATION,
status=TaskStatus.RUNNING,
)
os.makedirs(os.path.dirname(pidfile), exist_ok=True)
(tmp_dir / pidfile).dump_json(info.asdict())
mocker.patch.object(BaseExecutor, "fetch_exps", return_value=[str(exp_ref)])
results = dvc.experiments.show()
checkpoint_res = get_in(results, [baseline_rev, checkpoint_rev, "data"])
assert checkpoint_res["status"] == "Running"
assert checkpoint_res["executor"] == info.location
assert results["workspace"]["baseline"]["data"]["status"] == "Success"
tempdir_active.assert_called_once()
celery_active.assert_called_once()
def test_show_with_broken_repo(tmp_dir, scm, dvc, exp_stage, caplog):
baseline_rev = scm.get_rev()
exp1 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
exp2 = dvc.experiments.run(exp_stage.addressing, params=["foo=3"])
dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
with open("dvc.yaml", "a", encoding="utf-8") as fd:
fd.write("breaking the yaml!")
result = dvc.experiments.show()
rev1 = first(exp1)
rev2 = first(exp2)
results = dvc.experiments.show()
assert results[0].error
assert results[0].error.type == "YAMLSyntaxError"
baseline = result[baseline_rev]
paths = ["data", "params", "params.yaml"]
assert get_in(baseline[rev1], paths) == {"data": {"foo": 2}}
assert get_in(baseline[rev2], paths) == {"data": {"foo": 3}}
paths = ["workspace", "baseline", "error"]
assert isinstance(get_in(result, paths), (_CachedError, YAMLFileCorruptedError))
assert "YAML file structure is corrupted" in str(get_in(result, paths))
for exp_range in results[1].experiments:
assert not any(exp.error for exp in exp_range)
def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
@@ -646,8 +486,8 @@ def test_show_csv(tmp_dir, scm, dvc, exp_stage, capsys):
assert "metrics.yaml:foo,params.yaml:foo,copy.py" in cap.out
assert f",workspace,baseline,,,3,3,{data_hash}" in cap.out
assert (
"master,{},baseline,{},,1,1,{}".format(
baseline_rev[:7], _get_rev_isotimestamp(baseline_rev), data_hash
",master,baseline,{},,1,1,{}".format(
_get_rev_isotimestamp(baseline_rev), data_hash
)
in cap.out
)
@@ -810,35 +650,52 @@ def test_show_outs(tmp_dir, dvc, scm, erepo_dir, copy_script):
scm.commit("init")
outs = dvc.experiments.show()["workspace"]["baseline"]["data"]["outs"]
assert outs == {
"out": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": False,
}
results = dvc.experiments.show()
assert results[0].dumpd() == {
"rev": "workspace",
"name": None,
"data": make_data(
params=ANY,
outs={
"out": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": False,
}
},
),
"error": None,
"experiments": None,
}
tmp_dir.dvc_gen("out_add", "foo", commit="dvc add output")
outs = dvc.experiments.show()["workspace"]["baseline"]["data"]["outs"]
assert outs == {
"out": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": False,
},
"out_add": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": True,
},
results = dvc.experiments.show()
assert results[0].dumpd() == {
"rev": "workspace",
"name": None,
"data": make_data(
params=ANY,
outs={
"out": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": False,
},
"out_add": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": True,
},
},
),
"error": None,
"experiments": None,
}
with erepo_dir.chdir():
@@ -846,29 +703,38 @@ def test_show_outs(tmp_dir, dvc, scm, erepo_dir, copy_script):
dvc.imp(os.fspath(erepo_dir), "out", "out_imported")
outs = dvc.experiments.show()["workspace"]["baseline"]["data"]["outs"]
assert outs == {
"out": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": False,
},
"out_add": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": True,
},
"out_imported": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": True,
},
results = dvc.experiments.show()
assert results[0].dumpd() == {
"rev": "workspace",
"name": None,
"data": make_data(
params=ANY,
outs={
"out": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": False,
},
"out_add": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": True,
},
"out_imported": {
"hash": ANY,
"size": ANY,
"nfiles": None,
"use_cache": True,
"is_data_source": True,
},
},
),
"error": None,
"experiments": None,
}
@@ -927,12 +793,7 @@ def test_metrics_renaming(tmp_dir, dvc, scm, capsys, copy_script):
scm.gitpython.repo.rev_parse(rev).committed_date
).isoformat()
assert (
"master,{},baseline,{},,1,,1".format(
scores_rev[:7], _get_rev_isotimestamp(scores_rev)
)
in cap.out
)
assert f",master,baseline,{_get_rev_isotimestamp(scores_rev)},,1,,1" in cap.out
assert (
",{},baseline,{},,,1,1".format(
metrics_rev[:7], _get_rev_isotimestamp(metrics_rev)
@@ -961,8 +822,6 @@ def test_show_sorted_deps(tmp_dir, dvc, scm, capsys):
@pytest.mark.vscode
def test_show_queued_error(tmp_dir, scm, dvc, exp_stage, mocker):
baseline_rev = scm.get_rev()
dvc.experiments.run(
exp_stage.addressing, params=["foo=2"], queue=True, name="test_name"
)
@@ -983,17 +842,16 @@ def test_show_queued_error(tmp_dir, scm, dvc, exp_stage, mocker):
side_effect=mocker.MagicMock(side_effect=resolve_commit),
)
results = dvc.experiments.show()[baseline_rev]
results = dvc.experiments.show()[1].experiments
assert len(results) == 2
exp_2 = results[exp_rev_2]["data"]
assert exp_2["status"] == "Queued"
assert exp_2["params"]["params.yaml"] == {"data": {"foo": 2}}
queued = results[0]
assert queued.executor.state == "queued"
errored = results[1]
assert errored.revs[0].error
@pytest.mark.vscode
def test_show_completed_error(tmp_dir, scm, dvc, exp_stage, mocker):
baseline_rev = scm.get_rev()
result_2 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
exp_rev_2 = first(result_2)
commit_2 = scm.resolve_commit(exp_rev_2)
@@ -1010,49 +868,14 @@ def test_show_completed_error(tmp_dir, scm, dvc, exp_stage, mocker):
"resolve_commit",
side_effect=mocker.MagicMock(side_effect=resolve_commit),
)
experiments = dvc.experiments.show()[baseline_rev]
assert len(experiments) == 2
assert exp_rev_2 in experiments
@pytest.mark.vscode
def test_show_checkpoint_error(tmp_dir, scm, dvc, checkpoint_stage, mocker):
baseline_rev = scm.get_rev()
results = dvc.experiments.run(checkpoint_stage.addressing, params=["foo=2"])
exp_rev = first(results)
exp_ref = str(first(exp_refs_by_rev(scm, exp_rev)))
results = dvc.experiments.show()[baseline_rev]
# Assert 4 rows: baseline, 2 checkpoints, and final commit
assert len(results) == checkpoint_stage.iterations + 2
checkpoints = {}
for rev in results:
if rev != "baseline":
checkpoints[rev] = scm.resolve_commit(rev)
checkpoints[exp_ref] = scm.resolve_commit(exp_ref)
checkpoints[baseline_rev] = scm.resolve_commit(baseline_rev)
failed_rev = random.choice(list(checkpoints.keys()))
def resolve_commit(rev):
if rev == failed_rev:
raise SCMError
return checkpoints[rev]
mocker.patch.object(
scm,
"resolve_commit",
side_effect=mocker.MagicMock(side_effect=resolve_commit),
)
results = dvc.experiments.show(force=True)[baseline_rev]
results = dvc.experiments.show()[1].experiments
assert len(results) == 1
assert not results[0].revs[0].error
@pytest.mark.vscode
def test_show_baseline_error(tmp_dir, scm, dvc, exp_stage, mocker):
baseline_rev = scm.get_rev()
branch = scm.active_branch()
result_2 = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
exp_rev_2 = first(result_2)
@@ -1068,7 +891,7 @@ def test_show_baseline_error(tmp_dir, scm, dvc, exp_stage, mocker):
"resolve_commit",
side_effect=mocker.MagicMock(side_effect=resolve_commit),
)
experiments = dvc.experiments.show()[baseline_rev]
assert len(experiments) == 1
assert experiments["baseline"]["data"] == {"name": branch}
assert isinstance(experiments["baseline"]["error"], (SCMError, _CachedError))
results = dvc.experiments.show()
assert results[1].error
assert len(results[1].experiments) == 1
tests/unit/command/test_experiments.py
+1 -306
View File
@@ -1,7 +1,4 @@
import csv
import pathlib
import textwrap
from datetime import datetime
import pytest
@@ -18,7 +15,7 @@ from dvc.commands.experiments.push import CmdExperimentsPush
from dvc.commands.experiments.remove import CmdExperimentsRemove
from dvc.commands.experiments.run import CmdExperimentsRun
from dvc.commands.experiments.save import CmdExperimentsSave
from dvc.commands.experiments.show import CmdExperimentsShow, show_experiments
from dvc.commands.experiments.show import CmdExperimentsShow
from dvc.exceptions import InvalidArgumentError
from dvc.repo import Repo
from tests.utils import ANY
@@ -422,279 +419,6 @@ def test_experiments_remove_invalid(dvc, scm, mocker, capsys, caplog):
)
def test_show_experiments_csv(capsys):
all_experiments = {
"workspace": {
"baseline": {
"data": {
"timestamp": None,
"params": {
"params.yaml": {
"data": {
"featurize": {
"max_features": 3000,
"ngrams": 1,
},
"parent": 20170428,
"train": {
"n_est": 100,
"min_split": 36,
},
}
}
},
"status": "Success",
"executor": None,
"metrics": {
"scores.json": {
"data": {
"featurize": {
"max_features": 3000,
"ngrams": 1,
},
"avg_prec": 0.5843640011189556,
"roc_auc": 0.9544670443829399,
}
}
},
}
}
},
"b05eecc666734e899f79af228ff49a7ae5a18cc0": {
"baseline": {
"data": {
"timestamp": datetime(2021, 8, 2, 16, 48, 14),
"params": {
"params.yaml": {
"data": {
"featurize": {
"max_features": 3000,
"ngrams": 1,
},
"parent": 20170428,
"train": {
"n_est": 100,
"min_split": 2,
},
}
}
},
"status": "Success",
"executor": None,
"metrics": {
"scores.json": {
"data": {
"featurize": {
"max_features": 3000,
"ngrams": 1,
},
"avg_prec": 0.5325162867864254,
"roc_auc": 0.9106964878520005,
}
}
},
"name": "master",
}
},
"ae99936461d6c3092934160f8beafe66a294f98d": {
"data": {
"timestamp": datetime(2021, 8, 31, 14, 56, 55),
"params": {
"params.yaml": {
"data": {
"featurize": {
"max_features": 3000,
"ngrams": 1,
},
"parent": 20170428,
"train": {
"n_est": 100,
"min_split": 36,
},
}
}
},
"status": "Running",
"executor": None,
"metrics": {
"scores.json": {
"data": {
"featurize": {
"max_features": 3000,
"ngrams": 1,
},
"avg_prec": 0.5843640011189556,
"roc_auc": 0.9544670443829399,
}
}
},
"name": "exp-44136",
}
},
},
}
show_experiments(all_experiments, precision=None, fill_value="", iso=True, csv=True)
cap = capsys.readouterr()
assert (
"Experiment,rev,typ,Created,parent,State,scores.json:"
"featurize.max_features,scores.json:featurize.ngrams,"
"avg_prec,roc_auc,params.yaml:featurize.max_features,"
"params.yaml:featurize.ngrams,params.yaml:parent,"
"train.n_est,train.min_split" in cap.out
)
assert (
",workspace,baseline,,,,3000,1,0.5843640011189556,0.9544670443829399,"
"3000,1,20170428,100,36" in cap.out
)
assert (
"master,b05eecc,baseline,2021-08-02T16:48:14,,,3000,1,"
"0.5325162867864254,0.9106964878520005,3000,1,20170428,100,2" in cap.out
)
assert (
"exp-44136,ae99936,branch_base,2021-08-31T14:56:55,,Running,"
"3000,1,0.5843640011189556,0.9544670443829399,3000,1,20170428,100,36" in cap.out
)
def test_show_experiments_md(capsys):
all_experiments = {
"workspace": {
"baseline": {
"data": {
"timestamp": None,
"params": {"params.yaml": {"data": {"foo": 1}}},
"status": "Success",
"executor": None,
"metrics": {"scores.json": {"data": {"bar": 0.9544670443829399}}},
}
}
},
}
show_experiments(
all_experiments, precision=None, fill_value="", iso=True, markdown=True
)
cap = capsys.readouterr()
assert cap.out == textwrap.dedent(
"""\
| Experiment | Created | bar | foo |
|--------------|-----------|--------------------|-------|
| workspace | | 0.9544670443829399 | 1 |\n
"""
)
@pytest.mark.parametrize("sort_order", ["asc", "desc"])
def test_show_experiments_sort_by(capsys, sort_order):
sort_experiments = {
"workspace": {
"baseline": {
"data": {
"timestamp": None,
"params": {
"params.yaml": {
"data": {
"foo": 1,
}
}
},
"status": "Success",
"executor": None,
"metrics": {},
}
}
},
"233b132676792d89e848e5c9c12e408d7efde78a": {
"baseline": {
"data": {
"timestamp": datetime(2021, 8, 2, 16, 48, 14),
"params": {
"params.yaml": {
"data": {
"foo": 0,
}
}
},
"status": "Success",
"executor": None,
"metrics": {},
"name": "master",
}
},
"fad0a94": {
"data": {
"timestamp": datetime(2021, 8, 31, 14, 56, 55),
"params": {
"params.yaml": {
"data": {
"foo": 1,
}
}
},
"status": "Success",
"executor": None,
"metrics": {},
"name": "exp-89140",
}
},
"60fcda8": {
"data": {
"timestamp": datetime(2021, 8, 31, 14, 56, 55),
"params": {
"params.yaml": {
"data": {
"foo": 2,
}
}
},
"status": "Success",
"executor": None,
"metrics": {},
"name": "exp-43537",
}
},
"a7e9aaf": {
"data": {
"timestamp": datetime(2021, 8, 31, 14, 56, 55),
"params": {
"params.yaml": {
"data": {
"foo": 0,
}
}
},
"status": "Success",
"executor": None,
"metrics": {},
"name": "exp-4f89e",
}
},
},
}
show_experiments(
sort_experiments,
precision=None,
fill_value="",
iso=True,
csv=True,
sort_by="foo",
sort_order=sort_order,
)
cap = capsys.readouterr()
rows = list(csv.reader(cap.out.strip().split("\n")))
# [3:] To skip header, workspace and baseline(master)
# which are not affected by order
params = tuple(int(row[-1]) for row in rows[3:])
if sort_order == "asc":
assert params == (0, 1, 2)
else:
assert params == (2, 1, 0)
@pytest.mark.parametrize("extra_args", [(), ("--run",)])
def test_experiments_init(dvc, scm, mocker, capsys, extra_args):
stage = mocker.Mock(outs=[], addressing="train")
@@ -893,35 +617,6 @@ def test_experiments_init_displays_output_on_no_run(dvc, mocker, capsys, args):
assert out.splitlines() == expected_lines
def test_show_experiments_pcp(tmp_dir, mocker):
all_experiments = {
"workspace": {
"baseline": {
"data": {
"timestamp": None,
"params": {"params.yaml": {"data": {"foo": 1}}},
"status": "Success",
"executor": None,
"metrics": {"scores.json": {"data": {"bar": 0.9544670443829399}}},
}
}
},
}
experiments_table = mocker.patch("dvc.commands.experiments.show.experiments_table")
td = experiments_table.return_value
show_experiments(all_experiments, pcp=True)
td.drop.assert_called_with("Created")
td.dropna.assert_called_with("rows", how="all", subset=set())
td.drop_duplicates.assert_called_with("rows", subset=set())
kwargs = td.to_parallel_coordinates.call_args[1]
assert kwargs["output_path"] == str(tmp_dir / "dvc_plots" / "index.html")
assert kwargs["color_by"] == "Experiment"
def test_experiments_save(dvc, scm, mocker):
cli_args = parse_args(["exp", "save", "--name", "exp-name", "--force"])
assert cli_args.func == CmdExperimentsSave
tests/unit/repo/experiments/test_show.py
-217
View File
@@ -1,217 +0,0 @@
import pytest
from funcy import first
from dvc.repo.experiments.show import (
ExpStatus,
get_branch_names,
move_properties_to_head,
update_names,
)
def test_get_show_branch(tmp_dir, scm, dvc, exp_stage):
new_branch = "new"
baseline = scm.get_rev()
base_branch = scm.active_branch()
results = dvc.experiments.run(exp_stage.addressing, params=["foo=2"])
exp_a = first(results)
dvc.experiments.branch(exp_a, new_branch)
scm.checkout(new_branch, force=True)
result = {
"workspace": {"baseline": {"data": {}}},
exp_a: {"baseline": {"data": {}}},
baseline: {"baseline": {"data": {}}, exp_a: {"data": {}}},
}
branch_names = get_branch_names(scm, ["workspace", exp_a, baseline])
assert branch_names == {exp_a: new_branch, baseline: base_branch}
update_names(dvc, branch_names, result)
assert result[exp_a]["baseline"]["data"] == {"name": new_branch}
assert result[baseline]["baseline"]["data"] == {"name": base_branch}
assert result[baseline][exp_a]["data"] == {"name": new_branch}
@pytest.mark.parametrize(
"res_in,res_out",
[
(
{
"1": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
}
},
"2": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "a",
}
},
"3": {
"data": {
"checkpoint_tip": "3",
"executor": None,
"status": ExpStatus.Success.name,
}
},
"4": {
"data": {
"checkpoint_tip": "3",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "b",
}
},
},
{
"1": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "a",
}
},
"2": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
}
},
"3": {
"data": {
"checkpoint_tip": "3",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "b",
}
},
"4": {
"data": {
"checkpoint_tip": "3",
"executor": None,
"status": ExpStatus.Success.name,
}
},
},
),
(
{
"0": {"error": {"msg": "something"}},
"1": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
}
},
"2": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "a",
}
},
"3": {"data": {"name": "b"}},
},
{
"0": {"error": {"msg": "something"}},
"1": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "a",
}
},
"2": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
}
},
"3": {"data": {"name": "b"}},
},
),
(
{
"1": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
}
},
"2": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "a",
}
},
"3": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "b",
}
},
"4": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
}
},
},
{
"1": {
"data": {
"checkpoint_tip": "1",
"executor": "local",
"status": ExpStatus.Running.name,
"name": "a",
}
},
"2": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
}
},
"3": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
"name": "b",
}
},
"4": {
"data": {
"checkpoint_tip": "1",
"executor": None,
"status": ExpStatus.Success.name,
}
},
},
),
],
)
def test_move_properties_to_head(res_in, res_out):
move_properties_to_head({"baseline": res_in})
assert res_in == res_out