mirror of
https://github.com/All-Hands-AI/OpenHands.git
synced 2026-01-14 09:18:04 -05:00
87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
|
|
from openhands.events.tool import ToolCallMetadata
|
|
from openhands.llm.metrics import Metrics
|
|
|
|
|
|
class EventSource(str, Enum):
|
|
AGENT = 'agent'
|
|
USER = 'user'
|
|
ENVIRONMENT = 'environment'
|
|
|
|
|
|
@dataclass
|
|
class Event:
|
|
@property
|
|
def message(self) -> str | None:
|
|
if hasattr(self, '_message'):
|
|
return self._message # type: ignore[attr-defined]
|
|
return ''
|
|
|
|
@property
|
|
def id(self) -> int:
|
|
if hasattr(self, '_id'):
|
|
return self._id # type: ignore[attr-defined]
|
|
return -1
|
|
|
|
@property
|
|
def timestamp(self):
|
|
if hasattr(self, '_timestamp') and isinstance(self._timestamp, str):
|
|
return self._timestamp
|
|
|
|
@timestamp.setter
|
|
def timestamp(self, value: datetime) -> None:
|
|
if isinstance(value, datetime):
|
|
self._timestamp = value.isoformat()
|
|
|
|
@property
|
|
def source(self) -> EventSource | None:
|
|
if hasattr(self, '_source'):
|
|
return self._source # type: ignore[attr-defined]
|
|
return None
|
|
|
|
@property
|
|
def cause(self) -> int | None:
|
|
if hasattr(self, '_cause'):
|
|
return self._cause # type: ignore[attr-defined]
|
|
return None
|
|
|
|
@property
|
|
def timeout(self) -> int | None:
|
|
if hasattr(self, '_timeout'):
|
|
return self._timeout # type: ignore[attr-defined]
|
|
return None
|
|
|
|
@timeout.setter
|
|
def timeout(self, value: int | None) -> None:
|
|
self._timeout = value
|
|
|
|
# Check if .blocking is an attribute of the event
|
|
if hasattr(self, 'blocking'):
|
|
# .blocking needs to be set to True if .timeout is set
|
|
self.blocking = True
|
|
|
|
# optional metadata, LLM call cost of the edit
|
|
@property
|
|
def llm_metrics(self) -> Metrics | None:
|
|
if hasattr(self, '_llm_metrics'):
|
|
return self._llm_metrics # type: ignore[attr-defined]
|
|
return None
|
|
|
|
@llm_metrics.setter
|
|
def llm_metrics(self, value: Metrics) -> None:
|
|
self._llm_metrics = value
|
|
|
|
# optional field
|
|
@property
|
|
def tool_call_metadata(self) -> ToolCallMetadata | None:
|
|
if hasattr(self, '_tool_call_metadata'):
|
|
return self._tool_call_metadata # type: ignore[attr-defined]
|
|
return None
|
|
|
|
@tool_call_metadata.setter
|
|
def tool_call_metadata(self, value: ToolCallMetadata) -> None:
|
|
self._tool_call_metadata = value
|