Spaces:
Paused
Paused
File size: 2,899 Bytes
8d1819a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 |
import asyncio
from datetime import datetime, timedelta, timezone
from python.helpers.tool import Tool, Response
from python.helpers.print_style import PrintStyle
from python.helpers.wait import managed_wait
from python.helpers.localization import Localization
class WaitTool(Tool):
async def execute(self, **kwargs) -> Response:
await self.agent.handle_intervention()
seconds = self.args.get("seconds", 0)
minutes = self.args.get("minutes", 0)
hours = self.args.get("hours", 0)
days = self.args.get("days", 0)
until_timestamp_str = self.args.get("until")
is_duration_wait = not bool(until_timestamp_str)
now = datetime.now(timezone.utc)
target_time = None
if until_timestamp_str:
try:
target_time = Localization.get().localtime_str_to_utc_dt(until_timestamp_str)
if not target_time:
raise ValueError(f"Invalid timestamp format: {until_timestamp_str}")
except ValueError as e:
return Response(
message=str(e),
break_loop=False,
)
else:
wait_duration = timedelta(
days=int(days),
hours=int(hours),
minutes=int(minutes),
seconds=int(seconds),
)
if wait_duration.total_seconds() <= 0:
return Response(
message="Wait duration must be positive.",
break_loop=False,
)
target_time = now + wait_duration
if target_time <= now:
return Response(
message=f"Target time {target_time.isoformat()} is in the past.",
break_loop=False,
)
PrintStyle.info(f"Waiting until {target_time.isoformat()}...")
target_time = await managed_wait(
agent=self.agent,
target_time=target_time,
is_duration_wait=is_duration_wait,
log=self.log,
get_heading_callback=self.get_heading
)
if self.log:
self.log.update(heading=self.get_heading("Done", done=True))
message = self.agent.read_prompt(
"fw.wait_complete.md",
target_time=target_time.isoformat()
)
return Response(
message=message,
break_loop=False,
)
def get_log_object(self):
return self.agent.context.log.log(
type="progress",
heading=self.get_heading(),
content="",
kvps=self.args,
)
def get_heading(self, text: str = "", done: bool = False):
done_icon = " icon://done_all" if done else ""
if not text:
text = f"Waiting..."
return f"icon://timer Wait: {text}{done_icon}"
|