path: root/lib/yaml/emitter.py

# Events should obey 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 Emitter:

    def __init__(self, writer):
        self.writer = writer
        self.states = []
        self.state = self.expect_stream_start
        self.indents = []
        self.indent = None
        self.flow_level = 0
        self.key_context = False
        self.space = True
        self.line = True
        self.allow_inline_collection = False
        self.allow_indentless_sequence = False
        self.simple_key = False
        self.event_queue = []

    def emit(self, event):
        if self.event_queue:
            self.event_queue.append(event)
            event = self.event_queue.pop(0)
        self.state(event)

    def push_back(self, event):
        self.event_queue.insert(0, event)

    def expect_stream_start(self, event):
        if isinstance(event, StreamStartEvent):
            self.state = self.expect_document_start
        else:
            raise EmitterError("expected StreamStartEvent, but got %s" % event.__class__.__name__)

    def expect_document_start(self, event):
        if isinstance(event, DocumentStartEvent):
            self.write_document_start()
            self.states.append(self.expect_document_end)
            self.state = self.expect_root_node
            self.allow_inline_collection = False
            self.allow_indentless_sequence = False
        elif isinstance(event, StreamEndEvent):
            self.writer.flush()
            self.state = self.expect_nothing
        else:
            raise EmitterError("expected DocumentStartEvent, but got %s" % event.__class__.__name__)

    def expect_document_end(self, event):
        if isinstance(event, DocumentEndEvent):
            self.write_document_end()
            self.state = self.expect_document_start
            self.allow_inline_collection = False
            self.allow_indentless_sequence = False
        else:
            raiseEmitterError("expected DocumentEndEvent, but got %s" % event.__class__.__name__)

    def expect_root_node(self, event):
        self.expect_node(event)

    def expect_node(self, event):
        empty = None
        if isinstance(event, (SequenceEvent, MappingEvent)):
            if not self.event_queue:
                return self.push_back(event)
            empty = isinstance(self.event_queue[0], CollectionEndEvent)
        if isinstance(event, AliasEvent):
            self.expect_alias(event)
        elif isinstance(event, (ScalarEvent, SequenceEvent, MappingEvent)):
            if event.anchor:
                self.write_anchor("&", event.anchor)
                self.allow_inline_collection = False
            if event.tag not in [None, u'!']:
                self.write_tag(event.tag)
                self.allow_inline_collection = False
            if isinstance(event, ScalarEvent):
                self.expect_scalar(event)
            elif isinstance(event, SequenceEvent):
                self.expect_sequence(event, empty)
            elif isinstance(event, MappingEvent):
                self.expect_mapping(event, empty)
        else:
            raise EmitterError("Expected NodeEvent, but got %s" % event.__class__.__name__)

    def expect_alias(self, event):
        self.write_anchor("*", event.anchor)
        self.state = self.states.pop()

    def expect_scalar(self, event):
        self.indents.append(self.indent)
        if self.indent is None:
            self.indent = 2
        else:
            self.indent += 2
        self.write_scalar(event.value, event.implicit, event.style)
        self.indent = self.indents.pop()
        self.state = self.states.pop()
        self.allow_inline_collection = False
        self.allow_indentless_sequence = False

    def expect_sequence(self, event, empty):
        if self.flow_level or event.flow or empty:
            self.write_indicator("[", need_space=True, provide_space=True)
            self.flow_level += 1
            self.indents.append(self.indent)
            if self.indent is None:
                self.indent = 2
            else:
                self.indent += 2
            self.state = self.expect_first_flow_sequence_item
        else:
            self.indents.append(self.indent)
            if self.indent is None:
                self.indent = 0
            else:
                self.indent += 2
            self.state = self.expect_first_block_sequence_item

    def expect_mapping(self, event, empty):
        if self.flow_level or event.flow or empty:
            self.write_indicator("{", need_space=True, provide_space=True)
            self.flow_level += 1
            self.indents.append(self.indent)
            if self.indent is None:
                self.indent = 2
            else:
                self.indent += 2
            self.state = self.expect_first_flow_mapping_key
        else:
            self.indents.append(self.indent)
            if self.indent is None:
                self.indent = 0
            else:
                self.indent += 2
            self.state = self.expect_first_block_mapping_key

    def expect_first_flow_sequence_item(self, event):
        if isinstance(event, CollectionEndEvent):
            self.indent = self.indents.pop()
            self.write_indicator("]", provide_space=True)
            self.flow_level -= 1
            self.state = self.states.pop()
        else:
            self.write_indent()
            self.states.append(self.expect_flow_sequence_item)
            self.state = self.expect_node
            self.expect_node(event)

    def expect_flow_sequence_item(self, event):
        if isinstance(event, CollectionEndEvent):
            self.write_indicator(",")
            self.indent = self.indents.pop()
            self.write_indent()
            self.write_indicator("]")
            self.flow_level -= 1
            self.state = self.states.pop()
        else:
            self.write_indicator(",")
            self.write_indent()
            self.states.append(self.expect_flow_sequence_item)
            self.state = self.expect_node
            self.expect_node(event)

    def expect_first_block_sequence_item(self, event):
        assert not isinstance(event, CollectionEndEvent)
        if not self.allow_inline_collection:
            if self.allow_indentless_sequence:
                self.indent = self.indents.pop()
                self.indents.append(self.indent)
            self.write_indent()
        self.write_indicator("-", need_space=True)
        self.allow_indentless_sequence = False
        self.allow_inline_collection = True
        self.states.append(self.expect_block_sequence_item)
        self.state = self.expect_node
        self.expect_node(event)

    def expect_block_sequence_item(self, event):
        if isinstance(event, CollectionEndEvent):
            self.indent = self.indents.pop()
            self.state = self.states.pop()
        else:
            self.write_indent()
            self.write_indicator("-", need_space=True)
            self.allow_indentless_sequence = False
            self.allow_inline_collection = True
            self.states.append(self.expect_block_sequence_item)
            self.state = self.expect_node
            self.expect_node(event)

    def expect_first_flow_mapping_key(self, event):
        if isinstance(event, CollectionEndEvent):
            self.indent = self.indents.pop()
            self.write_indicator("}")
            self.flow_level -= 1
            self.state = self.states.pop()
        else:
            self.write_indent()
            if self.is_simple(event):
                self.simple_key = True
            else:
                self.write_indicator("?", need_space=True)
            self.states.append(self.expect_flow_mapping_value)
            self.state = self.expect_node
            self.expect_node(event)

    def expect_flow_mapping_key(self, event):
        if isinstance(event, CollectionEndEvent):
            self.indent = self.indents.pop()
            self.write_indent()
            self.write_indicator("}")
            self.flow_level -= 1
            self.state = self.states.pop()
        else:
            self.write_indicator(",")
            self.write_indent()
            if self.is_simple(event):
                self.simple_key = True
            else:
                self.write_indicator("?", need_space=True)
            self.states.append(self.expect_flow_mapping_value)
            self.state = self.expect_node
            self.expect_node(event)

    def expect_flow_mapping_value(self, event):
        if self.simple_key:
            self.write_indicator(":", need_space=False)
            self.simple_key = False
        else:
            self.write_indent()
            self.write_indicator(":", need_space=True)
        self.states.append(self.expect_flow_mapping_key)
        self.state = self.expect_node
        self.expect_node(event)

    def expect_first_block_mapping_key(self, event):
        assert not isinstance(event, CollectionEndEvent)
        simple = self.is_simple(event)
        if simple is None:
            return self.push_back(event)
        if not self.allow_inline_collection:
            self.write_indent()
        if self.is_simple(event):
            self.allow_indentless_sequence = True
            self.allow_inline_collection = False
            self.simple_key = True
        else:
            self.write_indicator("?", need_space=True)
            self.allow_indentless_sequence = True
            self.allow_inline_collection = True
        self.states.append(self.expect_block_mapping_value)
        self.state = self.expect_node
        self.expect_node(event)

    def expect_block_mapping_key(self, event):
        if isinstance(event, CollectionEndEvent):
            self.indent = self.indents.pop()
            self.state = self.states.pop()
        else:
            self.write_indent()
            if self.is_simple(event):
                self.allow_indentless_sequence = True
                self.allow_inline_collection = False
                self.simple_key = True
            else:
                self.write_indicator("?", need_space=True)
                self.allow_indentless_sequence = True
                self.allow_inline_collection = True
            self.states.append(self.expect_block_mapping_value)
            self.state = self.expect_node
            self.expect_node(event)

    def expect_block_mapping_value(self, event):
        if self.simple_key:
            self.write_indicator(":", need_space=False)
            self.allow_indentless_sequence = True
            self.allow_inline_collection = False
            self.simple_key = False
        else:
            self.write_indent()
            self.write_indicator(":", need_space=True)
            self.allow_indentless_sequence = True
            self.allow_inline_collection = True
        self.states.append(self.expect_block_mapping_key)
        self.state = self.expect_node
        self.expect_node(event)

    def expect_nothing(self, event):
        raise EmitterError("expected nothing, but got %s" % event.__class__.__name__)

    def write_document_start(self):
        self.writer.write("%YAML 1.1")
        self.write_line_break()
        self.writer.write("---")
        self.space = False
        self.line = False

    def write_document_end(self):
        if not self.line:
            self.write_line_break()
        self.writer.write("...")
        self.write_line_break()

    def write_line_break(self):
        self.writer.write('\n')
        self.space = True
        self.line = True

    def write_anchor(self, indicator, name):
        if not self.space:
            self.writer.write(" ")
        self.writer.write("%s%s" % (indicator, name))
        self.space = False
        self.line = False

    def write_tag(self, tag):
        if not self.space:
            self.writer.write(" ")
        if tag.startswith("tag:yaml.org,2002:"):
            self.writer.write("!!%s" % tag[len("tag.yaml.org,2002:"):])
        else:
            self.writer.write("!<%s>" % tag)
        self.space = False
        self.line = False

    def is_simple(self, event):
        if not isinstance(event, ScalarEvent):
            return False
        if '\n' in event.value or len(event.value) > 128:
            return False
        if event.style and event.style in '|>':
            return False
        return True

    def write_scalar(self, value, implicit, style):
        if implicit:
            if not self.space:
                self.writer.write(" ")
            self.writer.write(value.encode('utf-8'))
            self.space = False
            self.line = False
        elif style in ['>', '|'] and not self.flow_level and not self.simple_key:
            if not self.space:
                self.writer.write(" ")
            self.writer.write("|-")
            self.write_line_break()
            self.write_indent()
            self.writer.write(value.encode('utf-8'))
            self.write_line_break()
        else:
            if not self.space:
                self.writer.write(" ")
            self.writer.write("\"%s\"" % value.encode('utf-8'))
            self.space = False
            self.line = False

    def write_indicator(self, indicator, need_space=False, provide_space=False):
        if need_space and not self.space:
            self.writer.write(" ")
        self.writer.write(indicator)
        self.space = provide_space
        self.line = False

    def write_indent(self):
        if not self.line:
            self.write_line_break()
        if self.indent:
            self.writer.write(" "*self.indent)
            self.line = False
        self.space = True