pakages.utils package

Submodules

pakages.utils.command module

class pakages.utils.command.Capturing

Bases: object

capture output from stdout and stderr into capture object. This is based off of github.com/vsoch/gridtest but modified to write files. The stderr and stdout are set to temporary files at the init of the capture, and then they are closed when we exit. This means expected usage looks like:

with Capturing() as capture:

process = subprocess.Popen(…)

And then the output and error are retrieved from reading the files: and exposed as properties to the client:

capture.out capture.err

And cleanup means deleting these files, if they exist.

cleanup()

property err

Return error stream. Returns empty string if empty or doesn’t exist. Returns (str) : error stream written to file

property out

Return output stream. Returns empty string if empty or doesn’t exist. Returns (str) : output stream written to file

set_stderr()

set_stdout()

class pakages.utils.command.Command(cmd=None)

Bases: object

decode(line)

Given a line of output (error or regular) decode using the system default, if appropriate

export()

Return data as json.

get_command(cmd=None)

run(cmd=None)

Run and capture a shell command. We use Capturing and then save attributes like the pid, output, error to it, and return to the calling function.

summary()

pakages.utils.fileio module

pakages.utils.fileio.copyfile(source, destination, force=True)

Copy a file from a source to its destination.

pakages.utils.fileio.get_file_hash(image_path, algorithm='sha256')

Return an sha256 hash of the file based on a criteria level.

pakages.utils.fileio.get_tmpdir(tmpdir=None, prefix='', create=True)

Get a temporary directory for an operation.

pakages.utils.fileio.get_tmpfile(tmpdir=None, prefix='')

Get a temporary file with an optional prefix.

pakages.utils.fileio.mkdir_p(path)

mkdir_p attempts to get the same functionality as mkdir -p :param path: the path to create.

pakages.utils.fileio.mkdirp(dirnames)

Create one or more directories

pakages.utils.fileio.print_json(json_obj)

Print json pretty

pakages.utils.fileio.read_file(filename, mode='r')

Read a file.

pakages.utils.fileio.read_json(filename)

pakages.utils.fileio.read_yaml(filename)

pakages.utils.fileio.recursive_find(base, pattern=None)

Find filenames that match a particular pattern, and yield them.

pakages.utils.fileio.workdir(dirname)

Provide context for a working directory, e.g.,

with workdir(name):

# do stuff

pakages.utils.fileio.write_file(filename, content, mode='w')

Write content to a filename

pakages.utils.fileio.write_json(content, filename)

pakages.utils.terminal module

pakages.utils.terminal.check_install(software, quiet=True, command='--version')

check_install will attempt to run the singularity command, and return True if installed. The command line utils will not run without this check.

Parameters

software: the software to check if installed quiet: should we be quiet? (default True) command: the command to use to check (defaults to –version)

pakages.utils.terminal.confirm_action(question, force=False)

confirm if the user wants to perform a certain action

Parameters

question: the question that will be asked force: if the user wants to skip the prompt

pakages.utils.terminal.get_installdir()

get_installdir returns the installation directory of the application

pakages.utils.terminal.get_user()

Get the name of the user. We first try to import pwd, but fallback to extraction from the environment.

pakages.utils.terminal.get_userhome()

get the user home based on the effective uid. If import of pwd fails (not supported for Windows) then fall back to environment variable.

pakages.utils.terminal.run_command(cmd)

run_command uses subprocess to send a command to the terminal.

Parameters

cmd: the command to send, should be a list for subprocess error_message: the error message to give to user if fails, if none specified, will alert that command failed.

pakages.utils.terminal.stream_command(cmd)

stream a command (yield) back to the user, as each line is available.

# Example usage: results = [] for line in stream_command(cmd):

print(line, end=””) results.append(line)

Parameters

cmd: the command to send, should be a list for subprocess

pakages.utils.terminal.which(software, strip_newline=True)

get_install will return the path to where Singularity (or another executable) is installed.

Module contents