blob: 020e6db88b7c5f71711f852527a06064b27db20c [file] [log] [blame]
import yaml, yaml.composer, yaml.constructor, yaml.resolver
class CanonicalError(yaml.YAMLError):
pass
class CanonicalScanner:
def __init__(self, data):
try:
self.data = unicode(data, 'utf-8')+u'\0'
except UnicodeDecodeError:
raise CanonicalError("utf-8 stream is expected")
self.index = 0
self.tokens = []
self.scanned = False
def check_token(self, *choices):
if not self.scanned:
self.scan()
if self.tokens:
if not choices:
return True
for choice in choices:
if isinstance(self.tokens[0], choice):
return True
return False
def peek_token(self):
if not self.scanned:
self.scan()
if self.tokens:
return self.tokens[0]
def get_token(self, choice=None):
if not self.scanned:
self.scan()
token = self.tokens.pop(0)
if choice and not isinstance(token, choice):
raise CanonicalError("unexpected token "+repr(token))
return token
def get_token_value(self):
token = self.get_token()
return token.value
def scan(self):
self.tokens.append(yaml.StreamStartToken(None, None))
while True:
self.find_token()
ch = self.data[self.index]
if ch == u'\0':
self.tokens.append(yaml.StreamEndToken(None, None))
break
elif ch == u'%':
self.tokens.append(self.scan_directive())
elif ch == u'-' and self.data[self.index:self.index+3] == u'---':
self.index += 3
self.tokens.append(yaml.DocumentStartToken(None, None))
elif ch == u'[':
self.index += 1
self.tokens.append(yaml.FlowSequenceStartToken(None, None))
elif ch == u'{':
self.index += 1
self.tokens.append(yaml.FlowMappingStartToken(None, None))
elif ch == u']':
self.index += 1
self.tokens.append(yaml.FlowSequenceEndToken(None, None))
elif ch == u'}':
self.index += 1
self.tokens.append(yaml.FlowMappingEndToken(None, None))
elif ch == u'?':
self.index += 1
self.tokens.append(yaml.KeyToken(None, None))
elif ch == u':':
self.index += 1
self.tokens.append(yaml.ValueToken(None, None))
elif ch == u',':
self.index += 1
self.tokens.append(yaml.FlowEntryToken(None, None))
elif ch == u'*' or ch == u'&':
self.tokens.append(self.scan_alias())
elif ch == u'!':
self.tokens.append(self.scan_tag())
elif ch == u'"':
self.tokens.append(self.scan_scalar())
else:
raise CanonicalError("invalid token")
self.scanned = True
DIRECTIVE = u'%YAML 1.1'
def scan_directive(self):
if self.data[self.index:self.index+len(self.DIRECTIVE)] == self.DIRECTIVE and \
self.data[self.index+len(self.DIRECTIVE)] in u' \n\0':
self.index += len(self.DIRECTIVE)
return yaml.DirectiveToken('YAML', (1, 1), None, None)
else:
raise CanonicalError("invalid directive")
def scan_alias(self):
if self.data[self.index] == u'*':
TokenClass = yaml.AliasToken
else:
TokenClass = yaml.AnchorToken
self.index += 1
start = self.index
while self.data[self.index] not in u', \n\0':
self.index += 1
value = self.data[start:self.index]
return TokenClass(value, None, None)
def scan_tag(self):
self.index += 1
start = self.index
while self.data[self.index] not in u' \n\0':
self.index += 1
value = self.data[start:self.index]
if not value:
value = u'!'
elif value[0] == u'!':
value = 'tag:yaml.org,2002:'+value[1:]
elif value[0] == u'<' and value[-1] == u'>':
value = value[1:-1]
else:
value = u'!'+value
return yaml.TagToken(value, None, None)
QUOTE_CODES = {
'x': 2,
'u': 4,
'U': 8,
}
QUOTE_REPLACES = {
u'\\': u'\\',
u'\"': u'\"',
u' ': u' ',
u'a': u'\x07',
u'b': u'\x08',
u'e': u'\x1B',
u'f': u'\x0C',
u'n': u'\x0A',
u'r': u'\x0D',
u't': u'\x09',
u'v': u'\x0B',
u'N': u'\u0085',
u'L': u'\u2028',
u'P': u'\u2029',
u'_': u'_',
u'0': u'\x00',
}
def scan_scalar(self):
self.index += 1
chunks = []
start = self.index
ignore_spaces = False
while self.data[self.index] != u'"':
if self.data[self.index] == u'\\':
ignore_spaces = False
chunks.append(self.data[start:self.index])
self.index += 1
ch = self.data[self.index]
self.index += 1
if ch == u'\n':
ignore_spaces = True
elif ch in self.QUOTE_CODES:
length = self.QUOTE_CODES[ch]
code = int(self.data[self.index:self.index+length], 16)
chunks.append(unichr(code))
self.index += length
else:
if ch not in self.QUOTE_REPLACES:
raise CanonicalError("invalid escape code")
chunks.append(self.QUOTE_REPLACES[ch])
start = self.index
elif self.data[self.index] == u'\n':
chunks.append(self.data[start:self.index])
chunks.append(u' ')
self.index += 1
start = self.index
ignore_spaces = True
elif ignore_spaces and self.data[self.index] == u' ':
self.index += 1
start = self.index
else:
ignore_spaces = False
self.index += 1
chunks.append(self.data[start:self.index])
self.index += 1
return yaml.ScalarToken(u''.join(chunks), False, None, None)
def find_token(self):
found = False
while not found:
while self.data[self.index] in u' \t':
self.index += 1
if self.data[self.index] == u'#':
while self.data[self.index] != u'\n':
self.index += 1
if self.data[self.index] == u'\n':
self.index += 1
else:
found = True
class CanonicalParser:
def __init__(self):
self.events = []
self.parsed = False
def dispose(self):
pass
# stream: STREAM-START document* STREAM-END
def parse_stream(self):
self.get_token(yaml.StreamStartToken)
self.events.append(yaml.StreamStartEvent(None, None))
while not self.check_token(yaml.StreamEndToken):
if self.check_token(yaml.DirectiveToken, yaml.DocumentStartToken):
self.parse_document()
else:
raise CanonicalError("document is expected, got "+repr(self.tokens[0]))
self.get_token(yaml.StreamEndToken)
self.events.append(yaml.StreamEndEvent(None, None))
# document: DIRECTIVE? DOCUMENT-START node
def parse_document(self):
node = None
if self.check_token(yaml.DirectiveToken):
self.get_token(yaml.DirectiveToken)
self.get_token(yaml.DocumentStartToken)
self.events.append(yaml.DocumentStartEvent(None, None))
self.parse_node()
self.events.append(yaml.DocumentEndEvent(None, None))
# node: ALIAS | ANCHOR? TAG? (SCALAR|sequence|mapping)
def parse_node(self):
if self.check_token(yaml.AliasToken):
self.events.append(yaml.AliasEvent(self.get_token_value(), None, None))
else:
anchor = None
if self.check_token(yaml.AnchorToken):
anchor = self.get_token_value()
tag = None
if self.check_token(yaml.TagToken):
tag = self.get_token_value()
if self.check_token(yaml.ScalarToken):
self.events.append(yaml.ScalarEvent(anchor, tag, (False, False), self.get_token_value(), None, None))
elif self.check_token(yaml.FlowSequenceStartToken):
self.events.append(yaml.SequenceStartEvent(anchor, tag, None, None))
self.parse_sequence()
elif self.check_token(yaml.FlowMappingStartToken):
self.events.append(yaml.MappingStartEvent(anchor, tag, None, None))
self.parse_mapping()
else:
raise CanonicalError("SCALAR, '[', or '{' is expected, got "+repr(self.tokens[0]))
# sequence: SEQUENCE-START (node (ENTRY node)*)? ENTRY? SEQUENCE-END
def parse_sequence(self):
self.get_token(yaml.FlowSequenceStartToken)
if not self.check_token(yaml.FlowSequenceEndToken):
self.parse_node()
while not self.check_token(yaml.FlowSequenceEndToken):
self.get_token(yaml.FlowEntryToken)
if not self.check_token(yaml.FlowSequenceEndToken):
self.parse_node()
self.get_token(yaml.FlowSequenceEndToken)
self.events.append(yaml.SequenceEndEvent(None, None))
# mapping: MAPPING-START (map_entry (ENTRY map_entry)*)? ENTRY? MAPPING-END
def parse_mapping(self):
self.get_token(yaml.FlowMappingStartToken)
if not self.check_token(yaml.FlowMappingEndToken):
self.parse_map_entry()
while not self.check_token(yaml.FlowMappingEndToken):
self.get_token(yaml.FlowEntryToken)
if not self.check_token(yaml.FlowMappingEndToken):
self.parse_map_entry()
self.get_token(yaml.FlowMappingEndToken)
self.events.append(yaml.MappingEndEvent(None, None))
# map_entry: KEY node VALUE node
def parse_map_entry(self):
self.get_token(yaml.KeyToken)
self.parse_node()
self.get_token(yaml.ValueToken)
self.parse_node()
def parse(self):
self.parse_stream()
self.parsed = True
def get_event(self):
if not self.parsed:
self.parse()
return self.events.pop(0)
def check_event(self, *choices):
if not self.parsed:
self.parse()
if self.events:
if not choices:
return True
for choice in choices:
if isinstance(self.events[0], choice):
return True
return False
def peek_event(self):
if not self.parsed:
self.parse()
return self.events[0]
class CanonicalLoader(CanonicalScanner, CanonicalParser,
yaml.composer.Composer, yaml.constructor.Constructor, yaml.resolver.Resolver):
def __init__(self, stream):
if hasattr(stream, 'read'):
stream = stream.read()
CanonicalScanner.__init__(self, stream)
CanonicalParser.__init__(self)
yaml.composer.Composer.__init__(self)
yaml.constructor.Constructor.__init__(self)
yaml.resolver.Resolver.__init__(self)
yaml.CanonicalLoader = CanonicalLoader
def canonical_scan(stream):
return yaml.scan(stream, Loader=CanonicalLoader)
yaml.canonical_scan = canonical_scan
def canonical_parse(stream):
return yaml.parse(stream, Loader=CanonicalLoader)
yaml.canonical_parse = canonical_parse
def canonical_compose(stream):
return yaml.compose(stream, Loader=CanonicalLoader)
yaml.canonical_compose = canonical_compose
def canonical_compose_all(stream):
return yaml.compose_all(stream, Loader=CanonicalLoader)
yaml.canonical_compose_all = canonical_compose_all
def canonical_load(stream):
return yaml.load(stream, Loader=CanonicalLoader)
yaml.canonical_load = canonical_load
def canonical_load_all(stream):
return yaml.load_all(stream, Loader=CanonicalLoader)
yaml.canonical_load_all = canonical_load_all