tasks

tasks

The tasks module holds the TaskFiles class which organizes input and output paths and the Task class for steps in the runmanager.

Classes

Name Description
AutoFillDict Dictionary that gets populated by calling get_missing.
Task A task to be performed as as a step in the RunManager.
TaskFiles Class for Task input and output files and directories.

AutoFillDict

tasks.AutoFillDict(self, get_missing)

Dictionary that gets populated by calling get_missing.

Task

tasks.Task(self, runmng, f, kwargs=None, out=None)

A task to be performed as as a step in the RunManager.

A task consists of a function and its keyword arguments. Calling a taks calls the stored function. The function must return a TaskFiles object.

Parameters

Name Type Description Default
runmng Runmanager instance from which the task is called required
f Callable[…, Optional[TaskFiles]] Function that will be called when the task is called required
kwargs Optional[dict[str, Any]] kwargs to be passed to f None
out Optional[str] If not None, an output dir will be created with this name None

TaskFiles

tasks.TaskFiles(self, get_latest, input=dict(), output=dict(), outputdir=Path(), logger=logging.getLogger('kimmdy.basetask'))

Class for Task input and output files and directories.

Hosts the input and output file paths belonging to a task. A function or method that wants to be callable as a Task has to return a TaskFiles object. The input defaultdict is populated on the fly using get_latest of the runmanager to find newest files. Files which can not be found by get_latest must be added manually.

Attributes

Name Type Description
get_latest Callable Runmanager.get_latest function that returns paths to the latest file of given type.
input dict[str, Path] Input file paths for a Task. Is populated by get_latest or manually.
output dict[str, Path] Output file paths for a Task. Is populated by runmanager._discover_output_files or manually.
outputdir Path Output directory for a Task. Typically populated by create_task_directory called by Task.
logger logging.Logger Logger for a Task. Initialized in create_task_directory.

Examples

>>> class run():
>>>     def get_latest(self, s):
>>>         return f"latest {s}"
>>> runmng = run()
>>> files = TaskFiles(runmng)
>>> files.input
>>> files.input["top"]
{'top': 'latest top'}

Functions

Name Description
create_task_directory Creates TaskFiles object, output directory, logger and symlinks ff.

create_task_directory

tasks.create_task_directory(runmng, postfix)

Creates TaskFiles object, output directory, logger and symlinks ff.

Gets called when a Task is called (from the runmanager.tasks queue).

Back to top