"""Utility function to deal with the directory structure.
"""
import os
import subprocess
[docs]def sync(path, options=None, source=None, target=None, echo=False, file=False, from_dir=None):
"""Sync given filename and/or directory from or to jalapeno.
Either source or target has to be set to a remote repository (either a complete address or one of jal, imac, ibook).
"""
if source is None and target is None and from_dir is None:
raise ValueError("Either source or target should be set to remote repository or remote directory should be provided")
if options is None:
options = '' if file else '-avP'
if isinstance(options, str):
options = options.split()
if from_dir is None:
full_name = os.path.realpath(path)
if os.path.isdir(full_name):
full_name = os.path.join(full_name, '')
workdir = os.environ['WORKDIR']
if contains(workdir, full_name):
reference_name = full_name[len(workdir):]
if os.path.isabs(reference_name):
reference_name = reference_name[1:]
elif contains('/vols/Data', full_name):
reference_name = os.path.join('fmrib', 'data', full_name[len('/vols/Data/'):]) + '/'
elif contains('/vols/Scratch', full_name):
reference_name = os.path.join('fmrib', 'scratch', full_name[len('/vols/Scratch/'):]) + '/'
elif contains('/Scratch', full_name):
reference_name = os.path.join('fmrib', 'scratch', full_name[len('/Scratch/'):]) + '/'
else:
raise ValueError("Only paths in Work directory (%s), /vols/Data, or /vols/Scratch can be synced, not %s" % (workdir, full_name))
full_source = os.path.join(path_work(source), reference_name)
full_target = os.path.join(path_work(target), reference_name)
else:
full_source = from_dir
full_target = path
if file:
cmd = ['scp', *options, full_source, full_target]
else:
cmd = ['rsync', *options, full_source + '/', full_target + '/']
if echo:
print(' '.join(cmd))
else:
subprocess.call(cmd)
return
[docs]def contains(parent, child):
"""Test if child is a subdirectory/file contained in parent.
"""
full_parent = os.path.realpath(parent)
full_child = os.path.realpath(child)
return full_child[:len(full_parent)] == full_parent
def _remote_text(remote=None):
"""Converts `remote` to a string.
"""
if remote is None:
return ''
if remote in os.environ:
return os.environ[remote] + ':'
return remote + ':'
[docs]def path_work(remote=None):
"""Return the local (default) or remote WORKDIR.
"""
if not remote:
return os.environ['WORKDIR']
else:
return _remote_text(remote) + 'Work'
[docs]def path_storage(owner, scratch=True, remote=None):
"""Returns the local (default) or remote scratch/data directory of `owner`.
"""
name_group = 'scratch' if scratch else 'data'
return os.path.join(path_work(remote), 'fmrib', name_group, owner)