codekingpro/portable-devtools

1 2from __future__ import annotations3 4# partially from package six by Benjamin Peterson5 6import sys7import os8import io9from abc import abstractmethod10import collections.abc11 12 13from ruamel.yaml.docinfo import Version  # NOQA14# fmt: off15if False:  # MYPY16    from typing import Any, Dict, Optional, List, Union, BinaryIO, IO, Text, Tuple  # NOQA17    from typing import Optional  # NOQA18    try:19        from typing import SupportsIndex as SupportsIndex  # in order to reexport for mypy20    except ImportError:21        SupportsIndex = int  # type: ignore22 23    StreamType = Any24    StreamTextType = StreamType25    VersionType = Union[str , Tuple[int, int] , List[int] , Version , None]26# fmt: on27 28_DEFAULT_YAML_VERSION = (1, 2)29 30try:31    from collections import OrderedDict32except ImportError:33    from ordereddict import OrderedDict  # type: ignore34 35    # to get the right name import ... as ordereddict doesn't do that36 37 38class ordereddict(OrderedDict):  # type: ignore39    if not hasattr(OrderedDict, 'insert'):40 41        def insert(self, pos: int, key: Any, value: Any) -> None:42            if pos >= len(self):43                self[key] = value44                return45            od = ordereddict()46            od.update(self)47            for k in od:48                del self[k]49            for index, old_key in enumerate(od):50                if pos == index:51                    self[key] = value52                self[old_key] = od[old_key]53 54 55StringIO = io.StringIO56BytesIO = io.BytesIO57 58 59builtins_module = 'builtins'60 61 62def with_metaclass(meta: Any, *bases: Any) -> Any:63    """Create a base class with a metaclass."""64    return meta('NewBase', bases, {})65 66 67DBG_TOKEN = 168DBG_EVENT = 269DBG_NODE = 470 71 72_debug: Optional[int] = None73if 'RUAMEL_DEBUG' in os.environ:74    _debugx = os.environ.get('RUAMEL_DEBUG')75    if _debugx is None:76        _debug = 077    else:78        _debug = int(_debugx)79 80 81if bool(_debug):82 83    class ObjectCounter:84        def __init__(self) -> None:85            self.map: Dict[Any, Any] = {}86 87        def __call__(self, k: Any) -> None:88            self.map[k] = self.map.get(k, 0) + 189 90        def dump(self) -> None:91            for k in sorted(self.map):92                sys.stdout.write(f'{k} -> {self.map[k]}')93 94    object_counter = ObjectCounter()95 96 97# used from yaml util when testing98def dbg(val: Any = None) -> Any:99    debug = _debug100    if debug is None:101        # set to true or false102        _debugx = os.environ.get('YAMLDEBUG')103        if _debugx is None:104            debug = 0105        else:106            debug = int(_debugx)107    if val is None:108        return debug109    return debug & val110 111 112class Nprint:113    def __init__(self, file_name: Any = None) -> None:114        self._max_print: Any = None115        self._count: Any = None116        self._file_name = file_name117 118    def __call__(self, *args: Any, **kw: Any) -> None:119        if not bool(_debug):120            return121        import traceback122 123        out = sys.stdout if self._file_name is None else open(self._file_name, 'a')124        dbgprint = print  # to fool checking for print statements by dv utility125        kw1 = kw.copy()126        kw1['file'] = out127        dbgprint(*args, **kw1)128        out.flush()129        if self._max_print is not None:130            if self._count is None:131                self._count = self._max_print132            self._count -= 1133            if self._count == 0:134                dbgprint('forced exit\n')135                traceback.print_stack()136                out.flush()137                sys.exit(0)138        if self._file_name:139            out.close()140 141    def set_max_print(self, i: int) -> None:142        self._max_print = i143        self._count = None144 145    def fp(self, mode: str = 'a') -> Any:146        out = sys.stdout if self._file_name is None else open(self._file_name, mode)147        return out148 149 150nprint = Nprint()151nprintf = Nprint('/var/tmp/ruamel.yaml.log')152 153# char checkers following production rules154 155 156def check_namespace_char(ch: Any) -> bool:157    if '\x21' <= ch <= '\x7E':  # ! to ~158        return True159    if '\xA0' <= ch <= '\uD7FF':160        return True161    if ('\uE000' <= ch <= '\uFFFD') and ch != '\uFEFF':  # excl. byte order mark162        return True163    if '\U00010000' <= ch <= '\U0010FFFF':164        return True165    return False166 167 168def check_anchorname_char(ch: Any) -> bool:169    if ch in ',[]{}':170        return False171    return check_namespace_char(ch)172 173 174def version_tnf(t1: Any, t2: Any = None) -> Any:175    """176    return True if ruamel.yaml version_info < t1, None if t2 is specified and bigger else False177    """178    from ruamel.yaml import version_info  # NOQA179 180    if version_info < t1:181        return True182    if t2 is not None and version_info < t2:183        return None184    return False185 186 187class MutableSliceableSequence(collections.abc.MutableSequence):  # type: ignore188    __slots__ = ()189 190    def __getitem__(self, index: Any) -> Any:191        if not isinstance(index, slice):192            return self.__getsingleitem__(index)193        return type(self)([self[i] for i in range(*index.indices(len(self)))])  # type: ignore194 195    def __setitem__(self, index: Any, value: Any) -> None:196        if not isinstance(index, slice):197            return self.__setsingleitem__(index, value)198        assert iter(value)199        # nprint(index.start, index.stop, index.step, index.indices(len(self)))200        if index.step is None:201            del self[index.start : index.stop]202            for elem in reversed(value):203                self.insert(0 if index.start is None else index.start, elem)204        else:205            range_parms = index.indices(len(self))206            nr_assigned_items = (range_parms[1] - range_parms[0] - 1) // range_parms[2] + 1207            # need to test before changing, in case TypeError is caught208            if nr_assigned_items < len(value):209                raise TypeError(210                    f'too many elements in value {nr_assigned_items} < {len(value)}',211                )212            elif nr_assigned_items > len(value):213                raise TypeError(214                    f'not enough elements in value {nr_assigned_items} > {len(value)}',215                )216            for idx, i in enumerate(range(*range_parms)):217                self[i] = value[idx]218 219    def __delitem__(self, index: Any) -> None:220        if not isinstance(index, slice):221            return self.__delsingleitem__(index)222        # nprint(index.start, index.stop, index.step, index.indices(len(self)))223        for i in reversed(range(*index.indices(len(self)))):224            del self[i]225 226    @abstractmethod227    def __getsingleitem__(self, index: Any) -> Any:228        raise IndexError229 230    @abstractmethod231    def __setsingleitem__(self, index: Any, value: Any) -> None:232        raise IndexError233 234    @abstractmethod235    def __delsingleitem__(self, index: Any) -> None:236        raise IndexError237