149 lines
5.2 KiB
Python
149 lines
5.2 KiB
Python
from pathlib import Path
|
|
from typing import List
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from .entry import Entry
|
|
from .entry.background_entry import BackgroundEntry
|
|
from .entry.parse_error_entry import ParseErrorEntry
|
|
from .entry.read_entry import ReadEntry
|
|
from .entry.reasoning_entry import ReasoningEntry
|
|
from .entry.repeat_entry import RepeatEntry
|
|
from .entry.single_entry import SingleEntry
|
|
from .entry.write_entry import WriteEntry
|
|
|
|
class IterationParser:
|
|
"""Parses iteration XML files into entries"""
|
|
|
|
@staticmethod
|
|
def parse_iteration(content: str, work_dir: Path, io_buffer) -> tuple[str, str, List[Entry]]:
|
|
"""
|
|
Parse iteration XML content into context, response and entries
|
|
|
|
Args:
|
|
content: Raw XML content as string
|
|
io_buffer: Buffer to use for IO operations
|
|
|
|
Returns:
|
|
Tuple of (context_str, response_str, entry_list)
|
|
"""
|
|
root = ET.fromstring(content)
|
|
context_elem = root.find("context")
|
|
|
|
if context_elem is None:
|
|
raise ValueError("Invalid iteration file - missing context")
|
|
|
|
context = ET.fromstring(context_elem.text)
|
|
|
|
entries = []
|
|
for elem in context:
|
|
if elem.tag == "background":
|
|
entries.append(IterationParser._parse_background(elem, work_dir))
|
|
elif elem.tag == "parse_error":
|
|
entries.append(IterationParser._parse_parse_error(elem))
|
|
elif elem.tag == "read_stdin":
|
|
entries.append(IterationParser._parse_read(elem, io_buffer))
|
|
elif elem.tag == "reasoning":
|
|
entries.append(IterationParser._parse_reasoning(elem))
|
|
elif elem.tag == "repeat":
|
|
entries.append(IterationParser._parse_repeat(elem, work_dir))
|
|
elif elem.tag == "single":
|
|
entries.append(IterationParser._parse_single(elem, work_dir))
|
|
elif elem.tag == "write_stdout":
|
|
entries.append(IterationParser._parse_write(elem, io_buffer))
|
|
|
|
return entries
|
|
|
|
@staticmethod
|
|
def _parse_background(elem: ET.Element, work_dir: Path) -> BackgroundEntry:
|
|
entry = BackgroundEntry(
|
|
id=elem.get("id"),
|
|
script=elem.text,
|
|
work_dir=work_dir
|
|
)
|
|
if elem.get("exit_code"):
|
|
entry.exit_code = int(elem.get("exit_code"))
|
|
stdout = elem.find("stdout")
|
|
if stdout is not None:
|
|
entry.stdout = stdout.text or ""
|
|
stderr = elem.find("stderr")
|
|
if stderr is not None:
|
|
entry.stderr = stderr.text or ""
|
|
return entry
|
|
|
|
@staticmethod
|
|
def _parse_parse_error(elem: ET.Element) -> ParseErrorEntry:
|
|
error = elem.find("error")
|
|
content = elem.find("content")
|
|
return ParseErrorEntry(
|
|
id=elem.get("id"),
|
|
content=content.text if content is not None else "",
|
|
error=error.text if error is not None else ""
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_read(elem: ET.Element, io_buffer) -> ReadEntry:
|
|
entry = ReadEntry(
|
|
id=elem.get("id"),
|
|
io_buffer=io_buffer
|
|
)
|
|
entry.content = elem.text or ""
|
|
entry.read = True
|
|
return entry
|
|
|
|
@staticmethod
|
|
def _parse_reasoning(elem: ET.Element) -> ReasoningEntry:
|
|
return ReasoningEntry(
|
|
id=elem.get("id"),
|
|
content=elem.text or ""
|
|
)
|
|
|
|
@staticmethod
|
|
def _parse_repeat(elem: ET.Element, work_dir: Path) -> RepeatEntry:
|
|
entry = RepeatEntry(
|
|
id=elem.get("id"),
|
|
work_dir=work_dir,
|
|
script=elem.text,
|
|
timeout=float(elem.get("timeout")) if elem.get("timeout") else None,
|
|
limit=int(elem.get("limit")) if elem.get("limit") else None
|
|
)
|
|
if elem.get("exit_code"):
|
|
entry.exit_code = int(elem.get("exit_code"))
|
|
entry.timed_out = elem.get("timed_out") == "true"
|
|
stdout = elem.find("stdout")
|
|
if stdout is not None:
|
|
entry.stdout = stdout.text or ""
|
|
stderr = elem.find("stderr")
|
|
if stderr is not None:
|
|
entry.stderr = stderr.text or ""
|
|
return entry
|
|
|
|
@staticmethod
|
|
def _parse_single(elem: ET.Element, work_dir: Path) -> SingleEntry:
|
|
entry = SingleEntry(
|
|
id=elem.get("id"),
|
|
work_dir=work_dir,
|
|
script=elem.text,
|
|
timeout=float(elem.get("timeout")) if elem.get("timeout") else None,
|
|
limit=int(elem.get("limit")) if elem.get("limit") else None
|
|
)
|
|
if elem.get("exit_code"):
|
|
entry.exit_code = int(elem.get("exit_code"))
|
|
entry.executed = True
|
|
entry.timed_out = elem.get("timed_out") == "true"
|
|
stdout = elem.find("stdout")
|
|
if stdout is not None:
|
|
entry.stdout = stdout.text or ""
|
|
stderr = elem.find("stderr")
|
|
if stderr is not None:
|
|
entry.stderr = stderr.text or ""
|
|
return entry
|
|
|
|
@staticmethod
|
|
def _parse_write(elem: ET.Element, io_buffer) -> WriteEntry:
|
|
entry = WriteEntry(
|
|
id=elem.get("id"),
|
|
content=elem.text or "",
|
|
io_buffer=io_buffer
|
|
)
|
|
entry.written = True
|
|
return entry |