4 # --md5sum - display md5 of each file as read from disk
17 def parse_arguments(arguments):
18 parser = argparse.ArgumentParser(
19 description='Copy data from the local filesystem to Keep.')
21 parser.add_argument('paths', metavar='path', type=str, nargs='*',
23 Local file or directory. Default: read from standard input.
26 parser.add_argument('--max-manifest-depth', type=int, metavar='N',
28 Maximum depth of directory tree to represent in the manifest
29 structure. A directory structure deeper than this will be represented
30 as a single stream in the manifest. If N=0, the manifest will contain
31 a single stream. Default: -1 (unlimited), i.e., exactly one manifest
32 stream per filesystem directory that contains files.
35 group = parser.add_mutually_exclusive_group()
37 group.add_argument('--as-stream', action='store_true', dest='stream',
42 group.add_argument('--stream', action='store_true',
44 Store the file content and display the resulting manifest on
45 stdout. Do not write the manifest to Keep or save a Collection object
49 group.add_argument('--as-manifest', action='store_true', dest='manifest',
51 Synonym for --manifest.
54 group.add_argument('--in-manifest', action='store_true', dest='manifest',
56 Synonym for --manifest.
59 group.add_argument('--manifest', action='store_true',
61 Store the file data and resulting manifest in Keep, save a Collection
62 object in Arvados, and display the manifest locator (Collection uuid)
63 on stdout. This is the default behavior.
66 group.add_argument('--as-raw', action='store_true', dest='raw',
71 group.add_argument('--raw', action='store_true',
73 Store the file content and display the data block locators on stdout,
74 separated by commas, with a trailing newline. Do not store a
78 parser.add_argument('--use-filename', type=str, default=None,
79 dest='filename', help="""
80 Synonym for --filename.
83 parser.add_argument('--filename', type=str, default=None,
85 Use the given filename in the manifest, instead of the name of the
86 local file. This is useful when "-" or "/dev/stdin" is given as an
87 input file. It can be used only if there is exactly one path given and
88 it is not a directory. Implies --manifest.
91 group = parser.add_mutually_exclusive_group()
92 group.add_argument('--progress', action='store_true',
94 Display human-readable progress on stderr (bytes and, if possible,
95 percentage of total data size). This is the default behavior when
99 group.add_argument('--no-progress', action='store_true',
101 Do not display human-readable progress on stderr, even if stderr is a
105 group.add_argument('--batch-progress', action='store_true',
107 Display machine-readable progress on stderr (bytes and, if known,
111 group = parser.add_mutually_exclusive_group()
112 group.add_argument('--resume', action='store_true', default=True,
114 Continue interrupted uploads from cached state (default).
116 group.add_argument('--no-resume', action='store_false', dest='resume',
118 Do not continue interrupted uploads from cached state.
121 args = parser.parse_args(arguments)
123 if len(args.paths) == 0:
124 args.paths += ['/dev/stdin']
126 if len(args.paths) != 1 or os.path.isdir(args.paths[0]):
129 --filename argument cannot be used when storing a directory or
133 # Turn on --progress by default if stderr is a tty.
134 if (not (args.batch_progress or args.no_progress)
135 and os.isatty(sys.stderr.fileno())):
138 if args.paths == ['-']:
139 args.paths = ['/dev/stdin']
140 if not args.filename:
145 class ResumeCacheConflict(Exception):
149 class ResumeCache(object):
150 CACHE_DIR = os.path.expanduser('~/.cache/arvados/arv-put')
153 def setup_user_cache(cls):
155 os.makedirs(cls.CACHE_DIR)
156 except OSError as error:
157 if error.errno != errno.EEXIST:
160 os.chmod(cls.CACHE_DIR, 0o700)
162 def __init__(self, file_spec):
163 self.cache_file = open(file_spec, 'a+')
164 self._lock_file(self.cache_file)
165 self.filename = self.cache_file.name
168 def make_path(cls, args):
170 md5.update(arvados.config.get('ARVADOS_API_HOST', '!nohost'))
171 realpaths = sorted(os.path.realpath(path) for path in args.paths)
172 md5.update(''.join(realpaths))
173 if any(os.path.isdir(path) for path in realpaths):
174 md5.update(str(max(args.max_manifest_depth, -1)))
176 md5.update(args.filename)
177 return os.path.join(cls.CACHE_DIR, md5.hexdigest())
179 def _lock_file(self, fileobj):
181 fcntl.flock(fileobj, fcntl.LOCK_EX | fcntl.LOCK_NB)
183 raise ResumeCacheConflict("{} locked".format(fileobj.name))
186 self.cache_file.seek(0)
187 return json.load(self.cache_file)
189 def save(self, data):
191 new_cache_fd, new_cache_name = tempfile.mkstemp(
192 dir=os.path.dirname(self.filename))
193 self._lock_file(new_cache_fd)
194 new_cache = os.fdopen(new_cache_fd, 'r+')
195 json.dump(data, new_cache)
196 os.rename(new_cache_name, self.filename)
197 except (IOError, OSError, ResumeCacheConflict) as error:
199 os.unlink(new_cache_name)
200 except NameError: # mkstemp failed.
203 self.cache_file.close()
204 self.cache_file = new_cache
207 self.cache_file.close()
211 os.unlink(self.filename)
212 except OSError as error:
213 if error.errno != errno.ENOENT: # That's what we wanted anyway.
219 self.__init__(self.filename)
222 class ArvPutCollectionWriter(arvados.ResumableCollectionWriter):
223 STATE_PROPS = (arvados.ResumableCollectionWriter.STATE_PROPS +
226 def __init__(self, cache=None, reporter=None, bytes_expected=None):
227 self.bytes_written = 0
229 self.report_func = reporter
230 self.bytes_expected = bytes_expected
231 super(ArvPutCollectionWriter, self).__init__()
234 def from_cache(cls, cache, reporter=None, bytes_expected=None):
237 state['_data_buffer'] = [base64.decodestring(state['_data_buffer'])]
238 writer = cls.from_state(state, cache, reporter, bytes_expected)
239 except (TypeError, ValueError,
240 arvados.errors.StaleWriterStateError) as error:
241 return cls(cache, reporter, bytes_expected)
245 def preresume_hook(self):
246 print >>sys.stderr, "arv-put: Resuming previous upload. Bypass with the --no-resume option."
248 def checkpoint_state(self):
249 if self.cache is None:
251 state = self.dump_state()
252 # Transform attributes for serialization.
253 for attr, value in state.items():
254 if attr == '_data_buffer':
255 state[attr] = base64.encodestring(''.join(value))
256 elif hasattr(value, 'popleft'):
257 state[attr] = list(value)
258 self.cache.save(state)
260 def flush_data(self):
261 bytes_buffered = self._data_buffer_len
262 super(ArvPutCollectionWriter, self).flush_data()
263 self.bytes_written += (bytes_buffered - self._data_buffer_len)
264 if self.report_func is not None:
265 self.report_func(self.bytes_written, self.bytes_expected)
268 def expected_bytes_for(pathlist):
269 # Walk the given directory trees and stat files, adding up file sizes,
270 # so we can display progress as percent
272 for path in pathlist:
273 if os.path.isdir(path):
274 for filename in arvados.util.listdir_recursive(path):
275 bytesum += os.path.getsize(os.path.join(path, filename))
276 elif not os.path.isfile(path):
279 bytesum += os.path.getsize(path)
282 _machine_format = "{} {}: {{}} written {{}} total\n".format(sys.argv[0],
284 def machine_progress(bytes_written, bytes_expected):
285 return _machine_format.format(
286 bytes_written, -1 if (bytes_expected is None) else bytes_expected)
288 def human_progress(bytes_written, bytes_expected):
290 return "\r{}M / {}M {:.1%} ".format(
291 bytes_written >> 20, bytes_expected >> 20,
292 float(bytes_written) / bytes_expected)
294 return "\r{} ".format(bytes_written)
296 def progress_writer(progress_func, outfile=sys.stderr):
297 def write_progress(bytes_written, bytes_expected):
298 outfile.write(progress_func(bytes_written, bytes_expected))
299 return write_progress
301 def main(arguments=None):
302 ResumeCache.setup_user_cache()
303 args = parse_arguments(arguments)
306 reporter = progress_writer(human_progress)
307 elif args.batch_progress:
308 reporter = progress_writer(machine_progress)
313 resume_cache = ResumeCache(ResumeCache.make_path(args))
315 resume_cache.restart()
316 except ResumeCacheConflict:
317 print "arv-put: Another process is already uploading this data."
320 writer = ArvPutCollectionWriter.from_cache(
321 resume_cache, reporter, expected_bytes_for(args.paths))
323 # Copy file data to Keep.
324 for path in args.paths:
325 if os.path.isdir(path):
326 writer.write_directory_tree(
327 path, max_manifest_depth=args.max_manifest_depth)
329 writer.start_new_stream()
330 writer.write_file(path, args.filename or os.path.basename(path))
333 print writer.manifest_text(),
335 writer.finish_current_stream()
336 print ','.join(writer.data_locators())
338 # Register the resulting collection in Arvados.
339 arvados.api().collections().create(
341 'uuid': writer.finish(),
342 'manifest_text': writer.manifest_text(),
346 # Print the locator (uuid) of the new collection.
347 print writer.finish()
348 resume_cache.destroy()
350 if __name__ == '__main__':