pydra.engine.helpers_file module

Functions ported from Nipype 1, after removing parts that were related to py2.

class pydra.engine.helpers_file.MountIndentifier

Bases: object

Used to check the mount type that given file paths reside on in order to determine features that can be used (e.g. symlinks)

classmethod generate_cifs_table() → List[Tuple[str, str]]

Construct a reverse-length-ordered list of mount points that fall under a CIFS mount.

This precomputation allows efficient checking for whether a given path would be on a CIFS filesystem. On systems without a mount command, or with no CIFS mounts, returns an empty list.

classmethod get_mount(path: PathLike) → Tuple[Path, str]

Get the mount point for a given file-system path

Parameters:

path (os.PathLike) – the file-system path to identify the mount of

Returns:

mount_point (os.PathLike) – the root of the mount the path sits on
fstype (str) – the type of the file-system (e.g. ext4 or cifs)

classmethod get_mount_table() → List[Tuple[str, str]]

classmethod on_cifs(path: PathLike) → bool

Check whether a file path is on a CIFS filesystem mounted in a POSIX host.

POSIX hosts are assumed to have the mount command.

On Windows, Docker mounts host directories into containers through CIFS shares, which has support for Minshall+French symlinks, or text files that the CIFS driver exposes to the OS as symlinks. We have found that under concurrent access to the filesystem, this feature can result in failures to create or read recently-created symlinks, leading to inconsistent behavior and FileNotFoundError errors.

This check is written to support disabling symlinks on CIFS shares.

NB: This function and sub-functions are copied from the nipype.utils.filemanip module

NB: Adapted from https://github.com/nipy/nipype

classmethod on_same_mount(path1: PathLike, path2: PathLike) → bool: Checks whether two or paths are on the same logical file system

classmethod parse_mount_table(exit_code: int, output: str) → List[Tuple[str, str]]

Parse the output of mount to produce (path, fs_type) pairs.

Separated from _generate_cifs_table to enable testing logic with real outputs

classmethod patch_table(mount_table: List[Tuple[str, str]]): Patch the mount table with new values. Used in test routines

pydra.engine.helpers_file.copy_nested_files(value: Any, dest_dir: PathLike, supported_modes: CopyMode = CopyMode.any, **kwargs) → Any

Copies all “file-sets” found within the nested value (e.g. dict, list,…) into the destination directory. If no nested file-sets are found then the original value is returned. Note that multiple nested file-sets (e.g. a list) will to have unique names names (i.e. not differentiated by parent directories) otherwise there will be a path clash in the destination directory.

Parameters:

value (Any) – the value to copy files from (if required)
dest_dir (os.PathLike) – the destination directory to copy the files to
**kwargs – passed directly onto FileSet.copy()

pydra.engine.helpers_file.ensure_list(filename): Return a list given either a string or a list.

pydra.engine.helpers_file.is_container(item)

Check if item is a container (list, tuple, dict, set).

Parameters:: item (object) – Input object to check.
Returns:: output – True if container False otherwise.
Return type:: bool

pydra.engine.helpers_file.is_local_file(f)

pydra.engine.helpers_file.template_update(inputs, output_dir, state_ind=None, map_copyfiles=None)

Update all templates that are present in the input spec.

Should be run when all inputs used in the templates are already set.

pydra.engine.helpers_file.template_update_single(field, inputs, inputs_dict_st=None, output_dir=None, spec_type='input'): Update a single template from the input_spec or output_spec based on the value from inputs_dict (checking the types of the fields, that have “output_file_template)”