Mercurial > repos > shellac > sam_consensus_v3
diff env/lib/python3.9/site-packages/schema_salad/utils.py @ 0:4f3585e2f14b draft default tip
"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"
author | shellac |
---|---|
date | Mon, 22 Mar 2021 18:12:50 +0000 (2021-03-22) |
parents | |
children |
line wrap: on
line diff
--- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/env/lib/python3.9/site-packages/schema_salad/utils.py Mon Mar 22 18:12:50 2021 +0000 @@ -0,0 +1,110 @@ +import json +import os +from typing import ( + IO, + TYPE_CHECKING, + Any, + Callable, + Dict, + Iterable, + Mapping, + MutableSequence, + Tuple, + TypeVar, + Union, +) + +import requests +from rdflib.graph import Graph +from ruamel.yaml.comments import CommentedMap, CommentedSeq + +if TYPE_CHECKING: + from .fetcher import Fetcher + +ContextType = Dict[str, Union[Dict[str, Any], str, Iterable[str]]] +DocumentType = TypeVar("DocumentType", CommentedSeq, CommentedMap) +DocumentOrStrType = TypeVar("DocumentOrStrType", CommentedSeq, CommentedMap, str) +FieldType = TypeVar("FieldType", str, CommentedSeq, CommentedMap) +ResolveType = Union[int, float, str, CommentedMap, CommentedSeq, None] +ResolvedRefType = Tuple[ResolveType, CommentedMap] +IdxResultType = Union[CommentedMap, CommentedSeq, str, None] +IdxType = Dict[str, IdxResultType] +CacheType = Dict[str, Union[str, Graph, bool]] +FetcherCallableType = Callable[[CacheType, requests.sessions.Session], "Fetcher"] +AttachmentsType = Callable[[Union[CommentedMap, CommentedSeq]], bool] + + +def add_dictlist(di, key, val): # type: (Dict[Any, Any], Any, Any) -> None + if key not in di: + di[key] = [] + di[key].append(val) + + +def aslist(thing): # type: (Any) -> MutableSequence[Any] + """ + Convenience function to wrap single items and lists. + + Return lists unchanged. + """ + + if isinstance(thing, MutableSequence): + return thing + else: + return [thing] + + +# http://rightfootin.blogspot.com/2006/09/more-on-python-flatten.html + + +def flatten(thing, ltypes=(list, tuple)): + # type: (Any, Any) -> Any + if thing is None: + return [] + if not isinstance(thing, ltypes): + return [thing] + + ltype = type(thing) + lst = list(thing) + i = 0 + while i < len(lst): + while isinstance(lst[i], ltypes): + if not lst[i]: + lst.pop(i) + i -= 1 + break + else: + lst[i : i + 1] = lst[i] + i += 1 + return ltype(lst) + + +# Check if we are on windows OS +def onWindows(): + # type: () -> (bool) + return os.name == "nt" + + +def convert_to_dict(j4): # type: (Any) -> Any + if isinstance(j4, Mapping): + return {k: convert_to_dict(v) for k, v in j4.items()} + elif isinstance(j4, MutableSequence): + return [convert_to_dict(v) for v in j4] + else: + return j4 + + +def json_dump( + obj, # type: Any + fp, # type: IO[str] + **kwargs # type: Any +): # type: (...) -> None + """ Force use of unicode. """ + json.dump(convert_to_dict(obj), fp, **kwargs) + + +def json_dumps( + obj, # type: Any + **kwargs # type: Any +): # type: (...) -> str + """ Force use of unicode. """ + return json.dumps(convert_to_dict(obj), **kwargs)