Xml stream parser

osmxml/__init__.py

@@ -6,11 +6,13 @@
 from .attribute import XMLAttribute
 from .text import XMLTextElement
 from .parser import XMLParser
+from .stream_parser import XMLStreamParser
 
 __all__ = [
     "XML",
     "XMLElement",
     "XMLAttribute",
     "XMLTextElement",
     "XMLParser",
+    "XMLStreamParser",
 ]

osmxml/element.py

@@ -13,6 +13,7 @@ def __init__(
         self._name = name 
         self._attributes = attributes.copy() if attributes is not None else []
         self._children = children.copy() if children is not None else []
+        self._is_closed = True
 
     @property
     def name(self) -> str:
@@ -41,6 +42,15 @@ def children(self, children: List[XML]):
         self._children = children
 
 
+    @property
+    def is_closed(self) -> bool:
+        return self.is_closed
+
+    @is_closed.setter
+    def is_closed(self, value: bool):
+        self._is_closed = value
+
+
     def add_attribute(self, attribute: XMLAttribute):
         self._attributes.append(attribute) 
 
@@ -80,7 +90,12 @@ def to_string(self) -> str:
 
         children_str = "".join(list_children_str) 
 
-        element_str = "<{name}{attrs}>{children}\n</{name}>".format(
+        closed_template = "<{name}{attrs}>{children}\n</{name}>"
+        non_closed_template = "<{name}{attrs}>{children}\n"
+
+        template = closed_template if self.is_closed else non_closed_template
+
+        element_str = template.format(
             name=self.name,
             attrs=attrs_str,
             children=children_str
@@ -101,9 +116,11 @@ def __repr__(self):
         repr = 'XMLElement(name="{name}",'
         repr = "".join([repr, " attributes=len({attrs_len}),"])
         repr = "".join([repr, " children=len({children_len}))"])
+        repr = "".join([repr, " is_closed={is_closed})"])
         repr = repr.format(
                 name=self.name,
                 attrs_len=len(self.attributes),
                 children_len=len(self.children),
+                is_closed=self.is_closed,
         )
         return repr

osmxml/stream_parser.py

@@ -0,0 +1,74 @@
+from typing import Optional, List
+
+from .element import XMLElement
+from .attribute import XMLAttribute
+from .text import XMLTextElement
+
+class XMLStreamParser:
+    def __init__(self):
+        self._element_buffer = []
+        self._leftovers = ""
+
+        self._current_element = None
+        self._current_attribute = None
+
+    def _get_symbol(self, symbol: str) -> bool:
+        return self._leftovers.startswith(symbol)
+
+    def _get_text(self) -> tuple[bool, str]:
+        if self._leftovers[0] in "<>":
+            return False, None
+
+        i = 0
+        while i < len(self._leftovers) and self._leftovers[i] not in "<> ":
+            i += 1
+
+        text = self._leftovers[:i]
+        return True, text
+
+    def _consume_symbol(self, symbol: str) -> bool:
+        if self._leftovers.startswith(symbol):
+            self._leftovers = self._leftovers[len(symbol):]
+            return True
+        return False
+
+    def _consume_text(self, text) -> bool:
+        if self._leftovers.startswith(text):
+            self._leftovers = self._leftovers[len(text):]
+            return True
+        return False
+
+    def _consume_identation(self) -> bool:
+        while len(self._leftovers) > 0:
+            if self._leftovers[0] in [" ", "\n", "\t"]:
+                self._leftovers = self._leftovers[1:]
+            else:
+                break
+
+    def _is_last_element_closed(self) -> bool:
+        return self._element_buffer[-1].is_closed
+
+
+    def parse(self, xml_string: str) -> List[XMLElement]:
+        if (not xml_string):
+            return []
+
+        self._leftovers += xml_string
+
+        if (len(self._element_buffer) == 0) or (self._is_last_element_closed()):
+            self._consume_identation()
+
+            if (self._get_symbol("<")):
+
+                # Closing element
+                if (self._get_symbol("/")):
+                    # Closed element
+                    self._consume_symbol("/")
+                    name = self._get_text()
+                    self._consume_symbol(">")
+                    return self._element_buffer.pop()
+        else:
+            # Continue parsing current element
+            pass
+
+