cwlkernel Module

CoreExecutor

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)

CWLExecuteConfigurator

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>>)}

kernel_magics

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.kernel_magics.data(kernel: 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

IOManager

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]

clear()

files_counter

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

CWLLogger

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]]

save()

to_dict()

CWLBuilder

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

clear()

get_current_code() → str

CWLKernel

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.

history

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

results_manager

runtime_directory

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)

workflow_repository

CWLLoggerStorageManager

class cwlkernel.CWLLoggerStorageManager.CWLLoggerStorageManager(root_directory)

get_storage_path() → str

load(limit=None) → Iterator[Dict[KT, VT]]

save(logs) → str