119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
from typing import Optional
|
|
|
|
from . import Entry
|
|
from ..io_buffer import IOBuffer
|
|
from .background_entry import BackgroundEntry
|
|
from .parse_error_entry import ParseErrorEntry
|
|
from .read_entry import ReadEntry
|
|
from .reasoning_entry import ReasoningEntry
|
|
from .repeat_entry import RepeatEntry
|
|
from .single_entry import SingleEntry
|
|
from .write_entry import WriteEntry
|
|
|
|
class EntryFactory:
|
|
def create_entry(data: dict, io_buffer: IOBuffer) -> Entry:
|
|
entry_type = data.get("type")
|
|
entry_id = data.get("id")
|
|
|
|
if entry_type == "background":
|
|
return BackgroundEntry(entry_id, data["script"])
|
|
|
|
elif entry_type == "read_stdin":
|
|
return ReadEntry(entry_id, io_buffer)
|
|
|
|
elif entry_type == "reasoning":
|
|
return ReasoningEntry(entry_id, data["content"])
|
|
|
|
elif entry_type == "repeat":
|
|
return RepeatEntry(
|
|
entry_id,
|
|
data["script"],
|
|
data.get("timeout"),
|
|
data.get("limit")
|
|
)
|
|
|
|
elif entry_type == "single":
|
|
return SingleEntry(
|
|
entry_id,
|
|
data["script"],
|
|
data.get("timeout"),
|
|
data.get("limit")
|
|
)
|
|
|
|
elif entry_type == "write":
|
|
return WriteEntry(entry_id, data["content"], io_buffer)
|
|
|
|
raise ValueError(f"Unknown entry type: {entry_type}")
|
|
|
|
def update_entry(entry: Entry, data: dict):
|
|
if isinstance(entry, Entry):
|
|
if "id" in data:
|
|
entry.id = data["id"]
|
|
|
|
if isinstance(entry, SingleEntry):
|
|
if "script" in data:
|
|
entry.script = data["script"]
|
|
if "timeout" in data:
|
|
entry.timeout = float(data["timeout"]) if data["timeout"] is not None else None
|
|
if "limit" in data:
|
|
entry.limit = int(data["limit"]) if data["limit"] else None
|
|
if "stdout" in data:
|
|
entry.stdout = data["stdout"]
|
|
if "stderr" in data:
|
|
entry.stderr = data["stderr"]
|
|
if "exit_code" in data:
|
|
entry.exit_code = int(data["exit_code"]) if data["exit_code"] is not None else None
|
|
if "executed" in data:
|
|
entry.executed = bool(data["executed"])
|
|
if "timed_out" in data:
|
|
entry.timed_out = bool(data["timed_out"])
|
|
|
|
if isinstance(entry, RepeatEntry):
|
|
if "script" in data:
|
|
entry.script = data["script"]
|
|
if "timeout" in data:
|
|
entry.timeout = float(data["timeout"]) if data["timeout"] is not None else None
|
|
if "limit" in data:
|
|
entry.limit = int(data["limit"]) if data["limit"] else None
|
|
if "stdout" in data:
|
|
entry.stdout = data["stdout"]
|
|
if "stderr" in data:
|
|
entry.stderr = data["stderr"]
|
|
if "exit_code" in data:
|
|
entry.exit_code = int(data["exit_code"]) if data["exit_code"] is not None else None
|
|
if "executed" in data:
|
|
entry.executed = bool(data["executed"])
|
|
if "timed_out" in data:
|
|
entry.timed_out = bool(data["timed_out"])
|
|
|
|
elif isinstance(entry, BackgroundEntry):
|
|
if "script" in data:
|
|
entry.script = data["script"]
|
|
if "stdout" in data:
|
|
entry.stdout = data["stdout"]
|
|
if "stderr" in data:
|
|
entry.stderr = data["stderr"]
|
|
if "exit_code" in data:
|
|
entry.exit_code = int(data["exit_code"]) if data["exit_code"] is not None else None
|
|
|
|
elif isinstance(entry, ParseErrorEntry):
|
|
if "content" in data:
|
|
entry.content = data["content"]
|
|
if "error" in data:
|
|
entry.error = data["error"]
|
|
|
|
elif isinstance(entry, ReadEntry):
|
|
if "content" in data:
|
|
entry.content = data["content"]
|
|
if "read" in data:
|
|
entry.read = bool(data["read"])
|
|
|
|
elif isinstance(entry, ReasoningEntry):
|
|
if "content" in data:
|
|
entry.content = data["content"]
|
|
|
|
elif isinstance(entry, WriteEntry):
|
|
if "content" in data:
|
|
entry.content = data["content"]
|
|
if "written" in data:
|
|
entry.written = bool(data["written"]) |