Mercurial > repos > shellac > guppy_basecaller
diff env/lib/python3.7/site-packages/cwltool/update.py @ 5:9b1c78e6ba9c draft default tip
"planemo upload commit 6c0a8142489327ece472c84e558c47da711a9142"
author | shellac |
---|---|
date | Mon, 01 Jun 2020 08:59:25 -0400 |
parents | 79f47841a781 |
children |
line wrap: on
line diff
--- a/env/lib/python3.7/site-packages/cwltool/update.py Thu May 14 16:47:39 2020 -0400 +++ /dev/null Thu Jan 01 00:00:00 1970 +0000 @@ -1,203 +0,0 @@ -from __future__ import absolute_import - -import copy -import re -from typing import (Any, Callable, Dict, List, MutableMapping, MutableSequence, - Optional, Tuple, Union) - -from functools import partial - -from ruamel.yaml.comments import CommentedMap, CommentedSeq -from schema_salad import validate -from schema_salad.ref_resolver import Loader # pylint: disable=unused-import -from six import string_types -from six.moves import urllib -from typing_extensions import Text -from schema_salad.sourceline import SourceLine -from .loghandler import _logger - -# move to a regular typing import when Python 3.3-3.6 is no longer supported - -from .utils import visit_class, visit_field, aslist - - -def v1_0to1_1(doc, loader, baseuri): # pylint: disable=unused-argument - # type: (Any, Loader, Text) -> Tuple[Any, Text] - """Public updater for v1.0 to v1.1.""" - doc = copy.deepcopy(doc) - - rewrite = { - "http://commonwl.org/cwltool#WorkReuse": "WorkReuse", - "http://arvados.org/cwl#ReuseRequirement": "WorkReuse", - "http://commonwl.org/cwltool#TimeLimit": "ToolTimeLimit", - "http://commonwl.org/cwltool#NetworkAccess": "NetworkAccess", - "http://commonwl.org/cwltool#InplaceUpdateRequirement": "InplaceUpdateRequirement", - "http://commonwl.org/cwltool#LoadListingRequirement": "LoadListingRequirement" - } - def rewrite_requirements(t): # type: (MutableMapping[Text, Union[Text, Dict[Text, Any]]]) -> None - if "requirements" in t: - for r in t["requirements"]: - if isinstance(r, MutableMapping): - if r["class"] in rewrite: - r["class"] = rewrite[r["class"]] - else: - raise validate.ValidationException( - "requirements entries must be dictionaries: {} {}.".format( - type(r), r)) - if "hints" in t: - for r in t["hints"]: - if isinstance(r, MutableMapping): - if r["class"] in rewrite: - r["class"] = rewrite[r["class"]] - else: - raise validate.ValidationException( - "hints entries must be dictionaries: {} {}.".format( - type(r), r)) - if "steps" in t: - for s in t["steps"]: - if isinstance(s, MutableMapping): - rewrite_requirements(s) - else: - raise validate.ValidationException( - "steps entries must be dictionaries: {} {}.".format( - type(s), s)) - - - def update_secondaryFiles(t, top=False): - # type: (Any, bool) -> Union[MutableSequence[MutableMapping[Text, Text]], MutableMapping[Text, Text]] - if isinstance(t, CommentedSeq): - new_seq = copy.deepcopy(t) - for index, entry in enumerate(t): - new_seq[index] = update_secondaryFiles(entry) - return new_seq - elif isinstance(t, MutableSequence): - return CommentedSeq([update_secondaryFiles(p) for p in t]) - elif isinstance(t, MutableMapping): - return t - elif top: - return CommentedSeq([CommentedMap([("pattern", t)])]) - else: - return CommentedMap([("pattern", t)]) - - def fix_inputBinding(t): # type: (Dict[Text, Any]) -> None - for i in t["inputs"]: - if "inputBinding" in i: - ib = i["inputBinding"] - for k in list(ib.keys()): - if k != "loadContents": - _logger.warning(SourceLine(ib, k).makeError( - "Will ignore field '{}' which is not valid in {} " - "inputBinding".format(k, t["class"]))) - del ib[k] - - visit_class(doc, ("CommandLineTool","Workflow"), rewrite_requirements) - visit_class(doc, ("ExpressionTool","Workflow"), fix_inputBinding) - visit_field(doc, "secondaryFiles", partial(update_secondaryFiles, top=True)) - - upd = doc - if isinstance(upd, MutableMapping) and "$graph" in upd: - upd = upd["$graph"] - for proc in aslist(upd): - proc.setdefault("hints", CommentedSeq()) - proc["hints"].insert(0, CommentedMap([("class", "NetworkAccess"),( "networkAccess", True)])) - proc["hints"].insert(0, CommentedMap([("class", "LoadListingRequirement"),("loadListing", "deep_listing")])) - if "cwlVersion" in proc: - del proc["cwlVersion"] - - return (doc, "v1.1") - -def v1_1_0dev1to1_1(doc, loader, baseuri): # pylint: disable=unused-argument - # type: (Any, Loader, Text) -> Tuple[Any, Text] - return (doc, "v1.1") - -UPDATES = { - u"v1.0": v1_0to1_1, - u"v1.1": None -} # type: Dict[Text, Optional[Callable[[Any, Loader, Text], Tuple[Any, Text]]]] - -DEVUPDATES = { - u"v1.0": v1_0to1_1, - u"v1.1.0-dev1": v1_1_0dev1to1_1, - u"v1.1": None -} # type: Dict[Text, Optional[Callable[[Any, Loader, Text], Tuple[Any, Text]]]] - -ALLUPDATES = UPDATES.copy() -ALLUPDATES.update(DEVUPDATES) - -INTERNAL_VERSION = u"v1.1" - -def identity(doc, loader, baseuri): # pylint: disable=unused-argument - # type: (Any, Loader, Text) -> Tuple[Any, Union[Text, Text]] - """Default, do-nothing, CWL document upgrade function.""" - return (doc, doc["cwlVersion"]) - - -def checkversion(doc, # type: Union[CommentedSeq, CommentedMap] - metadata, # type: CommentedMap - enable_dev # type: bool -): - # type: (...) -> Tuple[CommentedMap, Text] - """Check the validity of the version of the give CWL document. - - Returns the document and the validated version string. - """ - cdoc = None # type: Optional[CommentedMap] - if isinstance(doc, CommentedSeq): - if not isinstance(metadata, CommentedMap): - raise Exception("Expected metadata to be CommentedMap") - lc = metadata.lc - metadata = copy.deepcopy(metadata) - metadata.lc.data = copy.copy(lc.data) - metadata.lc.filename = lc.filename - metadata[u"$graph"] = doc - cdoc = metadata - elif isinstance(doc, CommentedMap): - cdoc = doc - else: - raise Exception("Expected CommentedMap or CommentedSeq") - - version = metadata[u"cwlVersion"] - cdoc["cwlVersion"] = version - - if version not in UPDATES: - if version in DEVUPDATES: - if enable_dev: - pass - else: - keys = list(UPDATES.keys()) - keys.sort() - raise validate.ValidationException( - u"Version '%s' is a development or deprecated version.\n " - "Update your document to a stable version (%s) or use " - "--enable-dev to enable support for development and " - "deprecated versions." % (version, ", ".join(keys))) - else: - raise validate.ValidationException( - u"Unrecognized version %s" % version) - - return (cdoc, version) - - -def update(doc, loader, baseuri, enable_dev, metadata): - # type: (Union[CommentedSeq, CommentedMap], Loader, Text, bool, Any) -> CommentedMap - - if isinstance(doc, CommentedMap): - if metadata.get("http://commonwl.org/cwltool#original_cwlVersion") \ - or doc.get("http://commonwl.org/cwltool#original_cwlVersion"): - return doc - - (cdoc, version) = checkversion(doc, metadata, enable_dev) - originalversion = copy.copy(version) - - nextupdate = identity # type: Optional[Callable[[Any, Loader, Text], Tuple[Any, Text]]] - - while nextupdate: - (cdoc, version) = nextupdate(cdoc, loader, baseuri) - nextupdate = ALLUPDATES[version] - - cdoc[u"cwlVersion"] = version - metadata[u"cwlVersion"] = version - metadata[u"http://commonwl.org/cwltool#original_cwlVersion"] = originalversion - cdoc[u"http://commonwl.org/cwltool#original_cwlVersion"] = originalversion - - return cdoc