600955edd4
* Vendor ruamel.yaml v0.17.21 * Add unit test for whitespace regression * Add an abstraction layer in Spack to wrap ruamel.yaml All YAML operations are routed through spack.util.spack_yaml The custom classes have been adapted to the new ruamel.yaml class hierarchy. Fixed line annotation issue in "spack config blame"
257 lines
8.1 KiB
Python
257 lines
8.1 KiB
Python
# coding: utf-8
|
|
|
|
"""
|
|
some helper functions that might be generally useful
|
|
"""
|
|
|
|
import datetime
|
|
from functools import partial
|
|
import re
|
|
|
|
|
|
if False: # MYPY
|
|
from typing import Any, Dict, Optional, List, Text # NOQA
|
|
from .compat import StreamTextType # NOQA
|
|
|
|
|
|
class LazyEval:
|
|
"""
|
|
Lightweight wrapper around lazily evaluated func(*args, **kwargs).
|
|
|
|
func is only evaluated when any attribute of its return value is accessed.
|
|
Every attribute access is passed through to the wrapped value.
|
|
(This only excludes special cases like method-wrappers, e.g., __hash__.)
|
|
The sole additional attribute is the lazy_self function which holds the
|
|
return value (or, prior to evaluation, func and arguments), in its closure.
|
|
"""
|
|
|
|
def __init__(self, func, *args, **kwargs):
|
|
# type: (Any, Any, Any) -> None
|
|
def lazy_self():
|
|
# type: () -> Any
|
|
return_value = func(*args, **kwargs)
|
|
object.__setattr__(self, 'lazy_self', lambda: return_value)
|
|
return return_value
|
|
|
|
object.__setattr__(self, 'lazy_self', lazy_self)
|
|
|
|
def __getattribute__(self, name):
|
|
# type: (Any) -> Any
|
|
lazy_self = object.__getattribute__(self, 'lazy_self')
|
|
if name == 'lazy_self':
|
|
return lazy_self
|
|
return getattr(lazy_self(), name)
|
|
|
|
def __setattr__(self, name, value):
|
|
# type: (Any, Any) -> None
|
|
setattr(self.lazy_self(), name, value)
|
|
|
|
|
|
RegExp = partial(LazyEval, re.compile)
|
|
|
|
timestamp_regexp = RegExp(
|
|
"""^(?P<year>[0-9][0-9][0-9][0-9])
|
|
-(?P<month>[0-9][0-9]?)
|
|
-(?P<day>[0-9][0-9]?)
|
|
(?:((?P<t>[Tt])|[ \\t]+) # explictly not retaining extra spaces
|
|
(?P<hour>[0-9][0-9]?)
|
|
:(?P<minute>[0-9][0-9])
|
|
:(?P<second>[0-9][0-9])
|
|
(?:\\.(?P<fraction>[0-9]*))?
|
|
(?:[ \\t]*(?P<tz>Z|(?P<tz_sign>[-+])(?P<tz_hour>[0-9][0-9]?)
|
|
(?::(?P<tz_minute>[0-9][0-9]))?))?)?$""",
|
|
re.X,
|
|
)
|
|
|
|
|
|
def create_timestamp(
|
|
year, month, day, t, hour, minute, second, fraction, tz, tz_sign, tz_hour, tz_minute
|
|
):
|
|
# type: (Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any, Any) -> Any
|
|
# create a timestamp from match against timestamp_regexp
|
|
MAX_FRAC = 999999
|
|
year = int(year)
|
|
month = int(month)
|
|
day = int(day)
|
|
if not hour:
|
|
return datetime.date(year, month, day)
|
|
hour = int(hour)
|
|
minute = int(minute)
|
|
second = int(second)
|
|
frac = 0
|
|
if fraction:
|
|
frac_s = fraction[:6]
|
|
while len(frac_s) < 6:
|
|
frac_s += '0'
|
|
frac = int(frac_s)
|
|
if len(fraction) > 6 and int(fraction[6]) > 4:
|
|
frac += 1
|
|
if frac > MAX_FRAC:
|
|
fraction = 0
|
|
else:
|
|
fraction = frac
|
|
else:
|
|
fraction = 0
|
|
delta = None
|
|
if tz_sign:
|
|
tz_hour = int(tz_hour)
|
|
tz_minute = int(tz_minute) if tz_minute else 0
|
|
delta = datetime.timedelta(
|
|
hours=tz_hour, minutes=tz_minute, seconds=1 if frac > MAX_FRAC else 0
|
|
)
|
|
if tz_sign == '-':
|
|
delta = -delta
|
|
elif frac > MAX_FRAC:
|
|
delta = -datetime.timedelta(seconds=1)
|
|
# should do something else instead (or hook this up to the preceding if statement
|
|
# in reverse
|
|
# if delta is None:
|
|
# return datetime.datetime(year, month, day, hour, minute, second, fraction)
|
|
# return datetime.datetime(year, month, day, hour, minute, second, fraction,
|
|
# datetime.timezone.utc)
|
|
# the above is not good enough though, should provide tzinfo. In Python3 that is easily
|
|
# doable drop that kind of support for Python2 as it has not native tzinfo
|
|
data = datetime.datetime(year, month, day, hour, minute, second, fraction)
|
|
if delta:
|
|
data -= delta
|
|
return data
|
|
|
|
|
|
# originally as comment
|
|
# https://github.com/pre-commit/pre-commit/pull/211#issuecomment-186466605
|
|
# if you use this in your code, I suggest adding a test in your test suite
|
|
# that check this routines output against a known piece of your YAML
|
|
# before upgrades to this code break your round-tripped YAML
|
|
def load_yaml_guess_indent(stream, **kw):
|
|
# type: (StreamTextType, Any) -> Any
|
|
"""guess the indent and block sequence indent of yaml stream/string
|
|
|
|
returns round_trip_loaded stream, indent level, block sequence indent
|
|
- block sequence indent is the number of spaces before a dash relative to previous indent
|
|
- if there are no block sequences, indent is taken from nested mappings, block sequence
|
|
indent is unset (None) in that case
|
|
"""
|
|
from .main import YAML
|
|
|
|
# load a YAML document, guess the indentation, if you use TABs you are on your own
|
|
def leading_spaces(line):
|
|
# type: (Any) -> int
|
|
idx = 0
|
|
while idx < len(line) and line[idx] == ' ':
|
|
idx += 1
|
|
return idx
|
|
|
|
if isinstance(stream, str):
|
|
yaml_str = stream # type: Any
|
|
elif isinstance(stream, bytes):
|
|
# most likely, but the Reader checks BOM for this
|
|
yaml_str = stream.decode('utf-8')
|
|
else:
|
|
yaml_str = stream.read()
|
|
map_indent = None
|
|
indent = None # default if not found for some reason
|
|
block_seq_indent = None
|
|
prev_line_key_only = None
|
|
key_indent = 0
|
|
for line in yaml_str.splitlines():
|
|
rline = line.rstrip()
|
|
lline = rline.lstrip()
|
|
if lline.startswith('- '):
|
|
l_s = leading_spaces(line)
|
|
block_seq_indent = l_s - key_indent
|
|
idx = l_s + 1
|
|
while line[idx] == ' ': # this will end as we rstripped
|
|
idx += 1
|
|
if line[idx] == '#': # comment after -
|
|
continue
|
|
indent = idx - key_indent
|
|
break
|
|
if map_indent is None and prev_line_key_only is not None and rline:
|
|
idx = 0
|
|
while line[idx] in ' -':
|
|
idx += 1
|
|
if idx > prev_line_key_only:
|
|
map_indent = idx - prev_line_key_only
|
|
if rline.endswith(':'):
|
|
key_indent = leading_spaces(line)
|
|
idx = 0
|
|
while line[idx] == ' ': # this will end on ':'
|
|
idx += 1
|
|
prev_line_key_only = idx
|
|
continue
|
|
prev_line_key_only = None
|
|
if indent is None and map_indent is not None:
|
|
indent = map_indent
|
|
yaml = YAML()
|
|
return yaml.load(yaml_str, **kw), indent, block_seq_indent # type: ignore
|
|
|
|
|
|
def configobj_walker(cfg):
|
|
# type: (Any) -> Any
|
|
"""
|
|
walks over a ConfigObj (INI file with comments) generating
|
|
corresponding YAML output (including comments
|
|
"""
|
|
from configobj import ConfigObj # type: ignore
|
|
|
|
assert isinstance(cfg, ConfigObj)
|
|
for c in cfg.initial_comment:
|
|
if c.strip():
|
|
yield c
|
|
for s in _walk_section(cfg):
|
|
if s.strip():
|
|
yield s
|
|
for c in cfg.final_comment:
|
|
if c.strip():
|
|
yield c
|
|
|
|
|
|
def _walk_section(s, level=0):
|
|
# type: (Any, int) -> Any
|
|
from configobj import Section
|
|
|
|
assert isinstance(s, Section)
|
|
indent = ' ' * level
|
|
for name in s.scalars:
|
|
for c in s.comments[name]:
|
|
yield indent + c.strip()
|
|
x = s[name]
|
|
if '\n' in x:
|
|
i = indent + ' '
|
|
x = '|\n' + i + x.strip().replace('\n', '\n' + i)
|
|
elif ':' in x:
|
|
x = "'" + x.replace("'", "''") + "'"
|
|
line = '{0}{1}: {2}'.format(indent, name, x)
|
|
c = s.inline_comments[name]
|
|
if c:
|
|
line += ' ' + c
|
|
yield line
|
|
for name in s.sections:
|
|
for c in s.comments[name]:
|
|
yield indent + c.strip()
|
|
line = '{0}{1}:'.format(indent, name)
|
|
c = s.inline_comments[name]
|
|
if c:
|
|
line += ' ' + c
|
|
yield line
|
|
for val in _walk_section(s[name], level=level + 1):
|
|
yield val
|
|
|
|
|
|
# def config_obj_2_rt_yaml(cfg):
|
|
# from .comments import CommentedMap, CommentedSeq
|
|
# from configobj import ConfigObj
|
|
# assert isinstance(cfg, ConfigObj)
|
|
# #for c in cfg.initial_comment:
|
|
# # if c.strip():
|
|
# # pass
|
|
# cm = CommentedMap()
|
|
# for name in s.sections:
|
|
# cm[name] = d = CommentedMap()
|
|
#
|
|
#
|
|
# #for c in cfg.final_comment:
|
|
# # if c.strip():
|
|
# # yield c
|
|
# return cm
|