Source code for versuchung.files

# This file is part of versuchung.
# 
# versuchung is free software: you can redistribute it and/or modify it under the
# terms of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
# 
# versuchung is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
# PARTICULAR PURPOSE.  See the GNU General Public License for more details.
# 
# You should have received a copy of the GNU General Public License along with
# versuchung.  If not, see <http://www.gnu.org/licenses/>.


from versuchung.types import InputParameter, OutputParameter, Type
from versuchung.tools import before
from cStringIO import StringIO
import shutil
import csv
import os, stat
import hashlib

class FilesystemObject(InputParameter, OutputParameter, Type):
    def __init__(self, default_name=""):
        InputParameter.__init__(self)
        OutputParameter.__init__(self)
        Type.__init__(self)
        self.__object_name = default_name
        self.__enclosing_directory = os.path.abspath(os.curdir)
        self.__force_enclosing_directory = False

    def inp_setup_cmdline_parser(self, parser):
        self.inp_parser_add(parser, None, self.__object_name)

    def inp_extract_cmdline_parser(self, opts, args):
        self.__object_name = self.inp_parser_extract(opts, None)

    def inp_metadata(self):
        return {self.name: self.__object_name}

    @property
    def path(self):
        """:return: string -- path to the file/directory"""
        if not self.__force_enclosing_directory:
            if self.parameter_type == "input":
                if self.static_experiment == self.dynamic_experiment:
                    self.__enclosing_directory = os.path.abspath(os.curdir)
                else:
                    self.__enclosing_directory = self.static_experiment.base_directory
            elif self.parameter_type == "output":
                assert self.static_experiment == self.dynamic_experiment
                self.__enclosing_directory = self.dynamic_experiment.base_directory
            else:
                self.__enclosing_directory = os.path.abspath(os.curdir)
        return os.path.join(self.__enclosing_directory, self.__object_name)

    @property
    def basename(self):
        return os.path.basename(self.path)

    @property
    def dirname(self):
        return os.path.basename(self.path)

    def set_path(self, base_directory, object_name):
        assert base_directory[0] == "/"
        self.__force_enclosing_directory = True
        self.__enclosing_directory = base_directory
        self.__object_name         = object_name

