diff --git a/dvc/commands/experiments/show.py b/dvc/commands/experiments/show.py index 555f12e93..b94baf5ce 100644 --- a/dvc/commands/experiments/show.py +++ b/dvc/commands/experiments/show.py @@ -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, diff --git a/dvc/commands/queue/status.py b/dvc/commands/queue/status.py index d235d986a..6d1799fc2 100644 --- a/dvc/commands/queue/status.py +++ b/dvc/commands/queue/status.py @@ -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) diff --git a/dvc/repo/experiments/diff.py b/dvc/repo/experiments/diff.py index 0cda9bf38..15f21e73c 100644 --- a/dvc/repo/experiments/diff.py +++ b/dvc/repo/experiments/diff.py @@ -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"] diff --git a/dvc/repo/experiments/show.py b/dvc/repo/experiments/show.py index 767fa56b4..62c15f160 100644 --- a/dvc/repo/experiments/show.py +++ b/dvc/repo/experiments/show.py @@ -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] + ] diff --git a/tests/func/experiments/test_checkpoints.py b/tests/func/experiments/test_checkpoints.py index 302ffbfbe..e548efd87 100644 --- a/tests/func/experiments/test_checkpoints.py +++ b/tests/func/experiments/test_checkpoints.py @@ -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 diff --git a/tests/func/experiments/test_show.py b/tests/func/experiments/test_show.py index 7c98fa957..0753b349c 100644 --- a/tests/func/experiments/test_show.py +++ b/tests/func/experiments/test_show.py @@ -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 diff --git a/tests/unit/command/test_experiments.py b/tests/unit/command/test_experiments.py index fac882e24..5756538fe 100644 --- a/tests/unit/command/test_experiments.py +++ b/tests/unit/command/test_experiments.py @@ -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 diff --git a/tests/unit/repo/experiments/test_show.py b/tests/unit/repo/experiments/test_show.py deleted file mode 100644 index 2293e043b..000000000 --- a/tests/unit/repo/experiments/test_show.py +++ /dev/null @@ -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