|
-
- from __future__ import annotations
-
- # partially from package six by Benjamin Peterson
-
- import sys
- import os
- import io
- from abc import abstractmethod
- import collections.abc
-
-
- from ruamel.yaml.docinfo import Version # NOQA
- # fmt: off
- if False: # MYPY
- from typing import Any, Dict, Optional, List, Union, BinaryIO, IO, Text, Tuple # NOQA
- from typing import Optional # NOQA
- try:
- from typing import SupportsIndex as SupportsIndex # in order to reexport for mypy
- except ImportError:
- SupportsIndex = int # type: ignore
-
- StreamType = Any
- StreamTextType = StreamType
- VersionType = Union[str , Tuple[int, int] , List[int] , Version , None]
- # fmt: on
-
- _DEFAULT_YAML_VERSION = (1, 2)
-
- try:
- from collections import OrderedDict
- except ImportError:
- from ordereddict import OrderedDict # type: ignore
-
- # to get the right name import ... as ordereddict doesn't do that
-
-
- class ordereddict(OrderedDict): # type: ignore
- if not hasattr(OrderedDict, 'insert'):
-
- def insert(self, pos: int, key: Any, value: Any) -> None:
- if pos >= len(self):
- self[key] = value
- return
- od = ordereddict()
- od.update(self)
- for k in od:
- del self[k]
- for index, old_key in enumerate(od):
- if pos == index:
- self[key] = value
- self[old_key] = od[old_key]
-
-
- StringIO = io.StringIO
- BytesIO = io.BytesIO
-
-
- builtins_module = 'builtins'
-
-
- def with_metaclass(meta: Any, *bases: Any) -> Any:
- """Create a base class with a metaclass."""
- return meta('NewBase', bases, {})
-
-
- DBG_TOKEN = 1
- DBG_EVENT = 2
- DBG_NODE = 4
-
-
- _debug: Optional[int] = None
- if 'RUAMELDEBUG' in os.environ:
- _debugx = os.environ.get('RUAMELDEBUG')
- if _debugx is None:
- _debug = 0
- else:
- _debug = int(_debugx)
-
-
- if bool(_debug):
-
- class ObjectCounter:
- def __init__(self) -> None:
- self.map: Dict[Any, Any] = {}
-
- def __call__(self, k: Any) -> None:
- self.map[k] = self.map.get(k, 0) + 1
-
- def dump(self) -> None:
- for k in sorted(self.map):
- sys.stdout.write(f'{k} -> {self.map[k]}')
-
- object_counter = ObjectCounter()
-
-
- # used from yaml util when testing
- def dbg(val: Any = None) -> Any:
- debug = _debug
- if debug is None:
- # set to true or false
- _debugx = os.environ.get('YAMLDEBUG')
- if _debugx is None:
- debug = 0
- else:
- debug = int(_debugx)
- if val is None:
- return debug
- return debug & val
-
-
- class Nprint:
- def __init__(self, file_name: Any = None) -> None:
- self._max_print: Any = None
- self._count: Any = None
- self._file_name = file_name
-
- def __call__(self, *args: Any, **kw: Any) -> None:
- if not bool(_debug):
- return
- import traceback
-
- out = sys.stdout if self._file_name is None else open(self._file_name, 'a')
- dbgprint = print # to fool checking for print statements by dv utility
- kw1 = kw.copy()
- kw1['file'] = out
- dbgprint(*args, **kw1)
- out.flush()
- if self._max_print is not None:
- if self._count is None:
- self._count = self._max_print
- self._count -= 1
- if self._count == 0:
- dbgprint('forced exit\n')
- traceback.print_stack()
- out.flush()
- sys.exit(0)
- if self._file_name:
- out.close()
-
- def set_max_print(self, i: int) -> None:
- self._max_print = i
- self._count = None
-
- def fp(self, mode: str = 'a') -> Any:
- out = sys.stdout if self._file_name is None else open(self._file_name, mode)
- return out
-
-
- nprint = Nprint()
- nprintf = Nprint('/var/tmp/ruamel.yaml.log')
-
- # char checkers following production rules
-
-
- def check_namespace_char(ch: Any) -> bool:
- if '\x21' <= ch <= '\x7E': # ! to ~
- return True
- if '\xA0' <= ch <= '\uD7FF':
- return True
- if ('\uE000' <= ch <= '\uFFFD') and ch != '\uFEFF': # excl. byte order mark
- return True
- if '\U00010000' <= ch <= '\U0010FFFF':
- return True
- return False
-
-
- def check_anchorname_char(ch: Any) -> bool:
- if ch in ',[]{}':
- return False
- return check_namespace_char(ch)
-
-
- def version_tnf(t1: Any, t2: Any = None) -> Any:
- """
- return True if ruamel.yaml version_info < t1, None if t2 is specified and bigger else False
- """
- from ruamel.yaml import version_info # NOQA
-
- if version_info < t1:
- return True
- if t2 is not None and version_info < t2:
- return None
- return False
-
-
- class MutableSliceableSequence(collections.abc.MutableSequence): # type: ignore
- __slots__ = ()
-
- def __getitem__(self, index: Any) -> Any:
- if not isinstance(index, slice):
- return self.__getsingleitem__(index)
- return type(self)([self[i] for i in range(*index.indices(len(self)))]) # type: ignore
-
- def __setitem__(self, index: Any, value: Any) -> None:
- if not isinstance(index, slice):
- return self.__setsingleitem__(index, value)
- assert iter(value)
- # nprint(index.start, index.stop, index.step, index.indices(len(self)))
- if index.step is None:
- del self[index.start : index.stop]
- for elem in reversed(value):
- self.insert(0 if index.start is None else index.start, elem)
- else:
- range_parms = index.indices(len(self))
- nr_assigned_items = (range_parms[1] - range_parms[0] - 1) // range_parms[2] + 1
- # need to test before changing, in case TypeError is caught
- if nr_assigned_items < len(value):
- raise TypeError(
- f'too many elements in value {nr_assigned_items} < {len(value)}',
- )
- elif nr_assigned_items > len(value):
- raise TypeError(
- f'not enough elements in value {nr_assigned_items} > {len(value)}',
- )
- for idx, i in enumerate(range(*range_parms)):
- self[i] = value[idx]
-
- def __delitem__(self, index: Any) -> None:
- if not isinstance(index, slice):
- return self.__delsingleitem__(index)
- # nprint(index.start, index.stop, index.step, index.indices(len(self)))
- for i in reversed(range(*index.indices(len(self)))):
- del self[i]
-
- @abstractmethod
- def __getsingleitem__(self, index: Any) -> Any:
- raise IndexError
-
- @abstractmethod
- def __setsingleitem__(self, index: Any, value: Any) -> None:
- raise IndexError
-
- @abstractmethod
- def __delsingleitem__(self, index: Any) -> None:
- raise IndexError
|