[docs]class File(FilesystemObject): """Can be used as: **input parameter** and **output parameter** The File type represents the content of a single file. Its contents can be read and written most easily with the :attr:`value` property. Alternatively, the method :meth:`write` appends new content if the parameter `append` is set to `True`. NB: The content of the file is flushed only after the experiment finishes. Use :meth:`flush` to force writing the buffered data to disk before the experiment finishes. """ def __init__(self, default_filename=""): FilesystemObject.__init__(self, default_filename) self.__value = None @property def value(self): """This attribute can be read and written and represent the content of the specified file""" if not self.__value: try: with open(self.path) as fd: self.__value = self.after_read(fd.read()) except IOError: # File couldn't be read self.__value = self.after_read("") return self.__value @value.setter
[docs] def value(self, value): self.__value = value
[docs] def write(self, content, append = False): """Similar to the :attr:`value` property. If the parameter `append` is `False`, then the property :attr:`value` is reset (i.e., overwritten), otherwise the content is appendend""" if append: self.value += content else: self.value = content
def after_experiment_run(self, parameter_type): FilesystemObject.after_experiment_run(self, parameter_type) assert parameter_type in ["input", "output"] if parameter_type == "output": self.flush()
[docs] def flush(self): """Flush the cached content of the file to disk""" if not self.__value: return with open(self.path, "w+") as fd: v = self.before_write(self.value) if v is None: v = "" fd.write(v)
[docs] def copy_contents(self, filename): """Read the given file and replace the current .value with the files content. Flushes automatically afterwards.""" with open(filename) as fd: self.value = self.after_read(fd.read()) self.flush()
def make_executable(self): """makes a file exectuable (chmod +x $file)""" st = os.stat(self.path) os.chmod(self.path, st.st_mode | stat.S_IEXEC) def after_read(self, value): """To provide filtering of file contents in subclasses, overrwrite this method. It is gets the file content as a string and returns the value()""" return value def before_write(self, value): """To provide filtering of file contents in subclasses, overrwrite this method. This method gets the value() and returns a string, when the file is written to disk""" return value
class Executable(File): """Can be used as: **input parameter** An executable is a :class:`versuchung.files.File` that only references an executable. It checksums the executable and puts the checksum into the metadata. The executable is never changed. """ def __init__(self, default_filename): File.__init__(self, default_filename) @property def value(self): raise NotImplemented @value.setter def value(self, value): raise NotImplementedError def write(self, content, append = False): raise NotImplementedError def after_experiment_run(self, parameter_type): pass def flush(self): raise NotImplementedError def copy_contents(self, filename): raise NotImplementedError def make_executable(self): raise NotImplementedError def inp_metadata(self): return {self.name + "-md5": hashlib.md5(open(self.path).read()).hexdigest()} def execute(self, cmdline, *args): """Does start the executable with meth:`versuchung.execute.shell` and args, which is of type list, as arguments.""" from versuchung.execute import shell shell(self.path + " " + cmdline, *args) class Directory_op_with: def __init__(self): self.__olddir = [] def __enter__(self): self.__olddir.append(os.path.abspath(os.curdir)) os.chdir(self.path) return self.path def __exit__(self, *excinfo): path = self.__olddir[-1] del self.__olddir[-1] os.chdir(path)
[docs]class Directory(FilesystemObject, Directory_op_with): """Can be used as: **input parameter** and **output parameter** Represents the contents of directory. It can also be used with the **with**-keyword to change the current working directory temporarily to this directory:: with directory as dir: # Do something with adjusted current working directory print os.curdir """ def __init__(self, default_filename=""): FilesystemObject.__init__(self, default_filename) Directory_op_with.__init__(self) self.__value = None self.__new_files = [] def ___ensure_dir_exists(self): if not os.path.exists(self.path): os.mkdir(self.path) # Ensure dir exists DECORATOR __ensure_dir_exists = before(___ensure_dir_exists) @property
[docs] def value(self): """:return: list -- directories and files in given directory""" if not self.__value: self.__value = os.listdir(self.path) return self.__value
def __iter__(self): for name in self.value: p = os.path.join(self.path, name) if os.path.isdir(p): d = Directory(name) d.set_path(self.path, p) self.subobjects[name] = d yield d else: f = File(name) f.set_path(self.path, p) self.subobjects[name] = f yield f def before_experiment_run(self, parameter_type): FilesystemObject.before_experiment_run(self, parameter_type) if parameter_type == "output": self.___ensure_dir_exists() @__ensure_dir_exists
[docs] def new_file(self, name): """Generate a new :class:`~versuchung.files.File` in the directory. It will be flushed automatically if the experiment is over.""" f = File(name) f.set_path(self.path, name) self.subobjects[name] = f return f
@__ensure_dir_exists def new_directory(self, name): """Generate a new :class:`~versuchung.files.Directory` in the directory. The directory <name> must not be present before""" f = Directory(name) f.set_path(self.path, name) os.mkdir(f.path) self.subobjects[name] = f return f @__ensure_dir_exists
[docs] def mirror_directory(self, path, include_closure = None): """Copies the contents of the given directory to this directory. The include closure is a function, which checks for every (absolute) path in the origin directory, if it is mirrored. If it is None, all files are included.""" if not include_closure: include_closure = lambda arg: True if not os.path.exists(path) and os.path.isdir(path): raise RuntimeError("Argument is no directory") path = os.path.abspath(path) for root, dirs, files in os.walk(path): root = root[len(path)+1:] for d in dirs: src = os.path.join(path, root, d) if not include_closure(src): continue dst = os.path.join(self.path, root, d) if not os.path.isdir(dst): os.mkdir(dst) for f in files: src = os.path.join(path, root, f) if not include_closure(src): continue dst = os.path.join(self.path, root, f) shutil.copyfile(src,dst) self.__value = None
[docs]class CSV_File(File): """Can be used as: **input parameter** and **output parameter** It is a normal :class:`~versuchung.files.File` but the content of the file is interpreted as a csv file. It is parsed before the value is exposed to the user. And formatted before the content is written to disk. Internally the :mod:`csv` is used, so all arguments to ``csv.reader`` and ``csv.writer`` can be given in *csv_args*.""" value = File.value """Other than a normal CSV_File the value of a CSV_File is a list of lists, which represents the structure of the csv file. This value can be manipulated by the user. >>> CSV_File("csv_output").value [["1", "2", "3"]]""" def __init__(self, default_filename = "", **csv_args): File.__init__(self, default_filename) self.csv_args = csv_args def after_read(self, value): fd = StringIO(value) reader = csv.reader(fd, self.csv_args) return list(reader) def before_write(self, value): fd = StringIO() writer = csv.writer(fd, self.csv_args) writer.writerows(value) return fd.getvalue() def write(self): raise NotImplemented
[docs] def append(self, row): """Append a row to the csv file It is just a shorthand for >>> csv_file.value.append([1,2,3]) :param row: row to append :type row: list.""" if type(row) != list: raise TypeError("list of values required") self.value.append(row)