Mercurial > repos > shellac > guppy_basecaller

comparison env/lib/python3.7/site-packages/cwltool/context.py @ 2:6af9afd405e9 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit 0a63dd5f4d38a1f6944587f52a8cd79874177fc1"
author shellac
date Thu, 14 May 2020 14:56:58 -0400 (2020-05-14)
parents 26e78fe6e8c4
children
comparison
equal deleted inserted replaced
1:75ca89e9b81c 2:6af9afd405e9
1 """Shared context objects that replace use of kwargs."""
2 import copy
3 import threading # pylint: disable=unused-import
4 from typing import (Any, Callable, Dict, Iterable, List, MutableMapping,
5 Optional)
6
7 from schema_salad import schema
8 from schema_salad.ref_resolver import (ContextType, # pylint: disable=unused-import
9 Fetcher, Loader)
10 from typing_extensions import (TYPE_CHECKING, # pylint: disable=unused-import
11 Text)
12 # move to a regular typing import when Python 3.3-3.6 is no longer supported
13 from .builder import Builder, HasReqsHints
14 from .mutation import MutationManager
15 from .pathmapper import PathMapper
16 from .secrets import SecretStore
17 from .software_requirements import DependenciesConfiguration
18 from .stdfsaccess import StdFsAccess
19 from .utils import DEFAULT_TMP_PREFIX
20
21 if TYPE_CHECKING:
22 from .process import Process
23 from .provenance import (ResearchObject, # pylint: disable=unused-import
24 ProvenanceProfile)
25
26 class ContextBase(object):
27 def __init__(self, kwargs=None): # type: (Optional[Dict[str, Any]]) -> None
28 """Initialize."""
29 if kwargs:
30 for k, v in kwargs.items():
31 if hasattr(self, k):
32 setattr(self, k, v)
33
34 def make_tool_notimpl(toolpath_object, # type: MutableMapping[Text, Any]
35 loadingContext # type: LoadingContext
36 ): # type: (...) -> Process
37 raise NotImplementedError()
38
39
40 default_make_tool = make_tool_notimpl # type: Callable[[MutableMapping[Text, Any], LoadingContext], Process]
41
42 class LoadingContext(ContextBase):
43
44 def __init__(self, kwargs=None): # type: (Optional[Dict[str, Any]]) -> None
45 """Initialize the LoadingContext from the kwargs."""
46 self.debug = False # type: bool
47 self.metadata = {} # type: Dict[Text, Any]
48 self.requirements = None # type: Optional[List[Dict[Text, Any]]]
49 self.hints = None # type: Optional[List[Dict[Text, Any]]]
50 self.overrides_list = [] # type: List[Dict[Text, Any]]
51 self.loader = None # type: Optional[Loader]
52 self.avsc_names = None # type: Optional[schema.Names]
53 self.disable_js_validation = False # type: bool
54 self.js_hint_options_file = None
55 self.do_validate = True # type: bool
56 self.enable_dev = False # type: bool
57 self.strict = True # type: bool
58 self.resolver = None
59 self.fetcher_constructor = None
60 self.construct_tool_object = default_make_tool
61 self.research_obj = None # type: Optional[ResearchObject]
62 self.orcid = '' # type: str
63 self.cwl_full_name = "" # type: str
64 self.host_provenance = False # type: bool
65 self.user_provenance = False # type: bool
66 self.prov_obj = None # type: Optional[ProvenanceProfile]
67 self.do_update = None # type: Optional[bool]
68 self.jobdefaults = None # type: Optional[MutableMapping[Text, Any]]
69
70 super(LoadingContext, self).__init__(kwargs)
71
72 def copy(self):
73 # type: () -> LoadingContext
74 return copy.copy(self)
75
76 class RuntimeContext(ContextBase):
77 def __init__(self, kwargs=None): # type: (Optional[Dict[str, Any]]) -> None
78 """Initializet the RuntimeContext from the kwargs."""
79 select_resources_callable = Callable[ # pylint: disable=unused-variable
80 [Dict[str, int], RuntimeContext], Dict[str, int]]
81 self.user_space_docker_cmd = "" # type: Text
82 self.secret_store = None # type: Optional[SecretStore]
83 self.no_read_only = False # type: bool
84 self.custom_net = "" # type: Text
85 self.no_match_user = False # type: bool
86 self.preserve_environment = "" # type: Optional[Iterable[str]]
87 self.preserve_entire_environment = False # type: bool
88 self.use_container = True # type: bool
89 self.force_docker_pull = False # type: bool
90
91 self.tmp_outdir_prefix = DEFAULT_TMP_PREFIX # type: Text
92 self.tmpdir_prefix = DEFAULT_TMP_PREFIX # type: Text
93 self.tmpdir = "" # type: Text
94 self.rm_tmpdir = True # type: bool
95 self.pull_image = True # type: bool
96 self.rm_container = True # type: bool
97 self.move_outputs = "move" # type: Text
98
99 self.singularity = False # type: bool
100 self.disable_net = False # type: bool
101 self.debug = False # type: bool
102 self.compute_checksum = True # type: bool
103 self.name = "" # type: Text
104 self.default_container = "" # type: Text
105 self.find_default_container = None # type: Optional[Callable[[HasReqsHints], Optional[Text]]]
106 self.cachedir = None # type: Optional[Text]
107 self.outdir = None # type: Optional[Text]
108 self.stagedir = "" # type: Text
109 self.part_of = "" # type: Text
110 self.basedir = "" # type: Text
111 self.toplevel = False # type: bool
112 self.mutation_manager = None # type: Optional[MutationManager]
113 self.make_fs_access = StdFsAccess # type: Callable[[Text], StdFsAccess]
114 self.path_mapper = PathMapper
115 self.builder = None # type: Optional[Builder]
116 self.docker_outdir = "" # type: Text
117 self.docker_tmpdir = "" # type: Text
118 self.docker_stagedir = "" # type: Text
119 self.js_console = False # type: bool
120 self.job_script_provider = None # type: Optional[DependenciesConfiguration]
121 self.select_resources = None # type: Optional[select_resources_callable]
122 self.eval_timeout = 20 # type: float
123 self.postScatterEval = None # type: Optional[Callable[[MutableMapping[Text, Any]], Dict[Text, Any]]]
124 self.on_error = "stop" # type: Text
125 self.strict_memory_limit = False # type: bool
126
127 self.cidfile_dir = None
128 self.cidfile_prefix = None
129
130 self.workflow_eval_lock = None # type: Optional[threading.Condition]
131 self.research_obj = None # type: Optional[ResearchObject]
132 self.orcid = '' # type: str
133 self.cwl_full_name = "" # type: str
134 self.process_run_id = None # type: Optional[str]
135 self.prov_obj = None # type: Optional[ProvenanceProfile]
136 super(RuntimeContext, self).__init__(kwargs)
137
138
139 def copy(self):
140 # type: () -> RuntimeContext
141 return copy.copy(self)
142
143 def getdefault(val, default):
144 # type: (Any, Any) -> Any
145 if val is None:
146 return default
147 else:
148 return val