Spaces:
Runtime error
Runtime error
| # Emitter expects events obeying the following grammar: | |
| # stream ::= STREAM-START document* STREAM-END | |
| # document ::= DOCUMENT-START node DOCUMENT-END | |
| # node ::= SCALAR | sequence | mapping | |
| # sequence ::= SEQUENCE-START node* SEQUENCE-END | |
| # mapping ::= MAPPING-START (node node)* MAPPING-END | |
| __all__ = ['Emitter', 'EmitterError'] | |
| from .error import YAMLError | |
| from .events import * | |
| class EmitterError(YAMLError): | |
| pass | |
| class ScalarAnalysis: | |
| def __init__(self, scalar, empty, multiline, | |
| allow_flow_plain, allow_block_plain, | |
| allow_single_quoted, allow_double_quoted, | |
| allow_block): | |
| self.scalar = scalar | |
| self.empty = empty | |
| self.multiline = multiline | |
| self.allow_flow_plain = allow_flow_plain | |
| self.allow_block_plain = allow_block_plain | |
| self.allow_single_quoted = allow_single_quoted | |
| self.allow_double_quoted = allow_double_quoted | |
| self.allow_block = allow_block | |
| class Emitter: | |
| DEFAULT_TAG_PREFIXES = { | |
| '!' : '!', | |
| 'tag:yaml.org,2002:' : '!!', | |
| } | |
| def __init__(self, stream, canonical=None, indent=None, width=None, | |
| allow_unicode=None, line_break=None): | |
| # The stream should have the methods `write` and possibly `flush`. | |
| self.stream = stream | |
| # Encoding can be overridden by STREAM-START. | |
| self.encoding = None | |
| # Emitter is a state machine with a stack of states to handle nested | |
| # structures. | |
| self.states = [] | |
| self.state = self.expect_stream_start | |
| # Current event and the event queue. | |
| self.events = [] | |
| self.event = None | |
| # The current indentation level and the stack of previous indents. | |
| self.indents = [] | |
| self.indent = None | |
| # Flow level. | |
| self.flow_level = 0 | |
| # Contexts. | |
| self.root_context = False | |
| self.sequence_context = False | |
| self.mapping_context = False | |
| self.simple_key_context = False | |
| # Characteristics of the last emitted character: | |
| # - current position. | |
| # - is it a whitespace? | |
| # - is it an indention character | |
| # (indentation space, '-', '?', or ':')? | |
| self.line = 0 | |
| self.column = 0 | |
| self.whitespace = True | |
| self.indention = True | |
| # Whether the document requires an explicit document indicator | |
| self.open_ended = False | |
| # Formatting details. | |
| self.canonical = canonical | |
| self.allow_unicode = allow_unicode | |
| self.best_indent = 2 | |
| if indent and 1 < indent < 10: | |
| self.best_indent = indent | |
| self.best_width = 80 | |
| if width and width > self.best_indent*2: | |
| self.best_width = width | |
| self.best_line_break = '\n' | |
| if line_break in ['\r', '\n', '\r\n']: | |
| self.best_line_break = line_break | |
| # Tag prefixes. | |
| self.tag_prefixes = None | |
| # Prepared anchor and tag. | |
| self.prepared_anchor = None | |
| self.prepared_tag = None | |
| # Scalar analysis and style. | |
| self.analysis = None | |
| self.style = None | |
| def dispose(self): | |
| # Reset the state attributes (to clear self-references) | |
| self.states = [] | |
| self.state = None | |
| def emit(self, event): | |
| self.events.append(event) | |
| while not self.need_more_events(): | |
| self.event = self.events.pop(0) | |
| self.state() | |
| self.event = None | |
| # In some cases, we wait for a few next events before emitting. | |
| def need_more_events(self): | |
| if not self.events: | |
| return True | |
| event = self.events[0] | |
| if isinstance(event, DocumentStartEvent): | |
| return self.need_events(1) | |
| elif isinstance(event, SequenceStartEvent): | |
| return self.need_events(2) | |
| elif isinstance(event, MappingStartEvent): | |
| return self.need_events(3) | |
| else: | |
| return False | |
| def need_events(self, count): | |
| level = 0 | |
| for event in self.events[1:]: | |
| if isinstance(event, (DocumentStartEvent, CollectionStartEvent)): | |
| level += 1 | |
| elif isinstance(event, (DocumentEndEvent, CollectionEndEvent)): | |
| level -= 1 | |
| elif isinstance(event, StreamEndEvent): | |
| level = -1 | |
| if level < 0: | |
| return False | |
| return (len(self.events) < count+1) | |
| def increase_indent(self, flow=False, indentless=False): | |
| self.indents.append(self.indent) | |
| if self.indent is None: | |
| if flow: | |
| self.indent = self.best_indent | |
| else: | |
| self.indent = 0 | |
| elif not indentless: | |
| self.indent += self.best_indent | |
| # States. | |
| # Stream handlers. | |
| def expect_stream_start(self): | |
| if isinstance(self.event, StreamStartEvent): | |
| if self.event.encoding and not hasattr(self.stream, 'encoding'): | |
| self.encoding = self.event.encoding | |
| self.write_stream_start() | |
| self.state = self.expect_first_document_start | |
| else: | |
| raise EmitterError("expected StreamStartEvent, but got %s" | |
| % self.event) | |
| def expect_nothing(self): | |
| raise EmitterError("expected nothing, but got %s" % self.event) | |
| # Document handlers. | |
| def expect_first_document_start(self): | |
| return self.expect_document_start(first=True) | |
| def expect_document_start(self, first=False): | |
| if isinstance(self.event, DocumentStartEvent): | |
| if (self.event.version or self.event.tags) and self.open_ended: | |
| self.write_indicator('...', True) | |
| self.write_indent() | |
| if self.event.version: | |
| version_text = self.prepare_version(self.event.version) | |
| self.write_version_directive(version_text) | |
| self.tag_prefixes = self.DEFAULT_TAG_PREFIXES.copy() | |
| if self.event.tags: | |
| handles = sorted(self.event.tags.keys()) | |
| for handle in handles: | |
| prefix = self.event.tags[handle] | |
| self.tag_prefixes[prefix] = handle | |
| handle_text = self.prepare_tag_handle(handle) | |
| prefix_text = self.prepare_tag_prefix(prefix) | |
| self.write_tag_directive(handle_text, prefix_text) | |
| implicit = (first and not self.event.explicit and not self.canonical | |
| and not self.event.version and not self.event.tags | |
| and not self.check_empty_document()) | |
| if not implicit: | |
| self.write_indent() | |
| self.write_indicator('---', True) | |
| if self.canonical: | |
| self.write_indent() | |
| self.state = self.expect_document_root | |
| elif isinstance(self.event, StreamEndEvent): | |
| if self.open_ended: | |
| self.write_indicator('...', True) | |
| self.write_indent() | |
| self.write_stream_end() | |
| self.state = self.expect_nothing | |
| else: | |
| raise EmitterError("expected DocumentStartEvent, but got %s" | |
| % self.event) | |
| def expect_document_end(self): | |
| if isinstance(self.event, DocumentEndEvent): | |
| self.write_indent() | |
| if self.event.explicit: | |
| self.write_indicator('...', True) | |
| self.write_indent() | |
| self.flush_stream() | |
| self.state = self.expect_document_start | |
| else: | |
| raise EmitterError("expected DocumentEndEvent, but got %s" | |
| % self.event) | |
| def expect_document_root(self): | |
| self.states.append(self.expect_document_end) | |
| self.expect_node(root=True) | |
| # Node handlers. | |
| def expect_node(self, root=False, sequence=False, mapping=False, | |
| simple_key=False): | |
| self.root_context = root | |
| self.sequence_context = sequence | |
| self.mapping_context = mapping | |
| self.simple_key_context = simple_key | |
| if isinstance(self.event, AliasEvent): | |
| self.expect_alias() | |
| elif isinstance(self.event, (ScalarEvent, CollectionStartEvent)): | |
| self.process_anchor('&') | |
| self.process_tag() | |
| if isinstance(self.event, ScalarEvent): | |
| self.expect_scalar() | |
| elif isinstance(self.event, SequenceStartEvent): | |
| if self.flow_level or self.canonical or self.event.flow_style \ | |
| or self.check_empty_sequence(): | |
| self.expect_flow_sequence() | |
| else: | |
| self.expect_block_sequence() | |
| elif isinstance(self.event, MappingStartEvent): | |
| if self.flow_level or self.canonical or self.event.flow_style \ | |
| or self.check_empty_mapping(): | |
| self.expect_flow_mapping() | |
| else: | |
| self.expect_block_mapping() | |
| else: | |
| raise EmitterError("expected NodeEvent, but got %s" % self.event) | |
| def expect_alias(self): | |
| if self.event.anchor is None: | |
| raise EmitterError("anchor is not specified for alias") | |
| self.process_anchor('*') | |
| self.state = self.states.pop() | |
| def expect_scalar(self): | |
| self.increase_indent(flow=True) | |
| self.process_scalar() | |
| self.indent = self.indents.pop() | |
| self.state = self.states.pop() | |
| # Flow sequence handlers. | |
| def expect_flow_sequence(self): | |
| self.write_indicator('[', True, whitespace=True) | |
| self.flow_level += 1 | |
| self.increase_indent(flow=True) | |
| self.state = self.expect_first_flow_sequence_item | |
| def expect_first_flow_sequence_item(self): | |
| if isinstance(self.event, SequenceEndEvent): | |
| self.indent = self.indents.pop() | |
| self.flow_level -= 1 | |
| self.write_indicator(']', False) | |
| self.state = self.states.pop() | |
| else: | |
| if self.canonical or self.column > self.best_width: | |
| self.write_indent() | |
| self.states.append(self.expect_flow_sequence_item) | |
| self.expect_node(sequence=True) | |
| def expect_flow_sequence_item(self): | |
| if isinstance(self.event, SequenceEndEvent): | |
| self.indent = self.indents.pop() | |
| self.flow_level -= 1 | |
| if self.canonical: | |
| self.write_indicator(',', False) | |
| self.write_indent() | |
| self.write_indicator(']', False) | |
| self.state = self.states.pop() | |
| else: | |
| self.write_indicator(',', False) | |
| if self.canonical or self.column > self.best_width: | |
| self.write_indent() | |
| self.states.append(self.expect_flow_sequence_item) | |
| self.expect_node(sequence=True) | |
| # Flow mapping handlers. | |
| def expect_flow_mapping(self): | |
| self.write_indicator('{', True, whitespace=True) | |
| self.flow_level += 1 | |
| self.increase_indent(flow=True) | |
| self.state = self.expect_first_flow_mapping_key | |
| def expect_first_flow_mapping_key(self): | |
| if isinstance(self.event, MappingEndEvent): | |
| self.indent = self.indents.pop() | |
| self.flow_level -= 1 | |
| self.write_indicator('}', False) | |
| self.state = self.states.pop() | |
| else: | |
| if self.canonical or self.column > self.best_width: | |
| self.write_indent() | |
| if not self.canonical and self.check_simple_key(): | |
| self.states.append(self.expect_flow_mapping_simple_value) | |
| self.expect_node(mapping=True, simple_key=True) | |
| else: | |
| self.write_indicator('?', True) | |
| self.states.append(self.expect_flow_mapping_value) | |
| self.expect_node(mapping=True) | |
| def expect_flow_mapping_key(self): | |
| if isinstance(self.event, MappingEndEvent): | |
| self.indent = self.indents.pop() | |
| self.flow_level -= 1 | |
| if self.canonical: | |
| self.write_indicator(',', False) | |
| self.write_indent() | |
| self.write_indicator('}', False) | |
| self.state = self.states.pop() | |
| else: | |
| self.write_indicator(',', False) | |
| if self.canonical or self.column > self.best_width: | |
| self.write_indent() | |
| if not self.canonical and self.check_simple_key(): | |
| self.states.append(self.expect_flow_mapping_simple_value) | |
| self.expect_node(mapping=True, simple_key=True) | |
| else: | |
| self.write_indicator('?', True) | |
| self.states.append(self.expect_flow_mapping_value) | |
| self.expect_node(mapping=True) | |
| def expect_flow_mapping_simple_value(self): | |
| self.write_indicator(':', False) | |
| self.states.append(self.expect_flow_mapping_key) | |
| self.expect_node(mapping=True) | |
| def expect_flow_mapping_value(self): | |
| if self.canonical or self.column > self.best_width: | |
| self.write_indent() | |
| self.write_indicator(':', True) | |
| self.states.append(self.expect_flow_mapping_key) | |
| self.expect_node(mapping=True) | |
| # Block sequence handlers. | |
| def expect_block_sequence(self): | |
| indentless = (self.mapping_context and not self.indention) | |
| self.increase_indent(flow=False, indentless=indentless) | |
| self.state = self.expect_first_block_sequence_item | |
| def expect_first_block_sequence_item(self): | |
| return self.expect_block_sequence_item(first=True) | |
| def expect_block_sequence_item(self, first=False): | |
| if not first and isinstance(self.event, SequenceEndEvent): | |
| self.indent = self.indents.pop() | |
| self.state = self.states.pop() | |
| else: | |
| self.write_indent() | |
| self.write_indicator('-', True, indention=True) | |
| self.states.append(self.expect_block_sequence_item) | |
| self.expect_node(sequence=True) | |
| # Block mapping handlers. | |
| def expect_block_mapping(self): | |
| self.increase_indent(flow=False) | |
| self.state = self.expect_first_block_mapping_key | |
| def expect_first_block_mapping_key(self): | |
| return self.expect_block_mapping_key(first=True) | |
| def expect_block_mapping_key(self, first=False): | |
| if not first and isinstance(self.event, MappingEndEvent): | |
| self.indent = self.indents.pop() | |
| self.state = self.states.pop() | |
| else: | |
| self.write_indent() | |
| if self.check_simple_key(): | |
| self.states.append(self.expect_block_mapping_simple_value) | |
| self.expect_node(mapping=True, simple_key=True) | |
| else: | |
| self.write_indicator('?', True, indention=True) | |
| self.states.append(self.expect_block_mapping_value) | |
| self.expect_node(mapping=True) | |
| def expect_block_mapping_simple_value(self): | |
| self.write_indicator(':', False) | |
| self.states.append(self.expect_block_mapping_key) | |
| self.expect_node(mapping=True) | |
| def expect_block_mapping_value(self): | |
| self.write_indent() | |
| self.write_indicator(':', True, indention=True) | |
| self.states.append(self.expect_block_mapping_key) | |
| self.expect_node(mapping=True) | |
| # Checkers. | |
| def check_empty_sequence(self): | |
| return (isinstance(self.event, SequenceStartEvent) and self.events | |
| and isinstance(self.events[0], SequenceEndEvent)) | |
| def check_empty_mapping(self): | |
| return (isinstance(self.event, MappingStartEvent) and self.events | |
| and isinstance(self.events[0], MappingEndEvent)) | |
| def check_empty_document(self): | |
| if not isinstance(self.event, DocumentStartEvent) or not self.events: | |
| return False | |
| event = self.events[0] | |
| return (isinstance(event, ScalarEvent) and event.anchor is None | |
| and event.tag is None and event.implicit and event.value == '') | |
| def check_simple_key(self): | |
| length = 0 | |
| if isinstance(self.event, NodeEvent) and self.event.anchor is not None: | |
| if self.prepared_anchor is None: | |
| self.prepared_anchor = self.prepare_anchor(self.event.anchor) | |
| length += len(self.prepared_anchor) | |
| if isinstance(self.event, (ScalarEvent, CollectionStartEvent)) \ | |
| and self.event.tag is not None: | |
| if self.prepared_tag is None: | |
| self.prepared_tag = self.prepare_tag(self.event.tag) | |
| length += len(self.prepared_tag) | |
| if isinstance(self.event, ScalarEvent): | |
| if self.analysis is None: | |
| self.analysis = self.analyze_scalar(self.event.value) | |
| length += len(self.analysis.scalar) | |
| return (length < 128 and (isinstance(self.event, AliasEvent) | |
| or (isinstance(self.event, ScalarEvent) | |
| and not self.analysis.empty and not self.analysis.multiline) | |
| or self.check_empty_sequence() or self.check_empty_mapping())) | |
| # Anchor, Tag, and Scalar processors. | |
| def process_anchor(self, indicator): | |
| if self.event.anchor is None: | |
| self.prepared_anchor = None | |
| return | |
| if self.prepared_anchor is None: | |
| self.prepared_anchor = self.prepare_anchor(self.event.anchor) | |
| if self.prepared_anchor: | |
| self.write_indicator(indicator+self.prepared_anchor, True) | |
| self.prepared_anchor = None | |
| def process_tag(self): | |
| tag = self.event.tag | |
| if isinstance(self.event, ScalarEvent): | |
| if self.style is None: | |
| self.style = self.choose_scalar_style() | |
| if ((not self.canonical or tag is None) and | |
| ((self.style == '' and self.event.implicit[0]) | |
| or (self.style != '' and self.event.implicit[1]))): | |
| self.prepared_tag = None | |
| return | |
| if self.event.implicit[0] and tag is None: | |
| tag = '!' | |
| self.prepared_tag = None | |
| else: | |
| if (not self.canonical or tag is None) and self.event.implicit: | |
| self.prepared_tag = None | |
| return | |
| if tag is None: | |
| raise EmitterError("tag is not specified") | |
| if self.prepared_tag is None: | |
| self.prepared_tag = self.prepare_tag(tag) | |
| if self.prepared_tag: | |
| self.write_indicator(self.prepared_tag, True) | |
| self.prepared_tag = None | |
| def choose_scalar_style(self): | |
| if self.analysis is None: | |
| self.analysis = self.analyze_scalar(self.event.value) | |
| if self.event.style == '"' or self.canonical: | |
| return '"' | |
| if not self.event.style and self.event.implicit[0]: | |
| if (not (self.simple_key_context and | |
| (self.analysis.empty or self.analysis.multiline)) | |
| and (self.flow_level and self.analysis.allow_flow_plain | |
| or (not self.flow_level and self.analysis.allow_block_plain))): | |
| return '' | |
| if self.event.style and self.event.style in '|>': | |
| if (not self.flow_level and not self.simple_key_context | |
| and self.analysis.allow_block): | |
| return self.event.style | |
| if not self.event.style or self.event.style == '\'': | |
| if (self.analysis.allow_single_quoted and | |
| not (self.simple_key_context and self.analysis.multiline)): | |
| return '\'' | |
| return '"' | |
| def process_scalar(self): | |
| if self.analysis is None: | |
| self.analysis = self.analyze_scalar(self.event.value) | |
| if self.style is None: | |
| self.style = self.choose_scalar_style() | |
| split = (not self.simple_key_context) | |
| #if self.analysis.multiline and split \ | |
| # and (not self.style or self.style in '\'\"'): | |
| # self.write_indent() | |
| if self.style == '"': | |
| self.write_double_quoted(self.analysis.scalar, split) | |
| elif self.style == '\'': | |
| self.write_single_quoted(self.analysis.scalar, split) | |
| elif self.style == '>': | |
| self.write_folded(self.analysis.scalar) | |
| elif self.style == '|': | |
| self.write_literal(self.analysis.scalar) | |
| else: | |
| self.write_plain(self.analysis.scalar, split) | |
| self.analysis = None | |
| self.style = None | |
| # Analyzers. | |
| def prepare_version(self, version): | |
| major, minor = version | |
| if major != 1: | |
| raise EmitterError("unsupported YAML version: %d.%d" % (major, minor)) | |
| return '%d.%d' % (major, minor) | |
| def prepare_tag_handle(self, handle): | |
| if not handle: | |
| raise EmitterError("tag handle must not be empty") | |
| if handle[0] != '!' or handle[-1] != '!': | |
| raise EmitterError("tag handle must start and end with '!': %r" % handle) | |
| for ch in handle[1:-1]: | |
| if not ('0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' \ | |
| or ch in '-_'): | |
| raise EmitterError("invalid character %r in the tag handle: %r" | |
| % (ch, handle)) | |
| return handle | |
| def prepare_tag_prefix(self, prefix): | |
| if not prefix: | |
| raise EmitterError("tag prefix must not be empty") | |
| chunks = [] | |
| start = end = 0 | |
| if prefix[0] == '!': | |
| end = 1 | |
| while end < len(prefix): | |
| ch = prefix[end] | |
| if '0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' \ | |
| or ch in '-;/?!:@&=+$,_.~*\'()[]': | |
| end += 1 | |
| else: | |
| if start < end: | |
| chunks.append(prefix[start:end]) | |
| start = end = end+1 | |
| data = ch.encode('utf-8') | |
| for ch in data: | |
| chunks.append('%%%02X' % ord(ch)) | |
| if start < end: | |
| chunks.append(prefix[start:end]) | |
| return ''.join(chunks) | |
| def prepare_tag(self, tag): | |
| if not tag: | |
| raise EmitterError("tag must not be empty") | |
| if tag == '!': | |
| return tag | |
| handle = None | |
| suffix = tag | |
| prefixes = sorted(self.tag_prefixes.keys()) | |
| for prefix in prefixes: | |
| if tag.startswith(prefix) \ | |
| and (prefix == '!' or len(prefix) < len(tag)): | |
| handle = self.tag_prefixes[prefix] | |
| suffix = tag[len(prefix):] | |
| chunks = [] | |
| start = end = 0 | |
| while end < len(suffix): | |
| ch = suffix[end] | |
| if '0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' \ | |
| or ch in '-;/?:@&=+$,_.~*\'()[]' \ | |
| or (ch == '!' and handle != '!'): | |
| end += 1 | |
| else: | |
| if start < end: | |
| chunks.append(suffix[start:end]) | |
| start = end = end+1 | |
| data = ch.encode('utf-8') | |
| for ch in data: | |
| chunks.append('%%%02X' % ch) | |
| if start < end: | |
| chunks.append(suffix[start:end]) | |
| suffix_text = ''.join(chunks) | |
| if handle: | |
| return '%s%s' % (handle, suffix_text) | |
| else: | |
| return '!<%s>' % suffix_text | |
| def prepare_anchor(self, anchor): | |
| if not anchor: | |
| raise EmitterError("anchor must not be empty") | |
| for ch in anchor: | |
| if not ('0' <= ch <= '9' or 'A' <= ch <= 'Z' or 'a' <= ch <= 'z' \ | |
| or ch in '-_'): | |
| raise EmitterError("invalid character %r in the anchor: %r" | |
| % (ch, anchor)) | |
| return anchor | |
| def analyze_scalar(self, scalar): | |
| # Empty scalar is a special case. | |
| if not scalar: | |
| return ScalarAnalysis(scalar=scalar, empty=True, multiline=False, | |
| allow_flow_plain=False, allow_block_plain=True, | |
| allow_single_quoted=True, allow_double_quoted=True, | |
| allow_block=False) | |
| # Indicators and special characters. | |
| block_indicators = False | |
| flow_indicators = False | |
| line_breaks = False | |
| special_characters = False | |
| # Important whitespace combinations. | |
| leading_space = False | |
| leading_break = False | |
| trailing_space = False | |
| trailing_break = False | |
| break_space = False | |
| space_break = False | |
| # Check document indicators. | |
| if scalar.startswith('---') or scalar.startswith('...'): | |
| block_indicators = True | |
| flow_indicators = True | |
| # First character or preceded by a whitespace. | |
| preceded_by_whitespace = True | |
| # Last character or followed by a whitespace. | |
| followed_by_whitespace = (len(scalar) == 1 or | |
| scalar[1] in '\0 \t\r\n\x85\u2028\u2029') | |
| # The previous character is a space. | |
| previous_space = False | |
| # The previous character is a break. | |
| previous_break = False | |
| index = 0 | |
| while index < len(scalar): | |
| ch = scalar[index] | |
| # Check for indicators. | |
| if index == 0: | |
| # Leading indicators are special characters. | |
| if ch in '#,[]{}&*!|>\'\"%@`': | |
| flow_indicators = True | |
| block_indicators = True | |
| if ch in '?:': | |
| flow_indicators = True | |
| if followed_by_whitespace: | |
| block_indicators = True | |
| if ch == '-' and followed_by_whitespace: | |
| flow_indicators = True | |
| block_indicators = True | |
| else: | |
| # Some indicators cannot appear within a scalar as well. | |
| if ch in ',?[]{}': | |
| flow_indicators = True | |
| if ch == ':': | |
| flow_indicators = True | |
| if followed_by_whitespace: | |
| block_indicators = True | |
| if ch == '#' and preceded_by_whitespace: | |
| flow_indicators = True | |
| block_indicators = True | |
| # Check for line breaks, special, and unicode characters. | |
| if ch in '\n\x85\u2028\u2029': | |
| line_breaks = True | |
| if not (ch == '\n' or '\x20' <= ch <= '\x7E'): | |
| if (ch == '\x85' or '\xA0' <= ch <= '\uD7FF' | |
| or '\uE000' <= ch <= '\uFFFD' | |
| or '\U00010000' <= ch < '\U0010ffff') and ch != '\uFEFF': | |
| unicode_characters = True | |
| if not self.allow_unicode: | |
| special_characters = True | |
| else: | |
| special_characters = True | |
| # Detect important whitespace combinations. | |
| if ch == ' ': | |
| if index == 0: | |
| leading_space = True | |
| if index == len(scalar)-1: | |
| trailing_space = True | |
| if previous_break: | |
| break_space = True | |
| previous_space = True | |
| previous_break = False | |
| elif ch in '\n\x85\u2028\u2029': | |
| if index == 0: | |
| leading_break = True | |
| if index == len(scalar)-1: | |
| trailing_break = True | |
| if previous_space: | |
| space_break = True | |
| previous_space = False | |
| previous_break = True | |
| else: | |
| previous_space = False | |
| previous_break = False | |
| # Prepare for the next character. | |
| index += 1 | |
| preceded_by_whitespace = (ch in '\0 \t\r\n\x85\u2028\u2029') | |
| followed_by_whitespace = (index+1 >= len(scalar) or | |
| scalar[index+1] in '\0 \t\r\n\x85\u2028\u2029') | |
| # Let's decide what styles are allowed. | |
| allow_flow_plain = True | |
| allow_block_plain = True | |
| allow_single_quoted = True | |
| allow_double_quoted = True | |
| allow_block = True | |
| # Leading and trailing whitespaces are bad for plain scalars. | |
| if (leading_space or leading_break | |
| or trailing_space or trailing_break): | |
| allow_flow_plain = allow_block_plain = False | |
| # We do not permit trailing spaces for block scalars. | |
| if trailing_space: | |
| allow_block = False | |
| # Spaces at the beginning of a new line are only acceptable for block | |
| # scalars. | |
| if break_space: | |
| allow_flow_plain = allow_block_plain = allow_single_quoted = False | |
| # Spaces followed by breaks, as well as special character are only | |
| # allowed for double quoted scalars. | |
| if space_break or special_characters: | |
| allow_flow_plain = allow_block_plain = \ | |
| allow_single_quoted = allow_block = False | |
| # Although the plain scalar writer supports breaks, we never emit | |
| # multiline plain scalars. | |
| if line_breaks: | |
| allow_flow_plain = allow_block_plain = False | |
| # Flow indicators are forbidden for flow plain scalars. | |
| if flow_indicators: | |
| allow_flow_plain = False | |
| # Block indicators are forbidden for block plain scalars. | |
| if block_indicators: | |
| allow_block_plain = False | |
| return ScalarAnalysis(scalar=scalar, | |
| empty=False, multiline=line_breaks, | |
| allow_flow_plain=allow_flow_plain, | |
| allow_block_plain=allow_block_plain, | |
| allow_single_quoted=allow_single_quoted, | |
| allow_double_quoted=allow_double_quoted, | |
| allow_block=allow_block) | |
| # Writers. | |
| def flush_stream(self): | |
| if hasattr(self.stream, 'flush'): | |
| self.stream.flush() | |
| def write_stream_start(self): | |
| # Write BOM if needed. | |
| if self.encoding and self.encoding.startswith('utf-16'): | |
| self.stream.write('\uFEFF'.encode(self.encoding)) | |
| def write_stream_end(self): | |
| self.flush_stream() | |
| def write_indicator(self, indicator, need_whitespace, | |
| whitespace=False, indention=False): | |
| if self.whitespace or not need_whitespace: | |
| data = indicator | |
| else: | |
| data = ' '+indicator | |
| self.whitespace = whitespace | |
| self.indention = self.indention and indention | |
| self.column += len(data) | |
| self.open_ended = False | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| def write_indent(self): | |
| indent = self.indent or 0 | |
| if not self.indention or self.column > indent \ | |
| or (self.column == indent and not self.whitespace): | |
| self.write_line_break() | |
| if self.column < indent: | |
| self.whitespace = True | |
| data = ' '*(indent-self.column) | |
| self.column = indent | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| def write_line_break(self, data=None): | |
| if data is None: | |
| data = self.best_line_break | |
| self.whitespace = True | |
| self.indention = True | |
| self.line += 1 | |
| self.column = 0 | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| def write_version_directive(self, version_text): | |
| data = '%%YAML %s' % version_text | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| self.write_line_break() | |
| def write_tag_directive(self, handle_text, prefix_text): | |
| data = '%%TAG %s %s' % (handle_text, prefix_text) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| self.write_line_break() | |
| # Scalar streams. | |
| def write_single_quoted(self, text, split=True): | |
| self.write_indicator('\'', True) | |
| spaces = False | |
| breaks = False | |
| start = end = 0 | |
| while end <= len(text): | |
| ch = None | |
| if end < len(text): | |
| ch = text[end] | |
| if spaces: | |
| if ch is None or ch != ' ': | |
| if start+1 == end and self.column > self.best_width and split \ | |
| and start != 0 and end != len(text): | |
| self.write_indent() | |
| else: | |
| data = text[start:end] | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end | |
| elif breaks: | |
| if ch is None or ch not in '\n\x85\u2028\u2029': | |
| if text[start] == '\n': | |
| self.write_line_break() | |
| for br in text[start:end]: | |
| if br == '\n': | |
| self.write_line_break() | |
| else: | |
| self.write_line_break(br) | |
| self.write_indent() | |
| start = end | |
| else: | |
| if ch is None or ch in ' \n\x85\u2028\u2029' or ch == '\'': | |
| if start < end: | |
| data = text[start:end] | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end | |
| if ch == '\'': | |
| data = '\'\'' | |
| self.column += 2 | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end + 1 | |
| if ch is not None: | |
| spaces = (ch == ' ') | |
| breaks = (ch in '\n\x85\u2028\u2029') | |
| end += 1 | |
| self.write_indicator('\'', False) | |
| ESCAPE_REPLACEMENTS = { | |
| '\0': '0', | |
| '\x07': 'a', | |
| '\x08': 'b', | |
| '\x09': 't', | |
| '\x0A': 'n', | |
| '\x0B': 'v', | |
| '\x0C': 'f', | |
| '\x0D': 'r', | |
| '\x1B': 'e', | |
| '\"': '\"', | |
| '\\': '\\', | |
| '\x85': 'N', | |
| '\xA0': '_', | |
| '\u2028': 'L', | |
| '\u2029': 'P', | |
| } | |
| def write_double_quoted(self, text, split=True): | |
| self.write_indicator('"', True) | |
| start = end = 0 | |
| while end <= len(text): | |
| ch = None | |
| if end < len(text): | |
| ch = text[end] | |
| if ch is None or ch in '"\\\x85\u2028\u2029\uFEFF' \ | |
| or not ('\x20' <= ch <= '\x7E' | |
| or (self.allow_unicode | |
| and ('\xA0' <= ch <= '\uD7FF' | |
| or '\uE000' <= ch <= '\uFFFD'))): | |
| if start < end: | |
| data = text[start:end] | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end | |
| if ch is not None: | |
| if ch in self.ESCAPE_REPLACEMENTS: | |
| data = '\\'+self.ESCAPE_REPLACEMENTS[ch] | |
| elif ch <= '\xFF': | |
| data = '\\x%02X' % ord(ch) | |
| elif ch <= '\uFFFF': | |
| data = '\\u%04X' % ord(ch) | |
| else: | |
| data = '\\U%08X' % ord(ch) | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end+1 | |
| if 0 < end < len(text)-1 and (ch == ' ' or start >= end) \ | |
| and self.column+(end-start) > self.best_width and split: | |
| data = text[start:end]+'\\' | |
| if start < end: | |
| start = end | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| self.write_indent() | |
| self.whitespace = False | |
| self.indention = False | |
| if text[start] == ' ': | |
| data = '\\' | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| end += 1 | |
| self.write_indicator('"', False) | |
| def determine_block_hints(self, text): | |
| hints = '' | |
| if text: | |
| if text[0] in ' \n\x85\u2028\u2029': | |
| hints += str(self.best_indent) | |
| if text[-1] not in '\n\x85\u2028\u2029': | |
| hints += '-' | |
| elif len(text) == 1 or text[-2] in '\n\x85\u2028\u2029': | |
| hints += '+' | |
| return hints | |
| def write_folded(self, text): | |
| hints = self.determine_block_hints(text) | |
| self.write_indicator('>'+hints, True) | |
| if hints[-1:] == '+': | |
| self.open_ended = True | |
| self.write_line_break() | |
| leading_space = True | |
| spaces = False | |
| breaks = True | |
| start = end = 0 | |
| while end <= len(text): | |
| ch = None | |
| if end < len(text): | |
| ch = text[end] | |
| if breaks: | |
| if ch is None or ch not in '\n\x85\u2028\u2029': | |
| if not leading_space and ch is not None and ch != ' ' \ | |
| and text[start] == '\n': | |
| self.write_line_break() | |
| leading_space = (ch == ' ') | |
| for br in text[start:end]: | |
| if br == '\n': | |
| self.write_line_break() | |
| else: | |
| self.write_line_break(br) | |
| if ch is not None: | |
| self.write_indent() | |
| start = end | |
| elif spaces: | |
| if ch != ' ': | |
| if start+1 == end and self.column > self.best_width: | |
| self.write_indent() | |
| else: | |
| data = text[start:end] | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end | |
| else: | |
| if ch is None or ch in ' \n\x85\u2028\u2029': | |
| data = text[start:end] | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| if ch is None: | |
| self.write_line_break() | |
| start = end | |
| if ch is not None: | |
| breaks = (ch in '\n\x85\u2028\u2029') | |
| spaces = (ch == ' ') | |
| end += 1 | |
| def write_literal(self, text): | |
| hints = self.determine_block_hints(text) | |
| self.write_indicator('|'+hints, True) | |
| if hints[-1:] == '+': | |
| self.open_ended = True | |
| self.write_line_break() | |
| breaks = True | |
| start = end = 0 | |
| while end <= len(text): | |
| ch = None | |
| if end < len(text): | |
| ch = text[end] | |
| if breaks: | |
| if ch is None or ch not in '\n\x85\u2028\u2029': | |
| for br in text[start:end]: | |
| if br == '\n': | |
| self.write_line_break() | |
| else: | |
| self.write_line_break(br) | |
| if ch is not None: | |
| self.write_indent() | |
| start = end | |
| else: | |
| if ch is None or ch in '\n\x85\u2028\u2029': | |
| data = text[start:end] | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| if ch is None: | |
| self.write_line_break() | |
| start = end | |
| if ch is not None: | |
| breaks = (ch in '\n\x85\u2028\u2029') | |
| end += 1 | |
| def write_plain(self, text, split=True): | |
| if self.root_context: | |
| self.open_ended = True | |
| if not text: | |
| return | |
| if not self.whitespace: | |
| data = ' ' | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| self.whitespace = False | |
| self.indention = False | |
| spaces = False | |
| breaks = False | |
| start = end = 0 | |
| while end <= len(text): | |
| ch = None | |
| if end < len(text): | |
| ch = text[end] | |
| if spaces: | |
| if ch != ' ': | |
| if start+1 == end and self.column > self.best_width and split: | |
| self.write_indent() | |
| self.whitespace = False | |
| self.indention = False | |
| else: | |
| data = text[start:end] | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end | |
| elif breaks: | |
| if ch not in '\n\x85\u2028\u2029': | |
| if text[start] == '\n': | |
| self.write_line_break() | |
| for br in text[start:end]: | |
| if br == '\n': | |
| self.write_line_break() | |
| else: | |
| self.write_line_break(br) | |
| self.write_indent() | |
| self.whitespace = False | |
| self.indention = False | |
| start = end | |
| else: | |
| if ch is None or ch in ' \n\x85\u2028\u2029': | |
| data = text[start:end] | |
| self.column += len(data) | |
| if self.encoding: | |
| data = data.encode(self.encoding) | |
| self.stream.write(data) | |
| start = end | |
| if ch is not None: | |
| spaces = (ch == ' ') | |
| breaks = (ch in '\n\x85\u2028\u2029') | |
| end += 1 | |