cwlkernel Module


class cwlkernel.CoreExecutor.CoreExecutor(file_manager: cwlkernel.IOManager.IOFileManager, provenance_directory: Optional[pathlib.Path])
execute(provenance=False) → Tuple[uuid.UUID, Dict[KT, VT], Optional[Exception], Optional[cwltool.provenance.ResearchObject]]
Parameters:provenance – Execute with provenance enabled/disabled.
Returns:Run ID, dict with new files, exception if there is any.
set_data(data: List[str]) → List[str]
set_workflow_path(workflow_str: str) → str
Parameters:workflow_str – the cwl
Returns:the path where we executor stored the workflow
classmethod validate_input_files(yaml_input: Dict[KT, VT], cwd: pathlib.Path) → NoReturn
class cwlkernel.CoreExecutor.JupyterFactory(root_directory: str, executor: Optional[cwltool.executors.JobExecutor] = None, loading_context: Optional[cwltool.context.LoadingContext] = None, runtime_context: Optional[cwltool.context.RuntimeContext] = None)
class cwlkernel.CoreExecutor.ProvenanceFactory(workflow_uri_path: str, root_directory: str, stove_provenance_directory: str, executor: Optional[cwltool.executors.JobExecutor] = None, loading_context: Optional[cwltool.context.LoadingContext] = None, runtime_context: Optional[cwltool.context.RuntimeContext] = None)


class cwlkernel.CWLExecuteConfigurator.CWLExecuteConfigurator
properties = {'CWLKERNEL_BOOT_DIRECTORY': ('/tmp/CWLKERNEL_DATA', <function CWLExecuteConfigurator.<lambda>>), 'CWLKERNEL_MAGIC_COMMANDS_DIRECTORY': (None, <function CWLExecuteConfigurator.<lambda>>), 'CWLKERNEL_MODE': ('SIMPLE', <function CWLExecuteConfigurator.<lambda>>)}


class cwlkernel.kernel_magics.ExecutionMagics
static execute(kernel: cwlkernel.CWLKernel.CWLKernel, execute_argument_string: str)

Execute registered tool by id. % execute [tool-id] [yaml input …]

@param kernel: the kernel instance @param execute_argument_string: a multiple line string containins in the first line the tool id and in the next lines the input parameters in yaml syntax @return: None

static execute_with_provenance(kernel: cwlkernel.CWLKernel.CWLKernel, execute_argument_string: str)
static suggest_execution_id(query_token: str, *args, **kwargs) → List[str]
class cwlkernel.kernel_magics.MagicSnippetBuilder
static snippet(kernel: cwlkernel.CWLKernel.CWLKernel, command: str)

Submit a cwl workflow incrementally. Usage: % snippet add […] % snippet add […] % snippet build

@param kernel: @param command: @return:

class cwlkernel.kernel_magics.ManualWorkflowComposer
static new_workflow(kernel: cwlkernel.CWLKernel.CWLKernel, workflow_id: str)
static new_workflow_add_input(kernel: cwlkernel.CWLKernel.CWLKernel, args: str)
static new_workflow_add_output_source(kernel: cwlkernel.CWLKernel.CWLKernel, args: str)
static new_workflow_add_step(kernel: cwlkernel.CWLKernel.CWLKernel, ids: str)
static new_workflow_add_step_in(kernel: cwlkernel.CWLKernel.CWLKernel, args: str)
static new_workflow_build(kernel: cwlkernel.CWLKernel.CWLKernel, *args)
class cwlkernel.kernel_magics.Scatter
classmethod parse_args(args_line) → Tuple[str, str]
parser = ArgumentParser(prog='sphinx-build', usage=None, description=None, formatter_class=<class 'argparse.HelpFormatter'>, conflict_handler='error', add_help=True)
static scatter(kernel: cwlkernel.CWLKernel.CWLKernel, args_line: str)
scatter_template = {'class': 'Workflow', 'cwlVersion': None, 'inputs': None, 'outputs': None, 'requirements': {'ScatterFeatureRequirement': {}}, 'steps': None}
cwlkernel.kernel_magics.compile_executed_steps_as_workflow(kernel: cwlkernel.CWLKernel.CWLKernel, args: str)

Compose a workflow from executed workflows.

@param kernel: @param args: @return: cwlkernel.CWLKernel.CWLKernel, *args)

Display all the data which are registered in the kernel session.

cwlkernel.kernel_magics.display_data(kernel: cwlkernel.CWLKernel.CWLKernel, data_name: str) → None

Display the data generated by workflow. Usage % displayData [data id]

@param kernel: the kernel instance @param data_name: the data id @return None

