/usr/lib/python2.7/dist-packages/versuchung/archives.py is in python-versuchung 1.1-3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 | # 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 Type, InputParameter
from versuchung.files import Directory, Directory_op_with, File
from versuchung.execute import shell
import logging
import os
import sys
class TarArchive(Type, InputParameter, Directory_op_with):
"""Can be used as: **input parameter**
The archive will be extracted to a temporary directory. It will be
removed after the experiment is over.
``clone_url`` can either be a :class:`string` or any object that
has a ``.path`` attribute (like
e.g. :class:`~versuchung.filesystems.File`). Of course the
referenced file must be a single file.
This parameter can be used as argument to the with keyword, to
change to the temporary directory::
with self.inputs.tar_archive as path:
# Here we have path == os.path.abspath(os.curdir)
# Do something in the extracted copy
print path
"""
def __init__(self, filename = None):
"""The default_filename is either a string to a file. Or a
object with a path attribute (e.g. a :class:`~versuchung.files.File`)"""
Type.__init__(self)
InputParameter.__init__(self)
Directory_op_with.__init__(self)
self.__filename = filename
self.__value = None
def inp_setup_cmdline_parser(self, parser):
self.inp_parser_add(parser, None, self.__filename)
def inp_extract_cmdline_parser(self, opts, args):
self.__filename = self.inp_parser_extract(opts, None)
def before_experiment_run(self, parameter_type):
if parameter_type == "input" and "path" in dir(self.__filename):
self.subobjects["filename"] = self.__filename
Type.before_experiment_run(self, parameter_type)
self.__filename = self.__filename.path
else:
Type.before_experiment_run(self, parameter_type)
self.__filename = os.path.abspath(self.__filename)
def inp_metadata(self):
return {self.name: self.__filename}
def __setup_value(self):
if "path" in dir(self.__filename):
self.subobjects["filename"] = self.__filename
self.__filename = self.__filename.path
self.__filename = os.path.abspath(self.__filename)
fn = self.__filename
extract_mode = ""
if "tar.gz" in fn or "tgz" in fn:
extract_mode = "x"
if "tar.bz2" in fn or "bzip2" in fn:
extract_mode = "j"
with self.tmp_directory as d:
try:
os.mkdir(self.name)
except OSError:
# ignore errors if the directory should already exist for some reason
pass
with Directory(self.name) as d2:
dirname = os.path.abspath(".")
(out, ret) = shell("tar %szvf %s", extract_mode, fn)
if ret != 0:
raise RuntimeError("Extracting of %s failed" % fn)
cd = None
for line in out:
if (cd == None or len(line) < len(cd)) and line.endswith("/"):
cd = line
if cd and all([x.startswith(cd) for x in out]):
dirname = cd
return Directory(os.path.abspath(dirname))
@property
def value(self):
"""Return a :class:`versuchung.files.Directory` instance to the extracted
tar archive. If it contains only one directory the instance
will point there. Otherwise it will point to a directory
containing the contents of the archive"""
if not self.__value:
self.__value = self.__setup_value()
return self.__value
@property
def path(self):
"""Return the string to the extract directory (same as .value.path)"""
return self.value.path
class GitArchive(InputParameter, Type, Directory_op_with):
"""Can be used as: **input parameter**
The git repository given in ``clone_url`` will be cloned to a
temporary directory. It will be removed after the experiment is
over. If ``shallow == True`` Only the files and not the .git is
copied (cloned). This is especially useful for large git
repositories like the Linux kernel tree.
``clone_url`` can either be a :class:`string` or any object that
has a ``.path`` attribute (like e.g. :class:`TarArchive`). Of
course the refenced path must be a directory.
This parameter can be used as argument to the with keyword, to
change to the temporary directory::
with self.inputs.git_archive as path:
# Here we have path == os.path.abspath(os.curdir)
# Do something in the extracted copy
print path
"""
def __init__(self, clone_url = None, ref = "refs/heads/master", shallow = False):
"""clone_url: where to the git archive from
This might either be a string or a object with a path attribute
ref: which git reference to checkout
shallow: do a shallow copy (using git-archive).
The git archive will be cloned to self.name (which is the
key in the input parameters dict)"""
Type.__init__(self)
InputParameter.__init__(self)
Directory_op_with.__init__(self)
self.__clone_url = clone_url
self.__ref = ref
self.__shallow = shallow
self.__value = None
self.__hash = None
def inp_setup_cmdline_parser(self, parser):
self.inp_parser_add(parser, "clone-url", self.__clone_url)
self.inp_parser_add(parser, "ref", self.__ref)
def inp_extract_cmdline_parser(self, opts, args):
self.__clone_url = self.inp_parser_extract(opts, "clone-url")
self.__ref = self.inp_parser_extract(opts, "ref")
def before_experiment_run(self, parameter_type):
if parameter_type == "input" and "path" in dir(self.__clone_url):
self.subobjects["clone-url"] = self.__clone_url
Type.before_experiment_run(self, parameter_type)
self.__clone_url = self.__clone_url.path
else:
Type.before_experiment_run(self, parameter_type)
def checkout_hash(self):
"""Return the hash of the HEAD commit hash as string"""
if not self.__hash:
cmd = "git ls-remote %s %s" % (self.__clone_url,
self.__ref)
(lines, ret) = shell(cmd)
if ret != 0 or lines == 0:
print "\n".join(lines)
sys.exit(-1)
self.__hash = lines[0].split("\t")[0]
if self.__hash == "":
self.__hash = self.__ref
return self.__hash
def checkout_ref(self):
"""Return git ref which was checked out"""
return self.__ref
def checkout_url(self):
"""Return git url which was checked out"""
return self.__clone_url
def inp_metadata(self):
return {self.name + "-clone-url": str(self.__clone_url),
self.name + "-ref": self.__ref,
self.name + "-hash": self.checkout_hash()}
def __setup_value(self):
if "path" in dir(self.__clone_url):
self.subobjects["clone-url"] = self.__clone_url
self.__clone_url = self.__clone_url.path
logging.info("copying git archive %s", self.__clone_url)
with self.tmp_directory as d:
os.mkdir(self.name)
if self.__shallow:
cmd = "cd '%s' && git archive --format=tar --remote=%s %s | tar x"
args = (self.name,
self.__clone_url,
self.__ref)
else:
cmd = "git clone %s %s"
args = (self.__clone_url, self.name)
(lines, ret) = shell(cmd, *args)
if ret != 0:
print "\n".join(lines)
sys.exit(-1)
if not self.__shallow:
cmd = "cd %s && git gc && git fetch %s %s && git checkout FETCH_HEAD"
args = (self.name, self.__clone_url, self.__ref)
(lines, ret) = shell(cmd, *args)
if ret != 0:
print "\n".join(lines)
sys.exit(-1)
return Directory(os.path.abspath(self.name))
@property
def value(self):
"""Return a :class:`versuchung.files.Directory` instance to the cloned git directory"""
if not self.__value:
self.__value = self.__setup_value()
return self.__value
@property
def path(self):
"""Return the string to the extract directory (same as .value.path)"""
return self.value.path
class GzipFile(File):
def original_filename(self):
return self.__original_filename
@property
def path(self):
path = File.path.fget(self)
if self.parameter_type == "input" and not os.path.exists(path):
shell("gunzip < %s > %s", self.__original_filename,
path)
return path
@property
def value(self):
path = File.path.fget(self)
if self.parameter_type == "input" and not os.path.exists(path):
shell("gunzip < %s > %s", self.__original_filename,
path)
return File.value.fget(self)
@value.setter
def value(self, value):
File.value.fset(self, value)
def before_experiment_run(self, parameter_type):
self.parameter_type = parameter_type
if parameter_type == "input":
self.__original_filename = File.path.fget(self)
self.subobjects["filename"] = File(self.__original_filename)
filename = self.name + "_" + os.path.basename(self.path.rstrip(".gz"))
self.set_path(self.tmp_directory.path, filename)
File.before_experiment_run(self, parameter_type)
def after_experiment_run(self, parameter_type):
File.after_experiment_run(self, parameter_type)
if parameter_type == "output":
shell("gzip -c %s > %s.1", self.path, self.path)
shell("mv %s.1 %s", self.path, self.path)
|