dvc/dvc/project.py

from __future__ import unicode_literals

from dvc.utils.compat import str, builtin_str

import collections
import os
import dvc.prompt as prompt
import dvc.logger as logger

from dvc.exceptions import (
    DvcException,
    MoveNotDataSourceError,
    NotDvcProjectError,
)


class InitError(DvcException):
    def __init__(self, msg):
        super(InitError, self).__init__(msg)


class ReproductionError(DvcException):
    def __init__(self, dvc_file_name, ex):
        self.path = dvc_file_name
        msg = "failed to reproduce '{}'".format(dvc_file_name)
        super(ReproductionError, self).__init__(msg, cause=ex)


class Project(object):
    DVC_DIR = ".dvc"

    def __init__(self, root_dir=None):
        from dvc.config import Config
        from dvc.state import State
        from dvc.lock import Lock
        from dvc.scm import SCM
        from dvc.cache import Cache
        from dvc.data_cloud import DataCloud
        from dvc.updater import Updater

        root_dir = self.find_root(root_dir)

        self.root_dir = os.path.abspath(os.path.realpath(root_dir))
        self.dvc_dir = os.path.join(self.root_dir, self.DVC_DIR)

        self.config = Config(self.dvc_dir)

        self.scm = SCM(self.root_dir, project=self)
        self.lock = Lock(self.dvc_dir)
        # NOTE: storing state and link_state in the repository itself to avoid
        # any possible state corruption in 'shared cache dir' scenario.
        self.state = State(self, self.config.config)

        core = self.config.config[Config.SECTION_CORE]

        logger.set_level(core.get(Config.SECTION_CORE_LOGLEVEL))

        self.cache = Cache(self)
        self.cloud = DataCloud(self, config=self.config.config)
        self.updater = Updater(self.dvc_dir)

        self.files_to_git_add = []

        self._ignore()

        self.updater.check()

    def __repr__(self):
        return "Project: '{root_dir}'".format(root_dir=self.root_dir)

    @staticmethod
    def find_root(root=None):
        if root is None:
            root = os.getcwd()
        else:
            root = os.path.abspath(os.path.realpath(root))

        while True:
            dvc_dir = os.path.join(root, Project.DVC_DIR)
            if os.path.isdir(dvc_dir):
                return root
            if os.path.ismount(root):
                break
            root = os.path.dirname(root)
        raise NotDvcProjectError(root)

    @staticmethod
    def find_dvc_dir(root=None):
        root_dir = Project.find_root(root)
        return os.path.join(root_dir, Project.DVC_DIR)

    def _remind_to_git_add(self):
        if not self.files_to_git_add:
            return

        logger.info(
            "\n"
            "To track the changes with git run:\n"
            "\n"
            "\tgit add {files}".format(files=" ".join(self.files_to_git_add))
        )

    @staticmethod
    def init(root_dir=os.curdir, no_scm=False, force=False):
        """
        Creates an empty project on the given directory -- basically a
        `.dvc` directory with subdirectories for configuration and cache.

        It should be tracked by a SCM or use the `--no-scm` flag.

        If the given directory is not empty, you must use the `--force`
        flag to override it.

        Args:
            root_dir: Path to project's root directory.

        Returns:
            Project instance.

        Raises:
            KeyError: Raises an exception.
        """
        import shutil
        from dvc.scm import SCM, Base
        from dvc.config import Config

        root_dir = os.path.abspath(root_dir)
        dvc_dir = os.path.join(root_dir, Project.DVC_DIR)
        scm = SCM(root_dir)
        if type(scm) == Base and not no_scm:
            raise InitError(
                "{project} is not tracked by any supported scm tool"
                " (e.g. git). Use '--no-scm' if you don't want to use any scm.".format(
                    project=root_dir
                )
            )

        if os.path.isdir(dvc_dir):
            if not force:
                raise InitError(
                    "'{project}' exists. Use '-f' to force.".format(
                        project=os.path.relpath(dvc_dir)
                    )
                )

            shutil.rmtree(dvc_dir)

        os.mkdir(dvc_dir)

        config = Config.init(dvc_dir)
        proj = Project(root_dir)

        scm.add([config.config_file])

        if scm.ignore_file:
            scm.add([os.path.join(dvc_dir, scm.ignore_file)])
            logger.info("\nYou can now commit the changes to git.\n")

        proj._welcome_message()

        return proj

    def destroy(self):
        import shutil

        for stage in self.stages():
            stage.remove()

        shutil.rmtree(self.dvc_dir)

    def _ignore(self):
        flist = [
            self.state.state_file,
            self.lock.lock_file,
            self.config.config_local_file,
            self.updater.updater_file,
            self.updater.lock.lock_file,
        ] + self.state.temp_files

        if self.cache.local.cache_dir.startswith(self.root_dir):
            flist += [self.cache.local.cache_dir]

        self.scm.ignore_list(flist)

    def install(self):
        self.scm.install()

    @staticmethod
    def _check_cyclic_graph(graph):
        import networkx as nx
        from dvc.exceptions import CyclicGraphError

        cycles = list(nx.simple_cycles(graph))

        if cycles:
            raise CyclicGraphError(cycles[0])

    def add(self, fname, recursive=False):
        from dvc.stage import Stage

        fnames = []
        if recursive and os.path.isdir(fname):
            for root, dirs, files in os.walk(fname):
                for f in files:
                    path = os.path.join(root, f)
                    if Stage.is_stage_file(path):
                        continue
                    if os.path.basename(path) == self.scm.ignore_file:
                        continue
                    if self.scm.is_tracked(path):
                        continue
                    fnames.append(path)
        else:
            fnames = [fname]

        stages = []
        self.files_to_git_add = []
        with self.state:
            for f in fnames:
                stage = Stage.create(project=self, outs=[f], add=True)

                if stage is None:
                    stages.append(stage)
                    continue

                stage.save()
                stages.append(stage)

        self._check_dag(self.stages() + stages)

        for stage in stages:
            stage.dump()

        self._remind_to_git_add()

        return stages

    def _check_dag(self, stages):
        """Generate graph including the new stage to check for errors"""
        self.graph(stages=stages)

    def remove(self, target, outs_only=False):
        from dvc.stage import Stage

        stage = Stage.load(self, target)
        if outs_only:
            stage.remove_outs()
        else:
            stage.remove()

        return stage

    def lock_stage(self, target, unlock=False):
        from dvc.stage import Stage

        stage = Stage.load(self, target)
        stage.locked = False if unlock else True
        stage.dump()

        return stage

    def move(self, from_path, to_path):
        """
        Renames an output file and modifies the stage associated
        to reflect the change on the pipeline.

        If the output has the same name as its stage, it would
        also rename the corresponding stage file.

        E.g.
              Having: (hello, hello.dvc)

              $ dvc move hello greetings

              Result: (greeting, greeting.dvc)

        It only works with outputs generated by `add` or `import`,
        also known as data sources.
        """
        import dvc.output as Output
        from dvc.stage import Stage

        from_out = Output.loads_from(Stage(self, cwd=os.curdir), [from_path])[
            0
        ]

        to_path = self._expand_target_path(from_path, to_path)

        try:
            stage, out = next(
                (stage, out)
                for stage in self.stages()
                for out in stage.outs
                if from_out.path == out.path
            )
        except StopIteration:
            raise DvcException(
                "unable to find stage file with output '{path}'".format(
                    path=from_path
                )
            )

        if not stage.is_data_source:
            raise MoveNotDataSourceError(stage.relpath)

        stage_name = os.path.splitext(os.path.basename(stage.path))[0]
        from_name = os.path.basename(from_out.path)
        if stage_name == from_name:
            os.unlink(stage.path)

            stage.path = os.path.join(
                os.path.dirname(to_path),
                os.path.basename(to_path) + Stage.STAGE_FILE_SUFFIX,
            )

            stage.cwd = os.path.join(self.root_dir, os.path.dirname(to_path))

        to_out = Output.loads_from(
            stage, [os.path.basename(to_path)], out.cache, out.metric
        )[0]

        with self.state:
            out.move(to_out)

        stage.dump()

        self._remind_to_git_add()

    def _unprotect_file(self, path):
        import stat
        import uuid
        from dvc.system import System
        from dvc.utils import copyfile, move, remove

        if System.is_symlink(path) or System.is_hardlink(path):
            logger.debug("Unprotecting '{}'".format(path))

            tmp = os.path.join(os.path.dirname(path), "." + str(uuid.uuid4()))
            move(path, tmp)

            copyfile(tmp, path)

            remove(tmp)
        else:
            logger.debug(
                "Skipping copying for '{}', since it is not "
                "a symlink or a hardlink.".format(path)
            )

        os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE)

    def _unprotect_dir(self, path):
        for root, dirs, files in os.walk(path):
            for f in files:
                path = os.path.join(root, f)
                self._unprotect_file(path)

    def unprotect(self, path):
        if not os.path.exists(path):
            raise DvcException(
                "can't unprotect non-existing data '{}'".format(path)
            )

        if os.path.isdir(path):
            self._unprotect_dir(path)
        else:
            self._unprotect_file(path)

    def run(
        self,
        cmd=None,
        deps=[],
        outs=[],
        outs_no_cache=[],
        metrics_no_cache=[],
        fname=None,
        cwd=os.curdir,
        no_exec=False,
        overwrite=False,
        ignore_build_cache=False,
        remove_outs=False,
    ):
        from dvc.stage import Stage

        with self.state:
            stage = Stage.create(
                project=self,
                fname=fname,
                cmd=cmd,
                cwd=cwd,
                outs=outs,
                outs_no_cache=outs_no_cache,
                metrics_no_cache=metrics_no_cache,
                deps=deps,
                overwrite=overwrite,
                ignore_build_cache=ignore_build_cache,
                remove_outs=remove_outs,
            )

        if stage is None:
            return None

        self._check_dag(self.stages() + [stage])

        self.files_to_git_add = []
        with self.state:
            if not no_exec:
                stage.run()

        stage.dump()

        self._remind_to_git_add()

        return stage

    def imp(self, url, out, resume=False):
        from dvc.stage import Stage

        stage = Stage.create(project=self, cmd=None, deps=[url], outs=[out])

        if stage is None:
            return None

        self._check_dag(self.stages() + [stage])

        self.files_to_git_add = []
        with self.state:
            stage.run(resume=resume)

        stage.dump()

        self._remind_to_git_add()

        return stage

    def _reproduce_stage(self, stages, node, force, dry, interactive):
        stage = stages[node]

        if stage.locked:
            logger.warning(
                "DVC file '{path}' is locked. Its dependencies are"
                " not going to be reproduced.".format(path=stage.relpath)
            )

        stage = stage.reproduce(force=force, dry=dry, interactive=interactive)
        if not stage:
            return []

        if not dry:
            stage.dump()

        return [stage]

    def reproduce(
        self,
        target=None,
        recursive=True,
        force=False,
        dry=False,
        interactive=False,
        pipeline=False,
        all_pipelines=False,
        ignore_build_cache=False,
    ):
        from dvc.stage import Stage

        if not target and not all_pipelines:
            raise ValueError()

        if not interactive:
            config = self.config
            core = config.config[config.SECTION_CORE]
            interactive = core.get(config.SECTION_CORE_INTERACTIVE, False)

        targets = []
        if pipeline or all_pipelines:
            if pipeline:
                stage = Stage.load(self, target)
                node = os.path.relpath(stage.path, self.root_dir)
                pipelines = [self._get_pipeline(node)]
            else:
                pipelines = self.pipelines()

            for G in pipelines:
                for node in G.nodes():
                    if G.in_degree(node) == 0:
                        targets.append(os.path.join(self.root_dir, node))
        else:
            targets.append(target)

        self.files_to_git_add = []

        ret = []
        with self.state:
            for target in targets:
                stages = self._reproduce(
                    target,
                    recursive=recursive,
                    force=force,
                    dry=dry,
                    interactive=interactive,
                    ignore_build_cache=ignore_build_cache,
                )
                ret.extend(stages)

        self._remind_to_git_add()

        return ret

    def _reproduce(
        self,
        target,
        recursive=True,
        force=False,
        dry=False,
        interactive=False,
        ignore_build_cache=False,
    ):
        import networkx as nx
        from dvc.stage import Stage

        stage = Stage.load(self, target)
        G = self.graph()[1]
        stages = nx.get_node_attributes(G, "stage")
        node = os.path.relpath(stage.path, self.root_dir)

        if recursive:
            ret = self._reproduce_stages(
                G, stages, node, force, dry, interactive, ignore_build_cache
            )
        else:
            ret = self._reproduce_stage(stages, node, force, dry, interactive)

        return ret

    def _reproduce_stages(
        self, G, stages, node, force, dry, interactive, ignore_build_cache
    ):
        import networkx as nx

        result = []
        for n in nx.dfs_postorder_nodes(G, node):
            try:
                ret = self._reproduce_stage(stages, n, force, dry, interactive)

                if len(ret) == 0 and ignore_build_cache:
                    # NOTE: we are walking our pipeline from the top to the
                    # bottom. If one stage is changed, it will be reproduced,
                    # which tells us that we should force reproducing all of
                    # the other stages down below, even if their direct
                    # dependencies didn't change.
                    force = True

                result += ret
            except Exception as ex:
                raise ReproductionError(stages[n].relpath, ex)
        return result

    def _cleanup_unused_links(self, all_stages):
        used = []
        for stage in all_stages:
            for out in stage.outs:
                used.append(out.path)
        self.state.remove_unused_links(used)

    def checkout(
        self, target=None, with_deps=False, force=False, recursive=False
    ):

        if target and not recursive:
            all_stages = self.active_stages()
            stages = self._collect(target, with_deps=with_deps)
        else:
            all_stages = self.active_stages(target)
            stages = all_stages

        with self.state:
            self._cleanup_unused_links(all_stages)

            for stage in stages:
                if stage.locked:
                    logger.warning(
                        "DVC file '{path}' is locked. Its dependencies are"
                        " not going to be checked out.".format(
                            path=stage.relpath
                        )
                    )

                stage.checkout(force=force)

    def _get_pipeline(self, node):
        pipelines = list(filter(lambda g: node in g.nodes(), self.pipelines()))
        assert len(pipelines) == 1
        return pipelines[0]

    def _collect(self, target, with_deps=False):
        import networkx as nx
        from dvc.stage import Stage

        stage = Stage.load(self, target)
        if not with_deps:
            return [stage]

        node = os.path.relpath(stage.path, self.root_dir)
        G = self._get_pipeline(node)
        stages = nx.get_node_attributes(G, "stage")

        ret = [stage]
        for n in nx.dfs_postorder_nodes(G, node):
            ret.append(stages[n])

        return ret

    def _collect_dir_cache(
        self, out, branch=None, remote=None, force=False, jobs=None
    ):
        info = out.dumpd()
        ret = [info]
        r = out.remote
        md5 = info[r.PARAM_CHECKSUM]

        if self.cache.local.changed_cache_file(md5):
            try:
                self.cloud.pull(
                    ret, jobs=jobs, remote=remote, show_checksums=False
                )
            except DvcException as exc:
                msg = "Failed to pull cache for '{}': {}"
                logger.debug(msg.format(out, exc))

        if self.cache.local.changed_cache_file(md5):
            msg = (
                "Missing cache for directory '{}'. "
                "Cache for files inside will be lost. "
                "Would you like to continue? Use '-f' to force. "
            )
            if not force and not prompt.confirm(msg):
                raise DvcException(
                    "unable to fully collect used cache"
                    " without cache for directory '{}'".format(out)
                )
            else:
                return ret

        for i in self.cache.local.load_dir_cache(md5):
            i["branch"] = branch
            i[r.PARAM_PATH] = os.path.join(
                info[r.PARAM_PATH], i[r.PARAM_RELPATH]
            )
            ret.append(i)

        return ret

    def _collect_used_cache(
        self, out, branch=None, remote=None, force=False, jobs=None
    ):
        if not out.use_cache or not out.info:
            if not out.info:
                logger.warning(
                    "Output '{}'({}) is missing version "
                    "info. Cache for it will not be collected. "
                    "Use dvc repro to get your pipeline up to "
                    "date.".format(out, out.stage)
                )
            return []

        info = out.dumpd()
        info["branch"] = branch
        ret = [info]

        if out.scheme != "local":
            return ret

        md5 = info[out.remote.PARAM_CHECKSUM]
        cache = self.cache.local.get(md5)
        if not out.remote.is_dir_cache(cache):
            return ret

        return self._collect_dir_cache(
            out, branch=branch, remote=remote, force=force, jobs=jobs
        )

    def _used_cache(
        self,
        target=None,
        all_branches=False,
        active=True,
        with_deps=False,
        all_tags=False,
        remote=None,
        force=False,
        jobs=None,
        recursive=False,
    ):
        cache = {}
        cache["local"] = []
        cache["s3"] = []
        cache["gs"] = []
        cache["hdfs"] = []
        cache["ssh"] = []
        cache["azure"] = []

        for branch in self.scm.brancher(
            all_branches=all_branches, all_tags=all_tags
        ):
            if target:
                if recursive:
                    stages = self.stages(target)
                else:
                    stages = self._collect(target, with_deps=with_deps)
            elif active:
                stages = self.active_stages()
            else:
                stages = self.stages()

            for stage in stages:
                if active and not target and stage.locked:
                    logger.warning(
                        "DVC file '{path}' is locked. Its dependencies are"
                        " not going to be pushed/pulled/fetched.".format(
                            path=stage.relpath
                        )
                    )

                for out in stage.outs:
                    scheme = out.path_info["scheme"]
                    cache[scheme] += self._collect_used_cache(
                        out,
                        branch=branch,
                        remote=remote,
                        force=force,
                        jobs=jobs,
                    )

        return cache

    @staticmethod
    def merge_cache_lists(clists):
        merged_cache = collections.defaultdict(list)

        for cache_list in clists:
            for scheme, cache in cache_list.items():
                for item in cache:
                    if item not in merged_cache[scheme]:
                        merged_cache[scheme].append(item)

        return merged_cache

    @staticmethod
    def load_all_used_cache(
        projects,
        target=None,
        all_branches=False,
        active=True,
        with_deps=False,
        all_tags=False,
        remote=None,
        force=False,
        jobs=None,
    ):
        clists = []

        for project in projects:
            with project.state:
                project_clist = project._used_cache(
                    target=None,
                    all_branches=all_branches,
                    active=False,
                    with_deps=with_deps,
                    all_tags=all_tags,
                    remote=remote,
                    force=force,
                    jobs=jobs,
                )

                clists.append(project_clist)

        return clists

    def _do_gc(self, typ, func, clist):
        removed = func(clist)
        if not removed:
            logger.info("No unused {} cache to remove.".format(typ))

    def gc(
        self,
        all_branches=False,
        cloud=False,
        remote=None,
        with_deps=False,
        all_tags=False,
        force=False,
        jobs=None,
        projects=None,
    ):

        all_projects = [self]

        if projects:
            all_projects.extend(Project(path) for path in projects)

        all_clists = Project.load_all_used_cache(
            all_projects,
            target=None,
            all_branches=all_branches,
            active=False,
            with_deps=with_deps,
            all_tags=all_tags,
            remote=remote,
            force=force,
            jobs=jobs,
        )

        if len(all_clists) > 1:
            clist = Project.merge_cache_lists(all_clists)
        else:
            clist = all_clists[0]

        with self.state:
            self._do_gc("local", self.cache.local.gc, clist)

            if self.cache.s3:
                self._do_gc("s3", self.cache.s3.gc, clist)

            if self.cache.gs:
                self._do_gc("gs", self.cache.gs.gc, clist)

            if self.cache.ssh:
                self._do_gc("ssh", self.cache.ssh.gc, clist)

            if self.cache.hdfs:
                self._do_gc("hdfs", self.cache.hdfs.gc, clist)

            if self.cache.azure:
                self._do_gc("azure", self.cache.azure.gc, clist)

            if cloud:
                self._do_gc(
                    "remote", self.cloud._get_cloud(remote, "gc -c").gc, clist
                )

    def push(
        self,
        target=None,
        jobs=1,
        remote=None,
        all_branches=False,
        show_checksums=False,
        with_deps=False,
        all_tags=False,
        recursive=False,
    ):
        with self.state:
            used = self._used_cache(
                target,
                all_branches=all_branches,
                all_tags=all_tags,
                with_deps=with_deps,
                force=True,
                remote=remote,
                jobs=jobs,
                recursive=recursive,
            )["local"]
            self.cloud.push(
                used, jobs, remote=remote, show_checksums=show_checksums
            )

    def fetch(
        self,
        target=None,
        jobs=1,
        remote=None,
        all_branches=False,
        show_checksums=False,
        with_deps=False,
        all_tags=False,
        recursive=False,
    ):
        with self.state:
            used = self._used_cache(
                target,
                all_branches=all_branches,
                all_tags=all_tags,
                with_deps=with_deps,
                force=True,
                remote=remote,
                jobs=jobs,
                recursive=recursive,
            )["local"]
            self.cloud.pull(
                used, jobs, remote=remote, show_checksums=show_checksums
            )

    def pull(
        self,
        target=None,
        jobs=1,
        remote=None,
        all_branches=False,
        show_checksums=False,
        with_deps=False,
        all_tags=False,
        force=False,
        recursive=False,
    ):
        self.fetch(
            target,
            jobs,
            remote=remote,
            all_branches=all_branches,
            all_tags=all_tags,
            show_checksums=show_checksums,
            with_deps=with_deps,
            recursive=recursive,
        )
        self.checkout(
            target=target,
            with_deps=with_deps,
            force=force,
            recursive=recursive,
        )

    def _local_status(self, target=None, with_deps=False):
        status = {}

        if target:
            stages = self._collect(target, with_deps=with_deps)
        else:
            stages = self.active_stages()

        for stage in stages:
            if stage.locked:
                logger.warning(
                    "DVC file '{path}' is locked. Its dependencies are"
                    " not going to be shown in the status output.".format(
                        path=stage.relpath
                    )
                )

            status.update(stage.status())

        return status

    def _cloud_status(
        self,
        target=None,
        jobs=1,
        remote=None,
        show_checksums=False,
        all_branches=False,
        with_deps=False,
        all_tags=False,
    ):
        import dvc.remote.base as cloud

        used = self._used_cache(
            target,
            all_branches=all_branches,
            all_tags=all_tags,
            with_deps=with_deps,
            force=True,
            remote=remote,
            jobs=jobs,
        )["local"]

        ret = {}
        status_info = self.cloud.status(
            used, jobs, remote=remote, show_checksums=show_checksums
        )
        for md5, info in status_info.items():
            name = info["name"]
            status = info["status"]
            if status == cloud.STATUS_OK:
                continue

            prefix_map = {
                cloud.STATUS_DELETED: "deleted",
                cloud.STATUS_NEW: "new",
            }

            ret[name] = prefix_map[status]

        return ret

    def status(
        self,
        target=None,
        jobs=1,
        cloud=False,
        remote=None,
        show_checksums=False,
        all_branches=False,
        with_deps=False,
        all_tags=False,
    ):
        with self.state:
            if cloud:
                return self._cloud_status(
                    target,
                    jobs,
                    remote=remote,
                    show_checksums=show_checksums,
                    all_branches=all_branches,
                    with_deps=with_deps,
                    all_tags=all_tags,
                )
            return self._local_status(target, with_deps=with_deps)

    def _read_metric_json(self, fd, json_path):
        import json
        from jsonpath_rw import parse

        parser = parse(json_path)
        return [x.value for x in parser.find(json.load(fd))]

    def _do_read_metric_xsv(self, reader, row, col):
        if col is not None and row is not None:
            return [reader[row][col]]
        elif col is not None:
            return [r[col] for r in reader]
        elif row is not None:
            return reader[row]
        return None

    def _read_metric_hxsv(self, fd, hxsv_path, delimiter):
        import csv

        col, row = hxsv_path.split(",")
        row = int(row)
        reader = list(csv.DictReader(fd, delimiter=builtin_str(delimiter)))
        return self._do_read_metric_xsv(reader, row, col)

    def _read_metric_xsv(self, fd, xsv_path, delimiter):
        import csv

        col, row = xsv_path.split(",")
        row = int(row)
        col = int(col)
        reader = list(csv.reader(fd, delimiter=builtin_str(delimiter)))
        return self._do_read_metric_xsv(reader, row, col)

    def _read_metric(self, path, typ=None, xpath=None):
        ret = None

        if not os.path.exists(path):
            return ret

        try:
            with open(path, "r") as fd:
                if typ == "json":
                    ret = self._read_metric_json(fd, xpath)
                elif typ == "csv":
                    ret = self._read_metric_xsv(fd, xpath, ",")
                elif typ == "tsv":
                    ret = self._read_metric_xsv(fd, xpath, "\t")
                elif typ == "hcsv":
                    ret = self._read_metric_hxsv(fd, xpath, ",")
                elif typ == "htsv":
                    ret = self._read_metric_hxsv(fd, xpath, "\t")
                else:
                    ret = fd.read().strip()
        except Exception:
            logger.error("unable to read metric in '{}'".format(path))

        return ret

    def _find_output_by_path(self, path, outs=None, recursive=False):
        from dvc.exceptions import OutputDuplicationError

        if not outs:
            astages = self.active_stages()
            outs = [out for stage in astages for out in stage.outs]

        abs_path = os.path.abspath(path)
        if os.path.isdir(abs_path) and recursive:
            matched = [
                out
                for out in outs
                if os.path.abspath(out.path).startswith(abs_path)
            ]
        else:
            matched = [out for out in outs if out.path == abs_path]
            stages = [out.stage.relpath for out in matched]
            if len(stages) > 1:
                raise OutputDuplicationError(path, stages)

        return matched if matched else []

    def metrics_show(
        self,
        path=None,
        typ=None,
        xpath=None,
        all_branches=False,
        all_tags=False,
        recursive=False,
    ):
        res = {}
        for branch in self.scm.brancher(
            all_branches=all_branches, all_tags=all_tags
        ):

            astages = self.active_stages()
            outs = [out for stage in astages for out in stage.outs]

            if path:
                outs = self._find_output_by_path(
                    path, outs=outs, recursive=recursive
                )
                stages = [out.stage.path for out in outs]
                entries = []
                for out in outs:
                    if all(
                        [out.metric, not typ, isinstance(out.metric, dict)]
                    ):
                        entries += [
                            (
                                out.path,
                                out.metric.get(out.PARAM_METRIC_TYPE, None),
                                out.metric.get(out.PARAM_METRIC_XPATH, None),
                            )
                        ]
                    else:
                        entries += [(out.path, typ, xpath)]
                if not entries:
                    if os.path.isdir(path):
                        logger.warning(
                            "Path '{path}' is a directory. "
                            "Consider runnig with '-R'.".format(path=path)
                        )
                        return {}

                    else:
                        entries += [(path, typ, xpath)]

            else:
                metrics = filter(lambda o: o.metric, outs)
                stages = None
                entries = []
                for o in metrics:
                    if not typ and isinstance(o.metric, dict):
                        t = o.metric.get(o.PARAM_METRIC_TYPE, typ)
                        x = o.metric.get(o.PARAM_METRIC_XPATH, xpath)
                    else:
                        t = typ
                        x = xpath
                    entries.append((o.path, t, x))

            for fname, t, x in entries:
                if stages:
                    for stage in stages:
                        self.checkout(stage, force=True)

                rel = os.path.relpath(fname)
                metric = self._read_metric(fname, typ=t, xpath=x)
                if not metric:
                    continue

                if branch not in res:
                    res[branch] = {}

                res[branch][rel] = metric

        for branch, val in res.items():
            if all_branches or all_tags:
                logger.info("{}:".format(branch))
            for fname, metric in val.items():
                logger.info("\t{}: {}".format(fname, metric))

        if res:
            return res

        if path and os.path.isdir(path):
            msg = "file '{}' does not exist".format(path)
            return res

        if path:
            msg = "file '{}' does not exist".format(path)
        else:
            msg = (
                "no metric files in this repository."
                " use 'dvc metrics add' to add a metric file to track."
            )

        raise DvcException(msg)

    def _metrics_modify(self, path, typ=None, xpath=None, delete=False):
        outs = self._find_output_by_path(path)

        if not outs:
            msg = "unable to find file '{}' in the pipeline".format(path)
            raise DvcException(msg)

        if len(outs) != 1:
            msg = (
                "-R not yet supported for metrics modify. "
                "Make sure only one metric is referred to by '{}'".format(path)
            )
            raise DvcException(msg)

        out = outs[0]

        if out.scheme != "local":
            msg = "output '{}' scheme '{}' is not supported for metrics"
            raise DvcException(msg.format(out.path, out.path_info["scheme"]))

        if out.use_cache:
            msg = "cached output '{}' is not supported for metrics"
            raise DvcException(msg.format(out.rel_path))

        if typ:
            if not isinstance(out.metric, dict):
                out.metric = {}
            out.metric[out.PARAM_METRIC_TYPE] = typ

        if xpath:
            if not isinstance(out.metric, dict):
                out.metric = {}
            out.metric[out.PARAM_METRIC_XPATH] = xpath

        if delete:
            out.metric = None

        out._verify_metric()

        out.stage.dump()

    def metrics_modify(self, path=None, typ=None, xpath=None):
        self._metrics_modify(path, typ, xpath)

    def metrics_add(self, path, typ=None, xpath=None):
        if not typ:
            typ = "raw"
        self._metrics_modify(path, typ, xpath)

    def metrics_remove(self, path):
        self._metrics_modify(path, delete=True)

    def graph(self, stages=None, from_directory=None):
        import networkx as nx
        from dvc.exceptions import (
            OutputDuplicationError,
            WorkingDirectoryAsOutputError,
        )

        G = nx.DiGraph()
        G_active = nx.DiGraph()
        stages = stages or self.stages(from_directory)
        stages = [stage for stage in stages if stage]
        outs = []

        for stage in stages:
            for out in stage.outs:
                existing = [o.stage for o in outs if o.path == out.path]

                if existing:
                    stages = [stage.relpath, existing[0].relpath]
                    raise OutputDuplicationError(out.path, stages)

                outs.append(out)

        for stage in stages:
            for out in outs:
                overlaps = stage.cwd == out.path or stage.cwd.startswith(
                    out.path + os.sep
                )

                if overlaps:
                    raise WorkingDirectoryAsOutputError(
                        stage.cwd, stage.relpath
                    )

        # collect the whole DAG
        for stage in stages:
            node = os.path.relpath(stage.path, self.root_dir)

            G.add_node(node, stage=stage)
            G_active.add_node(node, stage=stage)

            for dep in stage.deps:
                for out in outs:
                    if (
                        out.path != dep.path
                        and not dep.path.startswith(out.path + out.sep)
                        and not out.path.startswith(dep.path + dep.sep)
                    ):
                        continue

                    dep_stage = out.stage
                    dep_node = os.path.relpath(dep_stage.path, self.root_dir)
                    G.add_node(dep_node, stage=dep_stage)
                    G.add_edge(node, dep_node)
                    if not stage.locked:
                        G_active.add_node(dep_node, stage=dep_stage)
                        G_active.add_edge(node, dep_node)

        self._check_cyclic_graph(G)

        return G, G_active

    def pipelines(self, from_directory=None):
        import networkx as nx

        G, G_active = self.graph(from_directory=from_directory)

        return [
            G.subgraph(c).copy() for c in nx.weakly_connected_components(G)
        ]

    def stages(self, from_directory=None):
        """
        Walks down the root directory looking for Dvcfiles,
        skipping the directories that are related with
        any SCM (e.g. `.git`), DVC itself (`.dvc`), or directories
        tracked by DVC (e.g. `dvc add data` would skip `data/`)

        NOTE: For large projects, this could be an expensive
              operation. Consider using some memoization.
        """
        from dvc.stage import Stage

        if not from_directory:
            from_directory = self.root_dir

        stages = []
        outs = []
        for root, dirs, files in os.walk(from_directory):
            for fname in files:
                path = os.path.join(root, fname)
                if not Stage.is_stage_file(path):
                    continue
                stage = Stage.load(self, path)
                for out in stage.outs:
                    outs.append(out.path + out.sep)
                stages.append(stage)

            def filter_dirs(dname, root=root):
                path = os.path.join(root, dname)
                if path in (self.dvc_dir, self.scm.dir):
                    return False
                for out in outs:
                    if path == os.path.normpath(out) or path.startswith(out):
                        return False
                return True

            dirs[:] = list(filter(filter_dirs, dirs))

        return stages

    def active_stages(self, from_directory=None):
        import networkx as nx

        stages = []
        for G in self.pipelines(from_directory):
            stages.extend(list(nx.get_node_attributes(G, "stage").values()))
        return stages

    def _welcome_message(self):
        import colorama

        logger.box(
            "DVC has enabled anonymous aggregate usage analytics.\n"
            "Read the analytics documentation (and how to opt-out) here:\n"
            "{blue}https://dvc.org/doc/user-guide/analytics{nc}".format(
                blue=colorama.Fore.BLUE, nc=colorama.Fore.RESET
            ),
            border_color="red",
        )

        logger.info(
            "{yellow}What's next?{nc}\n"
            "{yellow}------------{nc}\n"
            "- Check out the documentation: {blue}https://dvc.org/doc{nc}\n"
            "- Get help and share ideas: {blue}https://dvc.org/chat{nc}\n"
            "- Star us on GitHub: {blue}https://github.com/iterative/dvc{nc}".format(
                yellow=colorama.Fore.YELLOW,
                blue=colorama.Fore.BLUE,
                nc=colorama.Fore.RESET,
            )
        )

    def _expand_target_path(self, from_path, to_path):
        if os.path.isdir(to_path) and not os.path.isdir(from_path):
            return os.path.join(to_path, os.path.basename(from_path))
        return to_path