Mercurial > repos > shellac > sam_consensus_v3
comparison env/lib/python3.9/site-packages/cwltool/pathmapper.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 |
parents | |
children |
comparison
equal
deleted
inserted
replaced
-1:000000000000 | 0:4f3585e2f14b |
---|---|
1 import collections | |
2 import logging | |
3 import os | |
4 import stat | |
5 import urllib | |
6 import uuid | |
7 from typing import Dict, Iterator, List, Optional, Tuple, cast | |
8 | |
9 from schema_salad.exceptions import ValidationException | |
10 from schema_salad.ref_resolver import uri_file_path | |
11 from schema_salad.sourceline import SourceLine | |
12 | |
13 from .loghandler import _logger | |
14 from .stdfsaccess import abspath | |
15 from .utils import CWLObjectType, convert_pathsep_to_unix, dedup, downloadHttpFile | |
16 | |
17 MapperEnt = collections.namedtuple( | |
18 "MapperEnt", ["resolved", "target", "type", "staged"] | |
19 ) | |
20 | |
21 | |
22 class PathMapper: | |
23 """ | |
24 Mapping of files from relative path provided in the file to a tuple. | |
25 | |
26 (absolute local path, absolute container path) | |
27 | |
28 The tao of PathMapper: | |
29 | |
30 The initializer takes a list of File and Directory objects, a base | |
31 directory (for resolving relative references) and a staging directory | |
32 (where the files are mapped to). | |
33 | |
34 The purpose of the setup method is to determine where each File or | |
35 Directory should be placed on the target file system (relative to | |
36 stagedir). | |
37 | |
38 If separatedirs=True, unrelated files will be isolated in their own | |
39 directories under stagedir. If separatedirs=False, files and directories | |
40 will all be placed in stagedir (with the possibility for name | |
41 collisions...) | |
42 | |
43 The path map maps the "location" of the input Files and Directory objects | |
44 to a tuple (resolved, target, type). The "resolved" field is the "real" | |
45 path on the local file system (after resolving relative paths and | |
46 traversing symlinks). The "target" is the path on the target file system | |
47 (under stagedir). The type is the object type (one of File, Directory, | |
48 CreateFile, WritableFile, CreateWritableFile). | |
49 | |
50 The latter three (CreateFile, WritableFile, CreateWritableFile) are used by | |
51 InitialWorkDirRequirement to indicate files that are generated on the fly | |
52 (CreateFile and CreateWritableFile, in this case "resolved" holds the file | |
53 contents instead of the path because they file doesn't exist) or copied | |
54 into the output directory so they can be opened for update ("r+" or "a") | |
55 (WritableFile and CreateWritableFile). | |
56 | |
57 """ | |
58 | |
59 def __init__( | |
60 self, | |
61 referenced_files: List[CWLObjectType], | |
62 basedir: str, | |
63 stagedir: str, | |
64 separateDirs: bool = True, | |
65 ) -> None: | |
66 """Initialize the PathMapper.""" | |
67 self._pathmap = {} # type: Dict[str, MapperEnt] | |
68 self.stagedir = stagedir | |
69 self.separateDirs = separateDirs | |
70 self.setup(dedup(referenced_files), basedir) | |
71 | |
72 def visitlisting( | |
73 self, | |
74 listing: List[CWLObjectType], | |
75 stagedir: str, | |
76 basedir: str, | |
77 copy: bool = False, | |
78 staged: bool = False, | |
79 ) -> None: | |
80 for ld in listing: | |
81 self.visit( | |
82 ld, | |
83 stagedir, | |
84 basedir, | |
85 copy=cast(bool, ld.get("writable", copy)), | |
86 staged=staged, | |
87 ) | |
88 | |
89 def visit( | |
90 self, | |
91 obj: CWLObjectType, | |
92 stagedir: str, | |
93 basedir: str, | |
94 copy: bool = False, | |
95 staged: bool = False, | |
96 ) -> None: | |
97 stagedir = cast(Optional[str], obj.get("dirname")) or stagedir | |
98 tgt = convert_pathsep_to_unix( | |
99 os.path.join( | |
100 stagedir, | |
101 cast(str, obj["basename"]), | |
102 ) | |
103 ) | |
104 if obj["location"] in self._pathmap: | |
105 return | |
106 if obj["class"] == "Directory": | |
107 location = cast(str, obj["location"]) | |
108 if location.startswith("file://"): | |
109 resolved = uri_file_path(location) | |
110 else: | |
111 resolved = location | |
112 self._pathmap[location] = MapperEnt( | |
113 resolved, tgt, "WritableDirectory" if copy else "Directory", staged | |
114 ) | |
115 if location.startswith("file://"): | |
116 staged = False | |
117 self.visitlisting( | |
118 cast(List[CWLObjectType], obj.get("listing", [])), | |
119 tgt, | |
120 basedir, | |
121 copy=copy, | |
122 staged=staged, | |
123 ) | |
124 elif obj["class"] == "File": | |
125 path = cast(str, obj["location"]) | |
126 ab = abspath(path, basedir) | |
127 if "contents" in obj and path.startswith("_:"): | |
128 self._pathmap[path] = MapperEnt( | |
129 obj["contents"], | |
130 tgt, | |
131 "CreateWritableFile" if copy else "CreateFile", | |
132 staged, | |
133 ) | |
134 else: | |
135 with SourceLine( | |
136 obj, | |
137 "location", | |
138 ValidationException, | |
139 _logger.isEnabledFor(logging.DEBUG), | |
140 ): | |
141 deref = ab | |
142 if urllib.parse.urlsplit(deref).scheme in ["http", "https"]: | |
143 deref = downloadHttpFile(path) | |
144 else: | |
145 # Dereference symbolic links | |
146 st = os.lstat(deref) | |
147 while stat.S_ISLNK(st.st_mode): | |
148 rl = os.readlink(deref) | |
149 deref = ( | |
150 rl | |
151 if os.path.isabs(rl) | |
152 else os.path.join(os.path.dirname(deref), rl) | |
153 ) | |
154 st = os.lstat(deref) | |
155 | |
156 self._pathmap[path] = MapperEnt( | |
157 deref, tgt, "WritableFile" if copy else "File", staged | |
158 ) | |
159 self.visitlisting( | |
160 cast(List[CWLObjectType], obj.get("secondaryFiles", [])), | |
161 stagedir, | |
162 basedir, | |
163 copy=copy, | |
164 staged=staged, | |
165 ) | |
166 | |
167 def setup(self, referenced_files: List[CWLObjectType], basedir: str) -> None: | |
168 | |
169 # Go through each file and set the target to its own directory along | |
170 # with any secondary files. | |
171 stagedir = self.stagedir | |
172 for fob in referenced_files: | |
173 if self.separateDirs: | |
174 stagedir = os.path.join(self.stagedir, "stg%s" % uuid.uuid4()) | |
175 self.visit( | |
176 fob, | |
177 stagedir, | |
178 basedir, | |
179 copy=cast(bool, fob.get("writable", False)), | |
180 staged=True, | |
181 ) | |
182 | |
183 def mapper(self, src: str) -> MapperEnt: | |
184 if "#" in src: | |
185 i = src.index("#") | |
186 p = self._pathmap[src[:i]] | |
187 return MapperEnt(p.resolved, p.target + src[i:], p.type, p.staged) | |
188 return self._pathmap[src] | |
189 | |
190 def files(self) -> List[str]: | |
191 return list(self._pathmap.keys()) | |
192 | |
193 def items(self) -> List[Tuple[str, MapperEnt]]: | |
194 return list(self._pathmap.items()) | |
195 | |
196 def reversemap( | |
197 self, | |
198 target: str, | |
199 ) -> Optional[Tuple[str, str]]: | |
200 """Find the (source, resolved_path) for the given target, if any.""" | |
201 for k, v in self._pathmap.items(): | |
202 if v[1] == target: | |
203 return (k, v[0]) | |
204 return None | |
205 | |
206 def update( | |
207 self, key: str, resolved: str, target: str, ctype: str, stage: bool | |
208 ) -> MapperEnt: | |
209 m = MapperEnt(resolved, target, ctype, stage) | |
210 self._pathmap[key] = m | |
211 return m | |
212 | |
213 def __contains__(self, key: str) -> bool: | |
214 """Test for the presence of the given relative path in this mapper.""" | |
215 return key in self._pathmap | |
216 | |
217 def __iter__(self) -> Iterator[MapperEnt]: | |
218 """Get iterator for the maps.""" | |
219 return self._pathmap.values().__iter__() |