<filename>dvcrecord/trackedfile.py
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/trackedfile.ipynb (unless otherwise specified).

__all__ = ['TrackedFile']

# Cell
import yaml

from .utils import maybe_yaml, write_yaml

class TrackedFile:
    def __init__(self, fpath, *args, **kwargs):
        self.fpath = fpath
        self.args = args
        self.kwargs = kwargs
        self.file_obj = None

    def open(self, *args, **kwargs):
        self.file_obj = open(self.fpath, *args, **kwargs)
        return self.file_obj

    def __enter__(self):
        self.file_obj = open(self.fpath, *self.args, **self.kwargs)
        return self.file_obj

    def __exit__(self, *args):
        self.file_obj.close()

    @classmethod
    def from_previous_output(clz, stage, *args, pipeline_file=None, fpath=None, **kwargs):
        key = 'outs'
        pipeline_file = pipeline_file or  PIPELINE_FILE_DEFAULT
        with open(pipeline_file) as p:
            pipeline = yaml.safe_load(p)

        fnames = pipeline['stages'][stage][key]
        if fpath is not None:
            assert fpath in fnames
        fpath = fpath or fnames[0]
        return clz(fpath, *args, **kwargs)

    def render(self, as_yaml=False):
        return maybe_yaml(self.fpath, as_yaml)

<filename>dvcrecord/trackedfile.py
# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/trackedfile.ipynb (unless otherwise specified).

__all__ = ['TrackedFile']

# Cell
import yaml

from .utils import maybe_yaml, write_yaml

class TrackedFile:
    def __init__(self, fpath, *args, **kwargs):
        self.fpath = fpath
        self.args = args
        self.kwargs = kwargs
        self.file_obj = None

    def open(self, *args, **kwargs):
        self.file_obj = open(self.fpath, *args, **kwargs)
        return self.file_obj

    def __enter__(self):
        self.file_obj = open(self.fpath, *self.args, **self.kwargs)
        return self.file_obj

    def __exit__(self, *args):
        self.file_obj.close()

    @classmethod
    def from_previous_output(clz, stage, *args, pipeline_file=None, fpath=None, **kwargs):
        key = 'outs'
        pipeline_file = pipeline_file or  PIPELINE_FILE_DEFAULT
        with open(pipeline_file) as p:
            pipeline = yaml.safe_load(p)

        fnames = pipeline['stages'][stage][key]
        if fpath is not None:
            assert fpath in fnames
        fpath = fpath or fnames[0]
        return clz(fpath, *args, **kwargs)

    def render(self, as_yaml=False):
        return maybe_yaml(self.fpath, as_yaml)

# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/trackedfile.ipynb (unless otherwise specified). # Cell