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]¶
-
static
-
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:
-
static
-
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)¶
-
static
-
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}¶
-
classmethod
-
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
()¶
-
classmethod
CWLBuilder¶
CWLKernel¶
-
class
cwlkernel.CWLKernel.
CWLKernel
(**kwargs)¶ Jupyter Notebook kernel for CWL.
-
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
¶
-