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 args = parser.parse_args(arguments)
113 if len(args.paths) == 0:
114 args.paths += ['/dev/stdin']
116 if len(args.paths) != 1 or os.path.isdir(args.paths[0]):
119 --filename argument cannot be used when storing a directory or
123 # Turn on --progress by default if stderr is a tty.
124 if (not (args.batch_progress or args.no_progress)
125 and os.isatty(sys.stderr.fileno())):
128 if args.paths == ['-']:
129 args.paths = ['/dev/stdin']
130 if not args.filename:
135 class ResumeCacheConflict(Exception):
139 class ResumeCache(object):
140 CACHE_DIR = os.path.expanduser('~/.cache/arvados/arv-put')
142 def __init__(self, file_spec):
144 self.cache_file = open(file_spec, 'a+')
146 file_spec = self.make_path(file_spec)
147 self.cache_file = open(file_spec, 'a+')
148 self._lock_file(self.cache_file)
149 self.filename = self.cache_file.name
152 def make_path(cls, args):
154 md5.update(arvados.config.get('ARVADOS_API_HOST', '!nohost'))
155 realpaths = sorted(os.path.realpath(path) for path in args.paths)
156 md5.update(''.join(realpaths))
157 if any(os.path.isdir(path) for path in realpaths):
158 md5.update(str(max(args.max_manifest_depth, -1)))
160 md5.update(args.filename)
161 return os.path.join(cls.CACHE_DIR, md5.hexdigest())
163 def _lock_file(self, fileobj):
165 fcntl.flock(fileobj, fcntl.LOCK_EX | fcntl.LOCK_NB)
167 raise ResumeCacheConflict("{} locked".format(fileobj.name))
170 self.cache_file.seek(0)
171 return json.load(self.cache_file)
173 def save(self, data):
175 new_cache_fd, new_cache_name = tempfile.mkstemp(
176 dir=os.path.dirname(self.filename))
177 self._lock_file(new_cache_fd)
178 new_cache = os.fdopen(new_cache_fd, 'r+')
179 json.dump(data, new_cache)
180 os.rename(new_cache_name, self.filename)
181 except (IOError, OSError, ResumeCacheConflict) as error:
183 os.unlink(new_cache_name)
184 except NameError: # mkstemp failed.
187 self.cache_file.close()
188 self.cache_file = new_cache
191 self.cache_file.close()
195 os.unlink(self.filename)
196 except OSError as error:
197 if error.errno != errno.ENOENT: # That's what we wanted anyway.
203 self.__init__(self.filename)
206 class ArvPutCollectionWriter(arvados.ResumableCollectionWriter):
207 def __init__(self, cache=None, reporter=None, bytes_expected=None):
208 self.__init_locals__(cache, reporter, bytes_expected)
209 super(ArvPutCollectionWriter, self).__init__()
211 def __init_locals__(self, cache, reporter, bytes_expected):
213 self.report_func = reporter
214 self.bytes_written = 0
215 self.bytes_expected = bytes_expected
218 def from_cache(cls, cache, reporter=None, bytes_expected=None):
221 state['_data_buffer'] = [base64.decodestring(state['_data_buffer'])]
222 writer = cls.from_state(state)
223 except (TypeError, ValueError,
224 arvados.errors.StaleWriterStateError) as error:
225 return cls(cache, reporter, bytes_expected)
227 writer.__init_locals__(cache, reporter, bytes_expected)
230 def checkpoint_state(self):
231 if self.cache is None:
233 state = self.dump_state()
234 # Transform attributes for serialization.
235 for attr, value in state.items():
236 if attr == '_data_buffer':
237 state[attr] = base64.encodestring(''.join(value))
238 elif hasattr(value, 'popleft'):
239 state[attr] = list(value)
240 self.cache.save(state)
242 def flush_data(self):
243 bytes_buffered = self._data_buffer_len
244 super(ArvPutCollectionWriter, self).flush_data()
245 self.bytes_written += (bytes_buffered - self._data_buffer_len)
246 if self.report_func is not None:
247 self.report_func(self.bytes_written, self.bytes_expected)
250 def expected_bytes_for(pathlist):
251 # Walk the given directory trees and stat files, adding up file sizes,
252 # so we can display progress as percent
254 for path in pathlist:
255 if os.path.isdir(path):
256 for filename in arvados.util.listdir_recursive(path):
257 bytesum += os.path.getsize(os.path.join(path, filename))
258 elif not os.path.isfile(path):
261 bytesum += os.path.getsize(path)
264 _machine_format = "{} {}: {{}} written {{}} total\n".format(sys.argv[0],
266 def machine_progress(bytes_written, bytes_expected):
267 return _machine_format.format(
268 bytes_written, -1 if (bytes_expected is None) else bytes_expected)
270 def human_progress(bytes_written, bytes_expected):
272 return "\r{}M / {}M {:.1f}% ".format(
273 bytes_written >> 20, bytes_expected >> 20,
274 bytes_written / bytes_expected)
276 return "\r{} ".format(bytes_written)
278 def progress_writer(progress_func, outfile=sys.stderr):
279 def write_progress(bytes_written, bytes_expected):
280 outfile.write(progress_func(bytes_written, bytes_expected))
281 return write_progress
283 def main(arguments=None):
284 args = parse_arguments(arguments)
287 reporter = progress_writer(human_progress)
288 elif args.batch_progress:
289 reporter = progress_writer(machine_progress)
293 writer = ArvPutCollectionWriter(
294 reporter=reporter, bytes_expected=expected_bytes_for(args.paths))
296 # Copy file data to Keep.
297 for path in args.paths:
298 if os.path.isdir(path):
299 writer.write_directory_tree(
300 path, max_manifest_depth=args.max_manifest_depth)
302 writer.start_new_stream()
303 writer.write_file(path, args.filename or os.path.basename(path))
306 print writer.manifest_text(),
308 writer.finish_current_stream()
309 print ','.join(writer.data_locators())
311 # Register the resulting collection in Arvados.
312 arvados.api().collections().create(
314 'uuid': writer.finish(),
315 'manifest_text': writer.manifest_text(),
319 # Print the locator (uuid) of the new collection.
320 print writer.finish()
322 if __name__ == '__main__':