diff env/lib/python3.7/site-packages/cwltool/utils.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/utils.py	Thu May 14 16:47:39 2020 -0400
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,265 +0,0 @@
-"""Shared functions and other definitions."""
-from __future__ import absolute_import
-
-import collections
-import os
-import platform
-import random
-import shutil
-import string
-import sys
-import tempfile
-from functools import partial  # pylint: disable=unused-import
-from typing import (IO, Any, AnyStr, Callable,  # pylint: disable=unused-import
-                    Dict, Iterable, List, MutableMapping, MutableSequence,
-                    Optional, Union)
-
-import pkg_resources
-from mypy_extensions import TypedDict
-from schema_salad.utils import json_dump, json_dumps  # pylint: disable=unused-import
-from six.moves import urllib, zip_longest
-from typing_extensions import Deque, Text  # pylint: disable=unused-import
-# move to a regular typing import when Python 3.3-3.6 is no longer supported
-
-from pathlib2 import Path
-
-# no imports from cwltool allowed
-if os.name == 'posix':
-    if sys.version_info < (3, 5):
-        import subprocess32 as subprocess  # nosec # pylint: disable=unused-import
-    else:
-        import subprocess  # nosec # pylint: disable=unused-import
-else:
-    import subprocess  # type: ignore  # nosec
-
-windows_default_container_id = "frolvlad/alpine-bash"
-
-Directory = TypedDict('Directory',
-                      {'class': Text, 'listing': List[Dict[Text, Text]],
-                       'basename': Text})
-
-DEFAULT_TMP_PREFIX = tempfile.gettempdir() + os.path.sep
-
-processes_to_kill = collections.deque()  # type: Deque[subprocess.Popen]
-
-def versionstring():
-    # type: () -> Text
-    """Version of CWLtool used to execute the workflow."""
-    pkg = pkg_resources.require("cwltool")
-    if pkg:
-        return u"%s %s" % (sys.argv[0], pkg[0].version)
-    return u"%s %s" % (sys.argv[0], "unknown version")
-
-def aslist(l):  # type: (Any) -> MutableSequence[Any]
-    """Wrap any non-MutableSequence/list in a list."""
-    if isinstance(l, MutableSequence):
-        return l
-    return [l]
-
-def copytree_with_merge(src, dst):  # type: (Text, Text) -> None
-    if not os.path.exists(dst):
-        os.makedirs(dst)
-        shutil.copystat(src, dst)
-    lst = os.listdir(src)
-    for item in lst:
-        spath = os.path.join(src, item)
-        dpath = os.path.join(dst, item)
-        if os.path.isdir(spath):
-            copytree_with_merge(spath, dpath)
-        else:
-            shutil.copy2(spath, dpath)
-
-def docker_windows_path_adjust(path):
-    # type: (Text) -> Text
-    r"""
-    Adjust only windows paths for Docker.
-
-    The docker run command treats them as unix paths.
-
-    Example: 'C:\Users\foo to /C/Users/foo (Docker for Windows) or /c/Users/foo
-    (Docker toolbox).
-    """
-    if onWindows():
-        split = path.split(':')
-        if len(split) == 2:
-            if platform.win32_ver()[0] in ('7', '8'):  # type: ignore
-                split[0] = split[0].lower()  # Docker toolbox uses lowecase windows Drive letters
-            else:
-                split[0] = split[0].capitalize()
-                # Docker for Windows uses uppercase windows Drive letters
-            path = ':'.join(split)
-        path = path.replace(':', '').replace('\\', '/')
-        return path if path[0] == '/' else '/' + path
-    return path
-
-
-def docker_windows_reverse_path_adjust(path):
-    # type: (Text) -> (Text)
-    r"""
-    Change docker path (only on windows os) appropriately back to Windows path.
-
-    Example:  /C/Users/foo to C:\Users\foo
-    """
-    if path is not None and onWindows():
-        if path[0] == '/':
-            path = path[1:]
-        else:
-            raise ValueError("not a docker path")
-        splitpath = path.split('/')
-        splitpath[0] = splitpath[0]+':'
-        return '\\'.join(splitpath)
-    return path
-
-
-def docker_windows_reverse_fileuri_adjust(fileuri):
-    # type: (Text) -> (Text)
-    r"""
-    Convert fileuri to be MS Windows comptabile, if needed.
-
-    On docker in windows fileuri do not contain : in path
-    To convert this file uri to windows compatible add : after drive letter,
-    so file:///E/var becomes file:///E:/var
-    """
-    if fileuri is not None and onWindows():
-        if urllib.parse.urlsplit(fileuri).scheme == "file":
-            filesplit = fileuri.split("/")
-            if filesplit[3][-1] != ':':
-                filesplit[3] = filesplit[3]+':'
-                return '/'.join(filesplit)
-            return fileuri
-        raise ValueError("not a file URI")
-    return fileuri
-
-
-def onWindows():
-    # type: () -> (bool)
-    """Check if we are on Windows OS."""
-    return os.name == 'nt'
-
-
-def convert_pathsep_to_unix(path):  # type: (Text) -> (Text)
-    """
-    Convert path seperators to unix style.
-
-    On windows os.path.join would use backslash to join path, since we would
-    use these paths in Docker we would convert it to use forward slashes: /
-    """
-    if path is not None and onWindows():
-        return path.replace('\\', '/')
-    return path
-
-def cmp_like_py2(dict1, dict2):  # type: (Dict[Text, Any], Dict[Text, Any]) -> int
-    """
-    Compare in the same manner as Python2.
-
-    Comparision function to be used in sorting as python3 doesn't allow sorting
-    of different types like str() and int().
-    This function re-creates sorting nature in py2 of heterogeneous list of
-    `int` and `str`
-    """
-    # extract lists from both dicts
-    first, second = dict1["position"], dict2["position"]
-    # iterate through both list till max of their size
-    for i, j in zip_longest(first, second):
-        if i == j:
-            continue
-        # in case 1st list is smaller
-        # should come first in sorting
-        if i is None:
-            return -1
-        # if 1st list is longer,
-        # it should come later in sort
-        elif j is None:
-            return 1
-
-        # if either of the list contains str element
-        # at any index, both should be str before comparing
-        if isinstance(i, str) or isinstance(j, str):
-            return 1 if str(i) > str(j) else -1
-        # int comparison otherwise
-        return 1 if i > j else -1
-    # if both lists are equal
-    return 0
-
-
-def bytes2str_in_dicts(inp  # type: Union[MutableMapping[Text, Any], MutableSequence[Any], Any]
-                      ):
-    # type: (...) -> Union[Text, MutableSequence[Any], MutableMapping[Text, Any]]
-    """
-    Convert any present byte string to unicode string, inplace.
-
-    input is a dict of nested dicts and lists
-    """
-    # if input is dict, recursively call for each value
-    if isinstance(inp, MutableMapping):
-        for k in inp:
-            inp[k] = bytes2str_in_dicts(inp[k])
-        return inp
-
-    # if list, iterate through list and fn call
-    # for all its elements
-    if isinstance(inp, MutableSequence):
-        for idx, value in enumerate(inp):
-            inp[idx] = bytes2str_in_dicts(value)
-            return inp
-
-    # if value is bytes, return decoded string,
-    elif isinstance(inp, bytes):
-        return inp.decode('utf-8')
-
-    # simply return elements itself
-    return inp
-
-
-def visit_class(rec, cls, op):
-    # type: (Any, Iterable[Any], Union[Callable[..., Any], partial[Any]]) -> None
-    """Apply a function to with "class" in cls."""
-    if isinstance(rec, MutableMapping):
-        if "class" in rec and rec.get("class") in cls:
-            op(rec)
-        for d in rec:
-            visit_class(rec[d], cls, op)
-    if isinstance(rec, MutableSequence):
-        for d in rec:
-            visit_class(d, cls, op)
-
-def visit_field(rec, field, op):
-    # type: (Any, Text, Union[Callable[..., Any], partial[Any]]) -> None
-    """Apply a function to mapping with 'field'."""
-    if isinstance(rec, MutableMapping):
-        if field in rec:
-            rec[field] = op(rec[field])
-        for d in rec:
-            visit_field(rec[d], field, op)
-    if isinstance(rec, MutableSequence):
-        for d in rec:
-            visit_field(d, field, op)
-
-
-def random_outdir():  # type: () -> Text
-    """Return the random directory name chosen to use for tool / workflow output."""
-    # compute this once and store it as a function attribute - each subsequent call will return the same value
-    if not hasattr(random_outdir, 'outdir'):
-        random_outdir.outdir = '/' + ''.join([random.choice(string.ascii_letters) for _ in range(6)])  # type: ignore  # nosec
-    return random_outdir.outdir  # type: ignore
-
-#
-# Simple multi-platform (fcntl/msvrt) file locking wrapper
-#
-try:
-    import fcntl
-
-    def shared_file_lock(fd):  # type: (IO[Any]) -> None
-        fcntl.flock(fd.fileno(), fcntl.LOCK_SH)
-
-    def upgrade_lock(fd):  # type: (IO[Any]) -> None
-        fcntl.flock(fd.fileno(), fcntl.LOCK_EX)
-
-except ImportError:
-    import msvcrt
-
-    def shared_file_lock(fd):  # type: (IO[Any]) -> None
-        msvcrt.locking(fd.fileno(), msvcrt.LK_LOCK, 1024)  # type: ignore
-
-    def upgrade_lock(fd):  # type: (IO[Any]) -> None
-        pass