pydra.engine.helpers module

Administrative support for the engine framework.

class pydra.engine.helpers.PydraFileLock(lockfile)[source]

Bases: object

Wrapper for filelock’s SoftFileLock that makes it work with asyncio.

pydra.engine.helpers.argstr_formatting(argstr, inputs, value_updates=None)[source]

formatting argstr that have form {field_name}, using values from inputs and updating with value_update if provided

pydra.engine.helpers.copyfile_workflow(wf_path, result)[source]

if file in the wf results, the file will be copied to the workflow directory

pydra.engine.helpers.create_checksum(name, inputs)[source]

Generate a checksum name for a given combination of task name and inputs.

Parameters
  • name (str) – Task name.

  • inputs (str) – String of inputs.

pydra.engine.helpers.custom_validator(instance, attribute, value)[source]

simple custom validation take into account ty.Union, ty.List, ty.Dict (but only one level depth) adding an additional validator, if allowe_values provided

pydra.engine.helpers.ensure_list(obj, tuple2list=False)[source]

Return a list whatever the input object is.

Examples

>>> ensure_list(list("abc"))
['a', 'b', 'c']
>>> ensure_list("abc")
['abc']
>>> ensure_list(tuple("abc"))
[('a', 'b', 'c')]
>>> ensure_list(tuple("abc"), tuple2list=True)
['a', 'b', 'c']
>>> ensure_list(None)
[]
>>> ensure_list(5.0)
[5.0]
pydra.engine.helpers.execute(cmd, strip=False)[source]

Run the event loop with coroutine.

Uses read_and_display_async() unless a loop is already running, in which case read_and_display() is used.

Parameters
  • cmd (list or tuple) – The command line to be executed.

  • strip (bool) – TODO

pydra.engine.helpers.gather_runtime_info(fname)[source]

Extract runtime information from a file.

Parameters

fname (os.pathlike) – The file containing runtime information

Returns

runtime – A runtime object containing the collected information.

Return type

Runtime

pydra.engine.helpers.get_available_cpus()[source]

Return the number of CPUs available to the current process or, if that is not available, the total number of CPUs on the system.

Returns

n_proc – The number of available CPUs.

Return type

int

pydra.engine.helpers.get_open_loop()[source]

Get current event loop.

If the loop is closed, a new loop is created and set as the current event loop.

Returns

loop – The current event loop

Return type

asyncio.EventLoop

pydra.engine.helpers.hash_function(obj)[source]

Generate hash of object.

pydra.engine.helpers.hash_value(value, tp=None, metadata=None, precalculated=None)[source]

calculating hash or returning values recursively

pydra.engine.helpers.load_and_run(task_pkl, ind=None, rerun=False, submitter=None, plugin=None, **kwargs)[source]

loading a task from a pickle file, settings proper input and running the task

async pydra.engine.helpers.load_and_run_async(task_pkl, ind=None, submitter=None, rerun=False, **kwargs)[source]

loading a task from a pickle file, settings proper input and running the workflow

pydra.engine.helpers.load_result(checksum, cache_locations)[source]

Restore a result from the cache.

Parameters
  • checksum (str) – Unique identifier of the task to be loaded.

  • cache_locations (list of os.pathlike) – List of cache directories, in order of priority, where the checksum will be looked for.

pydra.engine.helpers.load_task(task_pkl, ind=None)[source]

loading a task from a pickle file, settings proper input for the specific ind

pydra.engine.helpers.make_klass(spec)[source]

Create a data class given a spec.

Parameters

spec – TODO

pydra.engine.helpers.output_from_inputfields(output_spec, input_spec)[source]

Collect values from output from input fields. If names_only is False, the output_spec is updated, if names_only is True only the names are returned

Parameters
  • output_spec – TODO

  • input_spec – TODO

pydra.engine.helpers.position_sort(args)[source]

Sort objects by position, following Python indexing conventions.

Ordering is positive positions, lowest to highest, followed by unspecified positions (None) and negative positions, lowest to highest.

>>> position_sort([(None, "d"), (-3, "e"), (2, "b"), (-2, "f"), (5, "c"), (1, "a")])
['a', 'b', 'c', 'd', 'e', 'f']
Parameters

args (list of (int/None, object) tuples)

Returns

Return type

list of objects

pydra.engine.helpers.print_help(obj)[source]

Visit a task object and print its input/output interface.

pydra.engine.helpers.read_and_display(*cmd, strip=False, hide_display=False)[source]

Capture a process’ standard output.

async pydra.engine.helpers.read_and_display_async(*cmd, hide_display=False, strip=False)[source]

Capture standard input and output of a process, displaying them as they arrive.

Works line-by-line.

async pydra.engine.helpers.read_stream_and_display(stream, display)[source]

Read from stream line by line until EOF, display, and capture the lines.

pydra.engine.helpers.record_error(error_path, error)[source]

Write an error file.

pydra.engine.helpers.save(task_path: Path, result=None, task=None, name_prefix=None)[source]

Save a TaskBase object and/or results.

Parameters
  • task_path (Path) – Write directory

  • result (Result) – Result to pickle and write

  • task (TaskBase) – Task to pickle and write

pydra.engine.helpers.task_hash(task)[source]

Calculate the checksum of a task.

input hash, output hash, environment hash

Parameters

task (TaskBase) – The input task.