""" Tatlock's permanent tools. These tools are always available to the butler agent: - Calculator: For all mathematical operations - Date/Time toolkit: For current time and time calculations - SearXNG search: For searching the web for current information """ import logging import math import re from datetime import datetime, timedelta from typing import Any import httpx from src.core.config import config logger = logging.getLogger(__name__) # ============================================================================ # Calculator Tool # ============================================================================ def calculate(expression: str) -> str: """ Safely evaluate mathematical expressions. Supports: - Basic arithmetic: +, -, *, /, //, %, ** - Parentheses for grouping - Common math functions: sqrt, sin, cos, tan, log, exp, etc. - Constants: pi, e Args: expression: Mathematical expression to evaluate (e.g., "2 + 2", "sqrt(16)", "pi * 2") Returns: String result of the calculation or error message Examples: calculate("2 + 2") -> "4" calculate("sqrt(16) + 10") -> "14.0" calculate("pi * 2") -> "6.283185307179586" """ try: # Clean the expression expression = expression.strip() # Create safe namespace with math functions safe_dict = { # Basic math functions 'sqrt': math.sqrt, 'pow': math.pow, 'abs': abs, 'round': round, # Trigonometric 'sin': math.sin, 'cos': math.cos, 'tan': math.tan, 'asin': math.asin, 'acos': math.acos, 'atan': math.atan, # Logarithmic 'log': math.log, 'log10': math.log10, 'log2': math.log2, 'exp': math.exp, # Other 'ceil': math.ceil, 'floor': math.floor, 'factorial': math.factorial, # Constants 'pi': math.pi, 'e': math.e, } # Evaluate the expression safely result = eval(expression, {"__builtins__": {}}, safe_dict) # Format result nicely if isinstance(result, float): # Remove unnecessary decimal places if result.is_integer(): return str(int(result)) return str(round(result, 10)) return str(result) except ZeroDivisionError: return "Error: Division by zero" except Exception as e: return f"Error calculating '{expression}': {str(e)}" # ============================================================================ # Date/Time Toolkit # ============================================================================ def get_current_datetime(format_str: str = "full") -> str: """ Get the current date and time. Args: format_str: Output format - "full": Full datetime with timezone (default) - "date": Just the date (YYYY-MM-DD) - "time": Just the time (HH:MM:SS) - "iso": ISO 8601 format - Custom strftime format string Returns: Formatted current datetime string Examples: get_current_datetime("full") -> "2024-01-15 14:30:45" get_current_datetime("date") -> "2024-01-15" get_current_datetime("time") -> "14:30:45" """ now = datetime.now() if format_str == "full": return now.strftime("%Y-%m-%d %H:%M:%S") elif format_str == "date": return now.strftime("%Y-%m-%d") elif format_str == "time": return now.strftime("%H:%M:%S") elif format_str == "iso": return now.isoformat() else: # Custom format try: return now.strftime(format_str) except Exception as e: return f"Error formatting date: {str(e)}" def calculate_time_offset(offset_description: str) -> str: """ Calculate a date/time relative to now. Args: offset_description: Natural language description of time offset Examples: "1 week ago", "2 days from now", "3 months ago", "1 year from now", "5 hours ago" Returns: Formatted datetime string (YYYY-MM-DD HH:MM:SS) or error message Examples: calculate_time_offset("1 week ago") -> "2024-01-08 14:30:45" calculate_time_offset("2 days from now") -> "2024-01-17 14:30:45" calculate_time_offset("3 months ago") -> "2023-10-15 14:30:45" """ try: now = datetime.now() # Parse the offset description # Pattern: "N unit(s) ago/from now" pattern = r'(\d+)\s+(second|minute|hour|day|week|month|year)s?\s+(ago|from\s+now)' match = re.match(pattern, offset_description.lower().strip()) if not match: return f"Error: Cannot parse '{offset_description}'. Use format like '1 week ago' or '2 days from now'" amount = int(match.group(1)) unit = match.group(2) direction = match.group(3) # Calculate the offset if direction == "ago": amount = -amount if unit == "second": target = now + timedelta(seconds=amount) elif unit == "minute": target = now + timedelta(minutes=amount) elif unit == "hour": target = now + timedelta(hours=amount) elif unit == "day": target = now + timedelta(days=amount) elif unit == "week": target = now + timedelta(weeks=amount) elif unit == "month": # Approximate month as 30 days target = now + timedelta(days=amount * 30) elif unit == "year": # Approximate year as 365 days target = now + timedelta(days=amount * 365) else: return f"Error: Unknown time unit '{unit}'" return target.strftime("%Y-%m-%d %H:%M:%S") except Exception as e: return f"Error calculating time offset: {str(e)}" def time_difference(date1_str: str, date2_str: str = "now") -> str: """ Calculate the difference between two dates. Args: date1_str: First date (YYYY-MM-DD or YYYY-MM-DD HH:MM:SS) date2_str: Second date or "now" for current time (default: "now") Returns: Human-readable description of the time difference Examples: time_difference("2024-01-01", "now") -> "14 days, 14 hours" time_difference("2024-01-01", "2024-01-15") -> "14 days" """ try: # Parse date1 if len(date1_str) == 10: # YYYY-MM-DD date1 = datetime.strptime(date1_str, "%Y-%m-%d") else: date1 = datetime.strptime(date1_str, "%Y-%m-%d %H:%M:%S") # Parse date2 if date2_str.lower() == "now": date2 = datetime.now() elif len(date2_str) == 10: date2 = datetime.strptime(date2_str, "%Y-%m-%d") else: date2 = datetime.strptime(date2_str, "%Y-%m-%d %H:%M:%S") # Calculate difference diff = abs(date2 - date1) # Format human-readable days = diff.days seconds = diff.seconds hours = seconds // 3600 minutes = (seconds % 3600) // 60 parts = [] if days > 0: parts.append(f"{days} day{'s' if days != 1 else ''}") if hours > 0: parts.append(f"{hours} hour{'s' if hours != 1 else ''}") if minutes > 0 and days == 0: # Only show minutes if less than a day parts.append(f"{minutes} minute{'s' if minutes != 1 else ''}") if not parts: return "Less than a minute" return ", ".join(parts) except Exception as e: return f"Error calculating time difference: {str(e)}" # ============================================================================ # SearXNG Search Tool # ============================================================================ async def search_web(query: str, num_results: int = 5) -> str: """ Search the web using SearXNG. Args: query: Search query string num_results: Number of results to return (default: 5, max: 10) Returns: Formatted search results as a string with titles, URLs, and snippets Examples: search_web("Python async programming") -> "1. Title: ...\n URL: ...\n ..." """ try: # Limit results num_results = min(num_results, 10) # Get SearXNG host with fallback logic searxng_host = str(config.SEARXNG_HOST) # Try production host first, fall back to localhost in development hosts_to_try = [searxng_host] if config.ENVIRONMENT.value == "development" and "localhost" not in searxng_host: # Add localhost fallback for development hosts_to_try.append("http://localhost:8087") last_error = None for host in hosts_to_try: try: logger.info(f"Attempting SearXNG search at {host}") async with httpx.AsyncClient(timeout=config.SEARXNG_TIMEOUT) as client: response = await client.get( f"{host}/search", params={ "q": query, "format": "json", "pageno": 1, } ) if response.status_code == 200: data = response.json() results = data.get("results", []) if not results: return f"No results found for '{query}'" # Format results formatted_results = [] for i, result in enumerate(results[:num_results], 1): title = result.get("title", "No title") url = result.get("url", "") content = result.get("content", "No description available") formatted_results.append( f"{i}. {title}\n" f" URL: {url}\n" f" {content}\n" ) return "\n".join(formatted_results) else: last_error = f"SearXNG returned status {response.status_code}" except httpx.ConnectError: last_error = f"Cannot connect to SearXNG at {host}" logger.warning(f"SearXNG connection failed at {host}, trying next host if available") continue except Exception as e: last_error = str(e) logger.warning(f"SearXNG error at {host}: {e}") continue # All hosts failed return f"Error searching: {last_error}. Please check that SearXNG is running." except Exception as e: logger.error(f"Unexpected error in search_web: {e}", exc_info=True) return f"Error searching: {str(e)}"