cwlkernel.kernel_magics.display_data_csv(kernel: cwlkernel.CWLKernel.CWLKernel, data_name: str)
cwlkernel.kernel_magics.display_data_image(kernel: cwlkernel.CWLKernel.CWLKernel, data_name: str)
cwlkernel.kernel_magics.edit(kernel: cwlkernel.CWLKernel.CWLKernel, args: str) → Optional[Dict[KT, VT]]
cwlkernel.kernel_magics.github_import(kernel: cwlkernel.CWLKernel.CWLKernel, url: str)
cwlkernel.kernel_magics.logs(kernel: cwlkernel.CWLKernel.CWLKernel, limit=None)
cwlkernel.kernel_magics.magics(kernel: cwlkernel.CWLKernel.CWLKernel, arg: str)
cwlkernel.kernel_magics.sample_csv(kernel: cwlkernel.CWLKernel.CWLKernel, args: str)
cwlkernel.kernel_magics.system(kernel: cwlkernel.CWLKernel.CWLKernel, commands: str)

Execute bash commands in the Runtime Directory of the session.

@param kernel: @param commands: @return:

cwlkernel.kernel_magics.view_tool(kernel: cwlkernel.CWLKernel.CWLKernel, workflow_id: str)
cwlkernel.kernel_magics.visualize_graph(kernel: cwlkernel.CWLKernel.CWLKernel, tool_id: str)

Visualize a Workflow


class cwlkernel.IOManager.IOFileManager(root_directory: str)
append_files(files_to_copy: List[str], relative_path: str = '.', metadata: Optional[Dict[KT, VT]] = None) → List[str]
get_files() → List[str]
get_files_registry() → Dict[KT, VT]
get_files_uri() → urllib.parse.ParseResult
read(relative_path: str) → bytes
remove(path: str)
write(relative_path: str, binary_data: bytes, metadata=None) → str
class cwlkernel.IOManager.ResultsManager(root_directory: str)
get_last_result_by_id(result_id: str) → Optional[str]

The results manager may have multiple results with the same id, from multiple executions. That function will return the path of the last result @param result_id id to the Results manager. If the result_id has the format of path then the last goes to the id and the previous one to the produced by [_produced_by]/[result_id] @return: the path of last result with the requested id or None


class cwlkernel.CWLLogger.CWLLogger(root_directory)
classmethod collect_infrastructure_metrics() → NamedTuple
classmethod get_hostname() → str
classmethod get_running_kernels() → List[int]
Returns:A list with the process ids of running kernels
load(limit=None) → Iterator[Dict[KT, VT]]


class cwlkernel.CWLBuilder.CWLBuilder
build() → cwlkernel.cwlrepository.CWLComponent.WorkflowComponent
class cwlkernel.CWLBuilder.CWLSnippetBuilder
append(code: str, indent: int = 0) → None
build() → cwlkernel.cwlrepository.CWLComponent.WorkflowComponent
get_current_code() → str


class cwlkernel.CWLKernel.CWLKernel(**kwargs)

Jupyter Notebook kernel for CWL.

banner = 'Common Workflow Language'
do_complete(code: str, cursor_pos: int)

Override in subclasses to find completions.

do_execute(code: str, silent=False, store_history: bool = True, user_expressions=None, allow_stdin: bool = False) → Dict[KT, VT]

Execute user code. Must be overridden by subclasses.

get_past_results() → List[str]
get_pid() → Tuple[int, int]
Returns:The process id and his parents id.

Returns a list of executed cells in the current session. The first item has the value “magic”/”register” and the second the code

implementation = 'CWLKernel'
implementation_version = '0.0.4'
language_info = {'file_extension': '.cwl', 'mimetype': 'text/x-cwl', 'name': 'yaml'}
language_version = '1.1'
class register_magic(magics_name: Optional[str] = None)

Registers magic commands. That method should be used as a decorator to register custom magic commands.

class register_magics_suggester(magic_command_name: str)

Decorator for registering functions for suggesting commands line arguments

send_error_response(text) → None

Sends a response to the jupyter notebook’s stderr. @param text: The message to display @return: None

send_json_response(json_data: Union[Dict[KT, VT], List[T]]) → None

Display a Dict or a List object as a JSON. The object must be json dumpable to use that function. @param json_data: Data to print in Jupyter Notebook @return: None

send_text_to_stdout(text: str)


class cwlkernel.CWLLoggerStorageManager.CWLLoggerStorageManager(root_directory)
get_storage_path() → str
load(limit=None) → Iterator[Dict[KT, VT]]
save(logs) → str