codekingpro/portable-devtools
114k
1 2from __future__ import annotations3 4"""5stuff to deal with comments and formatting on dict/list/ordereddict/set6these are not really related, formatting could be factored out as7a separate base8"""9 10import sys11import copy12 13 14from ruamel.yaml.compat import ordereddict15from ruamel.yaml.compat import MutableSliceableSequence, nprintf # NOQA16from ruamel.yaml.scalarstring import ScalarString17from ruamel.yaml.anchor import Anchor18from ruamel.yaml.tag import Tag19 20from collections.abc import MutableSet, Sized, Set, Mapping21 22if False: # MYPY23 from typing import Any, Dict, Optional, List, Union, Iterator # NOQA24 25# fmt: off26__all__ = ['CommentedSeq', 'CommentedKeySeq',27 'CommentedMap', 'CommentedOrderedMap',28 'CommentedSet', 'comment_attrib', 'merge_attrib',29 'TaggedScalar',30 'C_POST', 'C_PRE', 'C_SPLIT_ON_FIRST_BLANK', 'C_BLANK_LINE_PRESERVE_SPACE',31 ]32# fmt: on33 34# splitting of comments by the scanner35# an EOLC (End-Of-Line Comment) is preceded by some token36# an FLC (Full Line Comment) is a comment not preceded by a token, i.e. # is37# the first non-blank on line38# a BL is a blank line i.e. empty or spaces/tabs only39# bits 0 and 1 are combined, you can choose only one40C_POST = 0b0041C_PRE = 0b0142C_SPLIT_ON_FIRST_BLANK = 0b10 # as C_POST, but if blank line then C_PRE all lines before43# first blank goes to POST even if no following real FLC44# (first blank -> first of post)45# 0b11 -> reserved for future use46C_BLANK_LINE_PRESERVE_SPACE = 0b10047# C_EOL_PRESERVE_SPACE2 = 0b100048 49 50class IDX:51 # temporary auto increment, so rearranging is easier52 def __init__(self) -> None:53 self._idx = 054 55 def __call__(self) -> Any:56 x = self._idx57 self._idx += 158 return x59 60 def __str__(self) -> Any:61 return str(self._idx)62 63 64cidx = IDX()65 66# more or less in order of subjective expected likelyhood67# the _POST and _PRE ones are lists themselves68C_VALUE_EOL = C_ELEM_EOL = cidx()69C_KEY_EOL = cidx()70C_KEY_PRE = C_ELEM_PRE = cidx() # not this is not value71C_VALUE_POST = C_ELEM_POST = cidx() # not this is not value72C_VALUE_PRE = cidx()73C_KEY_POST = cidx()74C_TAG_EOL = cidx()75C_TAG_POST = cidx()76C_TAG_PRE = cidx()77C_ANCHOR_EOL = cidx()78C_ANCHOR_POST = cidx()79C_ANCHOR_PRE = cidx()80 81 82comment_attrib = '_yaml_comment'83format_attrib = '_yaml_format'84line_col_attrib = '_yaml_line_col'85merge_attrib = '_yaml_merge'86 87 88class Comment:89 # using sys.getsize tested the Comment objects, __slots__ makes them bigger90 # and adding self.end did not matter91 __slots__ = 'comment', '_items', '_post', '_pre'92 attrib = comment_attrib93 94 def __init__(self, old: bool = True) -> None:95 self._pre = None if old else [] # type: ignore96 self.comment = None # [post, [pre]]97 # map key (mapping/omap/dict) or index (sequence/list) to a list of98 # dict: post_key, pre_key, post_value, pre_value99 # list: pre item, post item100 self._items: Dict[Any, Any] = {}101 # self._start = [] # should not put these on first item102 self._post: List[Any] = [] # end of document comments103 104 def __str__(self) -> str:105 if bool(self._post):106 end = ',\n end=' + str(self._post)107 else:108 end = ""109 return f'Comment(comment={self.comment},\n items={self._items}{end})'110 111 def _old__repr__(self) -> str:112 if bool(self._post):113 end = ',\n end=' + str(self._post)114 else:115 end = ""116 try:117 ln = max([len(str(k)) for k in self._items]) + 1118 except ValueError:119 ln = '' # type: ignore120 it = ' '.join([f'{str(k) + ":":{ln}} {v}\n' for k, v in self._items.items()])121 if it:122 it = '\n ' + it + ' '123 return f'Comment(\n start={self.comment},\n items={{{it}}}{end})'124 125 def __repr__(self) -> str:126 if self._pre is None:127 return self._old__repr__()128 if bool(self._post):129 end = ',\n end=' + repr(self._post)130 else:131 end = ""132 try:133 ln = max([len(str(k)) for k in self._items]) + 1134 except ValueError:135 ln = '' # type: ignore136 it = ' '.join([f'{str(k) + ":":{ln}} {v}\n' for k, v in self._items.items()])137 if it:138 it = '\n ' + it + ' '139 return f'Comment(\n pre={self.pre},\n items={{{it}}}{end})'140 141 @property142 def items(self) -> Any:143 return self._items144 145 @property146 def end(self) -> Any:147 return self._post148 149 @end.setter150 def end(self, value: Any) -> None:151 self._post = value152 153 @property154 def pre(self) -> Any:155 return self._pre156 157 @pre.setter158 def pre(self, value: Any) -> None:159 self._pre = value160 161 def get(self, item: Any, pos: Any) -> Any:162 x = self._items.get(item)163 if x is None or len(x) < pos:164 return None165 return x[pos] # can be None166 167 def set(self, item: Any, pos: Any, value: Any) -> Any:168 x = self._items.get(item)169 if x is None:170 self._items[item] = x = [None] * (pos + 1)171 else:172 while len(x) <= pos:173 x.append(None)174 assert x[pos] is None175 x[pos] = value176 177 def __contains__(self, x: Any) -> Any:178 # test if a substring is in any of the attached comments179 if self.comment:180 if self.comment[0] and x in self.comment[0].value:181 return True182 if self.comment[1]:183 for c in self.comment[1]:184 if x in c.value:185 return True186 for value in self.items.values():187 if not value:188 continue189 for c in value:190 if c and x in c.value:191 return True192 if self.end:193 for c in self.end:194 if x in c.value:195 return True196 return False197 198 199# to distinguish key from None200class NotNone:201 pass # NOQA202 203 204class Format:205 __slots__ = ('_flow_style',)206 attrib = format_attrib207 208 def __init__(self) -> None:209 self._flow_style: Any = None210 211 def set_flow_style(self) -> None:212 self._flow_style = True213 214 def set_block_style(self) -> None:215 self._flow_style = False216 217 def flow_style(self, default: Optional[Any] = None) -> Any:218 """if default (the flow_style) is None, the flow style tacked on to219 the object explicitly will be taken. If that is None as well the220 default flow style rules the format down the line, or the type221 of the constituent values (simple -> flow, map/list -> block)"""222 if self._flow_style is None:223 return default224 return self._flow_style225 226 def __repr__(self) -> str:227 return f'Format({self._flow_style})'228 229 230class LineCol:231 """232 line and column information wrt document, values start at zero (0)233 """234 235 attrib = line_col_attrib236 237 def __init__(self) -> None:238 self.line = None239 self.col = None240 self.data: Optional[Dict[Any, Any]] = None241 242 def add_kv_line_col(self, key: Any, data: Any) -> None:243 if self.data is None:244 self.data = {}245 self.data[key] = data246 247 def key(self, k: Any) -> Any:248 return self._kv(k, 0, 1)249 250 def value(self, k: Any) -> Any:251 return self._kv(k, 2, 3)252 253 def _kv(self, k: Any, x0: Any, x1: Any) -> Any:254 if self.data is None:255 return None256 data = self.data[k]257 return data[x0], data[x1]258 259 def item(self, idx: Any) -> Any:260 if self.data is None:261 return None262 return self.data[idx][0], self.data[idx][1]263 264 def add_idx_line_col(self, key: Any, data: Any) -> None:265 if self.data is None:266 self.data = {}267 self.data[key] = data268 269 def __repr__(self) -> str:270 return f'LineCol({self.line}, {self.col})'271 272 273class CommentedBase:274 @property275 def ca(self):276 # type: () -> Any277 if not hasattr(self, Comment.attrib):278 setattr(self, Comment.attrib, Comment())279 return getattr(self, Comment.attrib)280 281 def yaml_end_comment_extend(self, comment: Any, clear: bool = False) -> None:282 if comment is None:283 return284 if clear or self.ca.end is None:285 self.ca.end = []286 self.ca.end.extend(comment)287 288 def yaml_key_comment_extend(self, key: Any, comment: Any, clear: bool = False) -> None:289 r = self.ca._items.setdefault(key, [None, None, None, None])290 if clear or r[1] is None:291 if comment[1] is not None:292 assert isinstance(comment[1], list)293 r[1] = comment[1]294 else:295 r[1].extend(comment[0])296 r[0] = comment[0]297 298 def yaml_value_comment_extend(self, key: Any, comment: Any, clear: bool = False) -> None:299 r = self.ca._items.setdefault(key, [None, None, None, None])300 if clear or r[3] is None:301 if comment[1] is not None:302 assert isinstance(comment[1], list)303 r[3] = comment[1]304 else:305 r[3].extend(comment[0])306 r[2] = comment[0]307 308 def yaml_set_start_comment(self, comment: Any, indent: Any = 0) -> None:309 """overwrites any preceding comment lines on an object310 expects comment to be without `#` and possible have multiple lines311 """312 from .error import CommentMark313 from .tokens import CommentToken314 315 pre_comments = self._yaml_clear_pre_comment() # type: ignore316 if comment[-1] == '\n':317 comment = comment[:-1] # strip final newline if there318 start_mark = CommentMark(indent)319 for com in comment.split('\n'):320 c = com.strip()321 if len(c) > 0 and c[0] != '#':322 com = '# ' + com323 pre_comments.append(CommentToken(com + '\n', start_mark))324 325 def yaml_set_comment_before_after_key(326 self,327 key: Any,328 before: Any = None,329 indent: Any = 0,330 after: Any = None,331 after_indent: Any = None,332 ) -> None:333 """334 expects comment (before/after) to be without `#` and possible have multiple lines335 """336 from ruamel.yaml.error import CommentMark337 from ruamel.yaml.tokens import CommentToken338 339 def comment_token(s: Any, mark: Any) -> Any:340 # handle empty lines as having no comment341 return CommentToken(('# ' if s else "") + s + '\n', mark)342 343 if after_indent is None:344 after_indent = indent + 2345 if before and (len(before) > 1) and before[-1] == '\n':346 before = before[:-1] # strip final newline if there347 if after and after[-1] == '\n':348 after = after[:-1] # strip final newline if there349 start_mark = CommentMark(indent)350 c = self.ca.items.setdefault(key, [None, [], None, None])351 if before is not None:352 if c[1] is None:353 c[1] = []354 if before == '\n':355 c[1].append(comment_token("", start_mark)) # type: ignore356 else:357 for com in before.split('\n'):358 c[1].append(comment_token(com, start_mark)) # type: ignore359 if after:360 start_mark = CommentMark(after_indent)361 if c[3] is None:362 c[3] = []363 for com in after.split('\n'):364 c[3].append(comment_token(com, start_mark)) # type: ignore365 366 @property367 def fa(self) -> Any:368 """format attribute369 370 set_flow_style()/set_block_style()"""371 if not hasattr(self, Format.attrib):372 setattr(self, Format.attrib, Format())373 return getattr(self, Format.attrib)374 375 def yaml_add_eol_comment(376 self, comment: Any, key: Optional[Any] = NotNone, column: Optional[Any] = None,377 ) -> None:378 """379 there is a problem as eol comments should start with ' #'380 (but at the beginning of the line the space doesn't have to be before381 the #. The column index is for the # mark382 """383 from .tokens import CommentToken384 from .error import CommentMark385 386 if column is None:387 try:388 column = self._yaml_get_column(key)389 except AttributeError:390 column = 0391 if comment[0] != '#':392 comment = '# ' + comment393 if column is None:394 if comment[0] == '#':395 comment = ' ' + comment396 column = 0397 start_mark = CommentMark(column)398 ct = [CommentToken(comment, start_mark), None]399 self._yaml_add_eol_comment(ct, key=key)400 401 @property402 def lc(self) -> Any:403 if not hasattr(self, LineCol.attrib):404 setattr(self, LineCol.attrib, LineCol())405 return getattr(self, LineCol.attrib)406 407 def _yaml_set_line_col(self, line: Any, col: Any) -> None:408 self.lc.line = line409 self.lc.col = col410 411 def _yaml_set_kv_line_col(self, key: Any, data: Any) -> None:412 self.lc.add_kv_line_col(key, data)413 414 def _yaml_set_idx_line_col(self, key: Any, data: Any) -> None:415 self.lc.add_idx_line_col(key, data)416 417 @property418 def anchor(self) -> Any:419 if not hasattr(self, Anchor.attrib):420 setattr(self, Anchor.attrib, Anchor())421 return getattr(self, Anchor.attrib)422 423 def yaml_anchor(self) -> Any:424 if not hasattr(self, Anchor.attrib):425 return None426 return self.anchor427 428 def yaml_set_anchor(self, value: Any, always_dump: bool = False) -> None:429 self.anchor.value = value430 self.anchor.always_dump = always_dump431 432 @property433 def tag(self) -> Any:434 if not hasattr(self, Tag.attrib):435 setattr(self, Tag.attrib, Tag())436 return getattr(self, Tag.attrib)437 438 def yaml_set_ctag(self, value: Tag) -> None:439 setattr(self, Tag.attrib, value)440 441 def copy_attributes(self, t: Any, memo: Any = None) -> Any:442 """443 copies the YAML related attributes, not e.g. .values444 returns target445 """446 # fmt: off447 for a in [Comment.attrib, Format.attrib, LineCol.attrib, Anchor.attrib,448 Tag.attrib, merge_attrib]:449 if hasattr(self, a):450 if memo is not None:451 setattr(t, a, copy.deepcopy(getattr(self, a, memo)))452 else:453 setattr(t, a, getattr(self, a))454 return t455 # fmt: on456 457 def _yaml_add_eol_comment(self, comment: Any, key: Any) -> None:458 raise NotImplementedError459 460 def _yaml_get_pre_comment(self) -> Any:461 raise NotImplementedError462 463 def _yaml_get_column(self, key: Any) -> Any:464 raise NotImplementedError465 466 467class CommentedSeq(MutableSliceableSequence, list, CommentedBase): # type: ignore468 __slots__ = (Comment.attrib, '_lst')469 470 def __init__(self, *args: Any, **kw: Any) -> None:471 list.__init__(self, *args, **kw)472 473 def __getsingleitem__(self, idx: Any) -> Any:474 return list.__getitem__(self, idx)475 476 def __setsingleitem__(self, idx: Any, value: Any) -> None:477 # try to preserve the scalarstring type if setting an existing key to a new value478 if idx < len(self):479 if (480 isinstance(value, str)481 and not isinstance(value, ScalarString)482 and isinstance(self[idx], ScalarString)483 ):484 value = type(self[idx])(value)485 list.__setitem__(self, idx, value)486 487 def __delsingleitem__(self, idx: Any = None) -> Any:488 list.__delitem__(self, idx)489 self.ca.items.pop(idx, None) # might not be there -> default value490 for list_index in sorted(self.ca.items):491 if list_index < idx:492 continue493 self.ca.items[list_index - 1] = self.ca.items.pop(list_index)494 495 def __len__(self) -> int:496 return list.__len__(self)497 498 def insert(self, idx: Any, val: Any) -> None:499 """the comments after the insertion have to move forward"""500 list.insert(self, idx, val)501 for list_index in sorted(self.ca.items, reverse=True):502 if list_index < idx:503 break504 self.ca.items[list_index + 1] = self.ca.items.pop(list_index)505 506 def extend(self, val: Any) -> None:507 list.extend(self, val)508 509 def __eq__(self, other: Any) -> bool:510 return list.__eq__(self, other)511 512 def _yaml_add_comment(self, comment: Any, key: Optional[Any] = NotNone) -> None:513 if key is not NotNone:514 self.yaml_key_comment_extend(key, comment)515 else:516 self.ca.comment = comment517 518 def _yaml_add_eol_comment(self, comment: Any, key: Any) -> None:519 self._yaml_add_comment(comment, key=key)520 521 def _yaml_get_columnX(self, key: Any) -> Any:522 return self.ca.items[key][0].start_mark.column523 524 def _yaml_get_column(self, key: Any) -> Any:525 column = None526 sel_idx = None527 pre, post = key - 1, key + 1528 if pre in self.ca.items:529 sel_idx = pre530 elif post in self.ca.items:531 sel_idx = post532 else:533 # self.ca.items is not ordered534 for row_idx, _k1 in enumerate(self):535 if row_idx >= key:536 break537 if row_idx not in self.ca.items:538 continue539 sel_idx = row_idx540 if sel_idx is not None:541 column = self._yaml_get_columnX(sel_idx)542 return column543 544 def _yaml_get_pre_comment(self) -> Any:545 pre_comments: List[Any] = []546 if self.ca.comment is None:547 self.ca.comment = [None, pre_comments]548 else:549 pre_comments = self.ca.comment[1]550 return pre_comments551 552 def _yaml_clear_pre_comment(self) -> Any:553 pre_comments: List[Any] = []554 if self.ca.comment is None:555 self.ca.comment = [None, pre_comments]556 else:557 self.ca.comment[1] = pre_comments558 return pre_comments559 560 def __deepcopy__(self, memo: Any) -> Any:561 res = self.__class__()562 memo[id(self)] = res563 for k in self:564 res.append(copy.deepcopy(k, memo))565 self.copy_attributes(res, memo=memo)566 return res567 568 def __add__(self, other: Any) -> Any:569 return list.__add__(self, other)570 571 def sort(self, key: Any = None, reverse: bool = False) -> None:572 if key is None:573 tmp_lst = sorted(zip(self, range(len(self))), reverse=reverse)574 list.__init__(self, [x[0] for x in tmp_lst])575 else:576 tmp_lst = sorted(577 zip(map(key, list.__iter__(self)), range(len(self))), reverse=reverse,578 )579 list.__init__(self, [list.__getitem__(self, x[1]) for x in tmp_lst])580 itm = self.ca.items581 self.ca._items = {}582 for idx, x in enumerate(tmp_lst):583 old_index = x[1]584 if old_index in itm:585 self.ca.items[idx] = itm[old_index]586 587 def __repr__(self) -> Any:588 return list.__repr__(self)589 590 591class CommentedKeySeq(tuple, CommentedBase): # type: ignore592 """This primarily exists to be able to roundtrip keys that are sequences"""593 594 def _yaml_add_comment(self, comment: Any, key: Optional[Any] = NotNone) -> None:595 if key is not NotNone:596 self.yaml_key_comment_extend(key, comment)597 else:598 self.ca.comment = comment599 600 def _yaml_add_eol_comment(self, comment: Any, key: Any) -> None:601 self._yaml_add_comment(comment, key=key)602 603 def _yaml_get_columnX(self, key: Any) -> Any:604 return self.ca.items[key][0].start_mark.column605 606 def _yaml_get_column(self, key: Any) -> Any:607 column = None608 sel_idx = None609 pre, post = key - 1, key + 1610 if pre in self.ca.items:611 sel_idx = pre612 elif post in self.ca.items:613 sel_idx = post614 else:615 # self.ca.items is not ordered616 for row_idx, _k1 in enumerate(self):617 if row_idx >= key:618 break619 if row_idx not in self.ca.items:620 continue621 sel_idx = row_idx622 if sel_idx is not None:623 column = self._yaml_get_columnX(sel_idx)624 return column625 626 def _yaml_get_pre_comment(self) -> Any:627 pre_comments: List[Any] = []628 if self.ca.comment is None:629 self.ca.comment = [None, pre_comments]630 else:631 pre_comments = self.ca.comment[1]632 return pre_comments633 634 def _yaml_clear_pre_comment(self) -> Any:635 pre_comments: List[Any] = []636 if self.ca.comment is None:637 self.ca.comment = [None, pre_comments]638 else:639 self.ca.comment[1] = pre_comments640 return pre_comments641 642 643class CommentedMapView(Sized):644 __slots__ = ('_mapping',)645 646 def __init__(self, mapping: Any) -> None:647 self._mapping = mapping648 649 def __len__(self) -> int:650 count = len(self._mapping)651 return count652 653 654class CommentedMapKeysView(CommentedMapView, Set): # type: ignore655 __slots__ = ()656 657 @classmethod658 def _from_iterable(self, it: Any) -> Any:659 return set(it)660 661 def __contains__(self, key: Any) -> Any:662 return key in self._mapping663 664 def __iter__(self) -> Any:665 # yield from self._mapping # not in py27, pypy666 # for x in self._mapping._keys():667 for x in self._mapping:668 yield x669 670 671class CommentedMapItemsView(CommentedMapView, Set): # type: ignore672 __slots__ = ()673 674 @classmethod675 def _from_iterable(self, it: Any) -> Any:676 return set(it)677 678 def __contains__(self, item: Any) -> Any:679 key, value = item680 try:681 v = self._mapping[key]682 except KeyError:683 return False684 else:685 return v == value686 687 def __iter__(self) -> Any:688 for key in self._mapping._keys():689 yield (key, self._mapping[key])690 691 692class CommentedMapValuesView(CommentedMapView):693 __slots__ = ()694 695 def __contains__(self, value: Any) -> Any:696 for key in self._mapping:697 if value == self._mapping[key]:698 return True699 return False700 701 def __iter__(self) -> Any:702 for key in self._mapping._keys():703 yield self._mapping[key]704 705 706class CommentedMap(ordereddict, CommentedBase):707 __slots__ = (Comment.attrib, '_ok', '_ref')708 709 def __init__(self, *args: Any, **kw: Any) -> None:710 self._ok: MutableSet[Any] = set() # own keys711 self._ref: List[CommentedMap] = []712 ordereddict.__init__(self, *args, **kw)713 714 def _yaml_add_comment(715 self, comment: Any, key: Optional[Any] = NotNone, value: Optional[Any] = NotNone,716 ) -> None:717 """values is set to key to indicate a value attachment of comment"""718 if key is not NotNone:719 self.yaml_key_comment_extend(key, comment)720 return721 if value is not NotNone:722 self.yaml_value_comment_extend(value, comment)723 else:724 self.ca.comment = comment725 726 def _yaml_add_eol_comment(self, comment: Any, key: Any) -> None:727 """add on the value line, with value specified by the key"""728 self._yaml_add_comment(comment, value=key)729 730 def _yaml_get_columnX(self, key: Any) -> Any:731 return self.ca.items[key][2].start_mark.column732 733 def _yaml_get_column(self, key: Any) -> Any:734 column = None735 sel_idx = None736 pre, post, last = None, None, None737 for x in self:738 if pre is not None and x != key:739 post = x740 break741 if x == key:742 pre = last743 last = x744 if pre in self.ca.items:745 sel_idx = pre746 elif post in self.ca.items:747 sel_idx = post748 else:749 # self.ca.items is not ordered750 for k1 in self:751 if k1 >= key:752 break753 if k1 not in self.ca.items:754 continue755 sel_idx = k1756 if sel_idx is not None:757 column = self._yaml_get_columnX(sel_idx)758 return column759 760 def _yaml_get_pre_comment(self) -> Any:761 pre_comments: List[Any] = []762 if self.ca.comment is None:763 self.ca.comment = [None, pre_comments]764 else:765 pre_comments = self.ca.comment[1]766 return pre_comments767 768 def _yaml_clear_pre_comment(self) -> Any:769 pre_comments: List[Any] = []770 if self.ca.comment is None:771 self.ca.comment = [None, pre_comments]772 else:773 self.ca.comment[1] = pre_comments774 return pre_comments775 776 def update(self, *vals: Any, **kw: Any) -> None:777 try:778 ordereddict.update(self, *vals, **kw)779 except TypeError:780 # probably a dict that is used781 for x in vals[0]:782 self[x] = vals[0][x]783 if vals:784 try:785 self._ok.update(vals[0].keys()) # type: ignore786 except AttributeError:787 # assume one argument that is a list/tuple of two element lists/tuples788 for x in vals[0]:789 self._ok.add(x[0])790 if kw:791 self._ok.update(*kw.keys()) # type: ignore792 793 def insert(self, pos: Any, key: Any, value: Any, comment: Optional[Any] = None) -> None:794 """insert key value into given position, as defined by source YAML795 attach comment if provided796 """797 if key in self._ok:798 del self[key]799 keys = [k for k in self.keys() if k in self._ok]800 try:801 merge_value = getattr(self, merge_attrib)802 merge_pos = merge_value.merge_pos803 except (AttributeError, IndexError):804 merge_pos = -1805 if merge_pos >= 0:806 if merge_pos >= pos:807 # getattr(self, merge_attrib)[0] = (merge_pos + 1, ma0[1])808 merge_value.merge_pos += 1809 idx_min = pos810 idx_max = len(self._ok)811 else:812 idx_min = pos - 1813 idx_max = len(self._ok)814 else:815 idx_min = pos816 idx_max = len(self._ok)817 self[key] = value # at the end818 # print(f'{idx_min=} {idx_max=}')819 for idx in range(idx_min, idx_max):820 self.move_to_end(keys[idx])821 self._ok.add(key)822 # for referer in self._ref:823 # for keytmp in keys:824 # referer.update_key_value(keytmp)825 if comment is not None:826 self.yaml_add_eol_comment(comment, key=key)827 828 def mlget(self, key: Any, default: Any = None, list_ok: Any = False) -> Any:829 """multi-level get that expects dicts within dicts"""830 if not isinstance(key, list):831 return self.get(key, default)832 # assume that the key is a list of recursively accessible dicts833 834 def get_one_level(key_list: Any, level: Any, d: Any) -> Any:835 if not list_ok:836 assert isinstance(d, dict)837 if level >= len(key_list):838 if level > len(key_list):839 raise IndexError840 return d[key_list[level - 1]]841 return get_one_level(key_list, level + 1, d[key_list[level - 1]])842 843 try:844 return get_one_level(key, 1, self)845 except KeyError:846 return default847 except (TypeError, IndexError):848 if not list_ok:849 raise850 return default851 852 def __getitem__(self, key: Any) -> Any:853 try:854 return ordereddict.__getitem__(self, key)855 except KeyError:856 for merged in getattr(self, merge_attrib, []):857 # if isinstance(merged, tuple):858 # if key in merged[1]:859 # return merged[1][key]860 # else:861 if True:862 if key in merged:863 return merged[key]864 raise865 866 def __setitem__(self, key: Any, value: Any) -> None:867 # try to preserve the scalarstring type if setting an existing key to a new value868 if key in self:869 if (870 isinstance(value, str)871 and not isinstance(value, ScalarString)872 and isinstance(self[key], ScalarString)873 ):874 value = type(self[key])(value)875 ordereddict.__setitem__(self, key, value)876 self._ok.add(key)877 878 def _unmerged_contains(self, key: Any) -> Any:879 if key in self._ok:880 return True881 return None882 883 def __contains__(self, key: Any) -> bool:884 return bool(ordereddict.__contains__(self, key))885 886 def get(self, key: Any, default: Any = None) -> Any:887 try:888 return self.__getitem__(key)889 except: # NOQA890 return default891 892 def __repr__(self) -> Any:893 res = '{'894 sep = ''895 for k, v in self.items():896 res += f'{sep}{k!r}: {v!r}'897 if not sep:898 sep = ', '899 res += '}'900 return res901 902 def non_merged_items(self) -> Any:903 for x in ordereddict.__iter__(self):904 if x in self._ok:905 yield x, ordereddict.__getitem__(self, x)906 907 def __delitem__(self, key: Any) -> None:908 # for merged in getattr(self, merge_attrib, []):909 # if key in merged[1]:910 # value = merged[1][key]911 # break912 # else:913 # # not found in merged in stuff914 # ordereddict.__delitem__(self, key)915 # for referer in self._ref:916 # referer.update=_key_value(key)917 # return918 #919 # ordereddict.__setitem__(self, key, value) # merge might have different value920 # self._ok.discard(key)921 922 try:923 merge_value = getattr(self, merge_attrib)924 merge_pos = merge_value.merge_pos925 except AttributeError:926 merge_pos = -1927 if merge_pos >= 0:928 try:929 pos = list(ordereddict.keys(self)).index(key)930 # the merge is not in the dict, so don't use >=931 if merge_pos > pos:932 merge_value.merge_pos -= 1933 except ValueError:934 pass # let the removal of the key throw a "normal" error935 self._ok.discard(key)936 ordereddict.__delitem__(self, key)937 for referer in self._ref:938 referer.update_key_value(key)939 940 def __iter__(self) -> Any:941 for x in ordereddict.__iter__(self):942 yield x943 944 def pop(self, key: Any, default: Any = NotNone) -> Any:945 try:946 result = self[key]947 except KeyError:948 if default is NotNone:949 raise950 return default951 del self[key]952 return result953 954 def _keys(self) -> Any:955 for x in ordereddict.__iter__(self):956 yield x957 958 def __len__(self) -> int:959 return int(ordereddict.__len__(self))960 961 def __eq__(self, other: Any) -> bool:962 return bool(dict(self) == other)963 964 def keys(self) -> Any:965 return CommentedMapKeysView(self)966 967 def values(self) -> Any:968 return CommentedMapValuesView(self)969 970 def _items(self) -> Any:971 for x in ordereddict.__iter__(self):972 yield x, ordereddict.__getitem__(self, x)973 974 def items(self) -> Any:975 return CommentedMapItemsView(self)976 977 @property978 def merge(self) -> Any:979 if not hasattr(self, merge_attrib):980 setattr(self, merge_attrib, [])981 return getattr(self, merge_attrib)982 983 def copy(self) -> Any:984 x = type(self)() # update doesn't work985 for k, v in self._items():986 x[k] = v987 self.copy_attributes(x)988 return x989 990 def add_referent(self, cm: Any) -> None:991 if cm not in self._ref:992 self._ref.append(cm)993 994 def add_yaml_merge(self, value: Any) -> None:995 assert not hasattr(self, merge_attrib)996 setattr(self, merge_attrib, value)997 for v in value:998 # if isinstance(v, tuple):999 # assert len(v) == 21000 # # print('vvv', v, type(v[1]))1001 # v[1].add_referent(self)1002 # for k1, v1 in v[1].items():1003 # if ordereddict.__contains__(self, k1):1004 # continue1005 # ordereddict.__setitem__(self, k1, v1)1006 # else:1007 if True:1008 v.add_referent(self)1009 for k1, v1 in v.items():1010 if ordereddict.__contains__(self, k1):1011 continue1012 ordereddict.__setitem__(self, k1, v1)1013 1014 def update_key_value(self, key: Any) -> None:1015 if key in self._ok:1016 return1017 for v in self.merge:1018 if key in v[1]:1019 ordereddict.__setitem__(self, key, v[1][key])1020 return1021 ordereddict.__delitem__(self, key)1022 1023 def __deepcopy__(self, memo: Any) -> Any:1024 res = self.__class__()1025 memo[id(self)] = res1026 for k in self:1027 res[k] = copy.deepcopy(self[k], memo)1028 self.copy_attributes(res, memo=memo)1029 return res1030 1031 1032# based on brownie mappings1033@classmethod # type: ignore1034def raise_immutable(cls: Any, *args: Any, **kwargs: Any) -> None:1035 raise TypeError(f'{cls.__name__} objects are immutable')1036 1037 1038class CommentedKeyMap(CommentedBase, Mapping): # type: ignore1039 __slots__ = Comment.attrib, '_od'1040 """This primarily exists to be able to roundtrip keys that are mappings"""1041 1042 def __init__(self, *args: Any, **kw: Any) -> None:1043 if hasattr(self, '_od'):1044 raise_immutable(self)1045 try:1046 self._od = ordereddict(*args, **kw)1047 except TypeError:1048 raise1049 1050 __delitem__ = __setitem__ = clear = pop = popitem = setdefault = update = raise_immutable1051 1052 # need to implement __getitem__, __iter__ and __len__1053 def __getitem__(self, index: Any) -> Any:1054 return self._od[index]1055 1056 def __iter__(self) -> Iterator[Any]:1057 for x in self._od.__iter__():1058 yield x1059 1060 def __len__(self) -> int:1061 return len(self._od)1062 1063 def __hash__(self) -> Any:1064 return hash(tuple(self.items()))1065 1066 def __repr__(self) -> Any:1067 if not hasattr(self, merge_attrib):1068 return self._od.__repr__()1069 return 'ordereddict(' + repr(list(self._od.items())) + ')'1070 1071 @classmethod1072 def fromkeys(keys: Any, v: Any = None) -> Any:1073 return CommentedKeyMap(dict.fromkeys(keys, v))1074 1075 def _yaml_add_comment(self, comment: Any, key: Optional[Any] = NotNone) -> None:1076 if key is not NotNone:1077 self.yaml_key_comment_extend(key, comment)1078 else:1079 self.ca.comment = comment1080 1081 def _yaml_add_eol_comment(self, comment: Any, key: Any) -> None:1082 self._yaml_add_comment(comment, key=key)1083 1084 def _yaml_get_columnX(self, key: Any) -> Any:1085 return self.ca.items[key][0].start_mark.column1086 1087 def _yaml_get_column(self, key: Any) -> Any:1088 column = None1089 sel_idx = None1090 pre, post = key - 1, key + 11091 if pre in self.ca.items:1092 sel_idx = pre1093 elif post in self.ca.items:1094 sel_idx = post1095 else:1096 # self.ca.items is not ordered1097 for row_idx, _k1 in enumerate(self):1098 if row_idx >= key:1099 break1100 if row_idx not in self.ca.items:1101 continue1102 sel_idx = row_idx1103 if sel_idx is not None:1104 column = self._yaml_get_columnX(sel_idx)1105 return column1106 1107 def _yaml_get_pre_comment(self) -> Any:1108 pre_comments: List[Any] = []1109 if self.ca.comment is None:1110 self.ca.comment = [None, pre_comments]1111 else:1112 self.ca.comment[1] = pre_comments1113 return pre_comments1114 1115 1116class CommentedOrderedMap(CommentedMap):1117 __slots__ = (Comment.attrib,)1118 1119 1120class CommentedSet(MutableSet, CommentedBase): # type: ignore # NOQA1121 __slots__ = Comment.attrib, 'odict'1122 1123 def __init__(self, values: Any = None) -> None:1124 self.odict = ordereddict()1125 MutableSet.__init__(self)1126 if values is not None:1127 self |= values1128 1129 def _yaml_add_comment(1130 self, comment: Any, key: Optional[Any] = NotNone, value: Optional[Any] = NotNone,1131 ) -> None:1132 """values is set to key to indicate a value attachment of comment"""1133 if key is not NotNone:1134 self.yaml_key_comment_extend(key, comment)1135 return1136 if value is not NotNone:1137 self.yaml_value_comment_extend(value, comment)1138 else:1139 self.ca.comment = comment1140 1141 def _yaml_add_eol_comment(self, comment: Any, key: Any) -> None:1142 """add on the value line, with value specified by the key"""1143 self._yaml_add_comment(comment, value=key)1144 1145 def add(self, value: Any) -> None:1146 """Add an element."""1147 self.odict[value] = None1148 1149 def discard(self, value: Any) -> None:1150 """Remove an element. Do not raise an exception if absent."""1151 del self.odict[value]1152 1153 def __contains__(self, x: Any) -> Any:1154 return x in self.odict1155 1156 def __iter__(self) -> Any:1157 for x in self.odict:1158 yield x1159 1160 def __len__(self) -> int:1161 return len(self.odict)1162 1163 def __repr__(self) -> str:1164 return f'set({self.odict.keys()!r})'1165 1166 1167class TaggedScalar(CommentedBase):1168 # the value and style attributes are set during roundtrip construction1169 def __init__(self, value: Any = None, style: Any = None, tag: Any = None) -> None:1170 self.value = value1171 self.style = style1172 if tag is not None:1173 if isinstance(tag, str):1174 tag = Tag(suffix=tag)1175 self.yaml_set_ctag(tag)1176 1177 def __str__(self) -> Any:1178 return self.value1179 1180 def count(self, s: str, start: Optional[int] = None, end: Optional[int] = None) -> Any:1181 return self.value.count(s, start, end)1182 1183 def __getitem__(self, pos: int) -> Any:1184 return self.value[pos]1185 1186 def __repr__(self) -> str:1187 return f'TaggedScalar(value={self.value!r}, style={self.style!r}, tag={self.tag!r})'1188 1189 1190def dump_comments(d: Any, name: str = "", sep: str = '.', out: Any = sys.stdout) -> None:1191 """1192 recursively dump comments, all but the toplevel preceded by the path1193 in dotted form x.0.a1194 """1195 if isinstance(d, dict) and hasattr(d, 'ca'):1196 if name:1197 out.write(f'{name} {type(d)}\n')1198 out.write(f'{d.ca!r}\n')1199 for k in d:1200 dump_comments(d[k], name=(name + sep + str(k)) if name else k, sep=sep, out=out)