236 lines
7.3 KiB
Python
236 lines
7.3 KiB
Python
from typing import TypedDict, Generator, Iterator, Optional, Callable, List, Any, Union
|
|
from parser import LineData, createLineData, parseLine
|
|
import io
|
|
|
|
# Type for a reader function
|
|
Reader = Callable[[], Optional[str]]
|
|
|
|
class TerraceNode:
|
|
"""Represents a single node/line in a Terrace document"""
|
|
|
|
def __init__(self, line_data: LineData, content: str, line_number: int, document: 'TerraceDocument'):
|
|
self._line_data = line_data.copy() # Copy to avoid mutations
|
|
self._content = content
|
|
self._line_number = line_number
|
|
self._document = document
|
|
|
|
@property
|
|
def head(self) -> str:
|
|
"""Get the first word of the line"""
|
|
return self._content[self._line_data['offsetHead']:self._line_data['offsetTail']]
|
|
|
|
@property
|
|
def tail(self) -> str:
|
|
"""Get everything after the first word"""
|
|
if self._line_data['offsetTail'] >= len(self._content) or self._content[self._line_data['offsetTail']] != ' ':
|
|
return ""
|
|
return self._content[self._line_data['offsetTail'] + 1:]
|
|
|
|
@property
|
|
def content(self) -> str:
|
|
"""Get the line content without indentation"""
|
|
return self._content[self._line_data['offsetHead']:]
|
|
|
|
@property
|
|
def level(self) -> int:
|
|
"""Get the indentation level"""
|
|
return self._line_data['level']
|
|
|
|
@property
|
|
def line_number(self) -> int:
|
|
"""Get the line number (zero-indexed)"""
|
|
return self._line_number
|
|
|
|
def is_(self, value: str) -> bool:
|
|
"""Check if the head matches the given value"""
|
|
return self.head == value
|
|
|
|
def is_empty(self) -> bool:
|
|
"""Check if the line is empty/blank"""
|
|
return self.content.strip() == ''
|
|
|
|
def raw(self, offset: Optional[int] = None) -> str:
|
|
"""Get raw content with custom offset"""
|
|
if offset is None:
|
|
offset = self.level
|
|
return self._content[offset:]
|
|
|
|
def children(self) -> Generator['TerraceNode', None, None]:
|
|
"""Iterate through all descendant nodes (supports arbitrary nesting)"""
|
|
parent_level = self.level
|
|
|
|
while True:
|
|
node = self._document._get_next_node()
|
|
if node is None:
|
|
break
|
|
|
|
if node.level <= parent_level:
|
|
# Put back the node for parent iteration
|
|
self._document._push_back(node)
|
|
break
|
|
|
|
# Yield any node that is deeper than the parent
|
|
# This supports arbitrary nesting as per Terrace spec
|
|
yield node
|
|
|
|
def siblings(self) -> Generator['TerraceNode', None, None]:
|
|
"""Iterate through sibling nodes at the same level"""
|
|
current_level = self.level
|
|
|
|
while True:
|
|
node = self._document._get_next_node()
|
|
if node is None:
|
|
break
|
|
|
|
if node.level < current_level:
|
|
self._document._push_back(node)
|
|
break
|
|
|
|
if node.level == current_level:
|
|
yield node
|
|
|
|
class TerraceDocument:
|
|
"""Main document iterator for Terrace documents"""
|
|
|
|
def __init__(self, reader: Reader, indent: str = ' '):
|
|
if len(indent) != 1:
|
|
raise ValueError(f"Terrace currently only allows single-character indent strings - you passed '{indent}'")
|
|
|
|
self._reader = reader
|
|
self._indent = indent
|
|
self._line_data = createLineData(indent)
|
|
self._current_line_number = -1
|
|
self._pushed_back_node: Optional[TerraceNode] = None
|
|
|
|
def __iter__(self) -> Iterator[TerraceNode]:
|
|
"""Make the document iterable"""
|
|
return self._create_iterator()
|
|
|
|
def _create_iterator(self) -> Generator[TerraceNode, None, None]:
|
|
"""Create the main iterator generator"""
|
|
while True:
|
|
# Check for pushed back node first
|
|
if self._pushed_back_node is not None:
|
|
node = self._pushed_back_node
|
|
self._pushed_back_node = None
|
|
yield node
|
|
continue
|
|
|
|
line = self._reader()
|
|
if line is None:
|
|
break
|
|
|
|
self._current_line_number += 1
|
|
parseLine(line, self._line_data)
|
|
|
|
node = TerraceNode(
|
|
self._line_data,
|
|
line,
|
|
self._current_line_number,
|
|
self
|
|
)
|
|
|
|
yield node
|
|
|
|
def _get_next_node(self) -> Optional[TerraceNode]:
|
|
"""Get the next node from the document"""
|
|
if self._pushed_back_node is not None:
|
|
node = self._pushed_back_node
|
|
self._pushed_back_node = None
|
|
return node
|
|
|
|
line = self._reader()
|
|
if line is None:
|
|
return None
|
|
|
|
self._current_line_number += 1
|
|
parseLine(line, self._line_data)
|
|
|
|
return TerraceNode(
|
|
self._line_data,
|
|
line,
|
|
self._current_line_number,
|
|
self
|
|
)
|
|
|
|
def _push_back(self, node: TerraceNode) -> None:
|
|
"""Push back a node to be returned by the next iteration"""
|
|
self._pushed_back_node = node
|
|
|
|
# Utility methods for functional programming style
|
|
def filter(self, predicate: Callable[[TerraceNode], bool]) -> List[TerraceNode]:
|
|
"""Filter nodes by predicate"""
|
|
return [node for node in self if predicate(node)]
|
|
|
|
def find(self, predicate: Callable[[TerraceNode], bool]) -> Optional[TerraceNode]:
|
|
"""Find the first node matching predicate"""
|
|
for node in self:
|
|
if predicate(node):
|
|
return node
|
|
return None
|
|
|
|
def map(self, mapper: Callable[[TerraceNode], Any]) -> List[Any]:
|
|
"""Map nodes through a function"""
|
|
return [mapper(node) for node in self]
|
|
|
|
def to_list(self) -> List[TerraceNode]:
|
|
"""Convert all nodes to a list"""
|
|
return list(self)
|
|
|
|
# Convenience functions for creating readers
|
|
def create_string_reader(content: str) -> Reader:
|
|
"""Create a reader from a string"""
|
|
lines = content.split('\n')
|
|
|
|
# Remove trailing empty line if content ended with newline (like Rust fix)
|
|
if len(lines) > 0 and content.endswith('\n') and lines[-1] == '':
|
|
lines = lines[:-1]
|
|
|
|
index = 0
|
|
|
|
def reader() -> Optional[str]:
|
|
nonlocal index
|
|
if index >= len(lines):
|
|
return None
|
|
line = lines[index]
|
|
index += 1
|
|
return line
|
|
|
|
return reader
|
|
|
|
def create_file_reader(file_path: str) -> Reader:
|
|
"""Create a reader from a file path"""
|
|
file_handle = open(file_path, 'r', encoding='utf-8')
|
|
|
|
def reader() -> Optional[str]:
|
|
line = file_handle.readline()
|
|
if not line:
|
|
file_handle.close()
|
|
return None
|
|
return line.rstrip('\n\r')
|
|
|
|
return reader
|
|
|
|
def create_lines_reader(lines: List[str]) -> Reader:
|
|
"""Create a reader from a list of lines"""
|
|
index = 0
|
|
|
|
def reader() -> Optional[str]:
|
|
nonlocal index
|
|
if index >= len(lines):
|
|
return None
|
|
line = lines[index]
|
|
index += 1
|
|
return line.rstrip('\n\r')
|
|
|
|
return reader
|
|
|
|
# Main factory function
|
|
def use_document(reader: Reader, indent: str = ' ') -> TerraceDocument:
|
|
"""Create a new Terrace document iterator"""
|
|
return TerraceDocument(reader, indent)
|
|
|
|
# Legacy compatibility
|
|
def useDocument(reader: Reader, indent: str = ' ') -> TerraceDocument:
|
|
"""Legacy alias for use_document"""
|
|
return use_document(reader, indent) |