Added weather, file system, and memory

1 files changed, 232 insertions, 1 deletions

diff --git a/tools.py b/tools.py
index 64238e7..2ae8857 100644
--- a/tools.py
+++ b/tools.py
@@ -1,3 +1,6 @@
+from datetime import datetime, timedelta, timezone, UTC
+from pathlib import Path
+import requests
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 
@@ -77,11 +80,12 @@ class HelloTool(Tool):
         name = arguments.get("name", "stranger")
         return ToolResult(content=f"Hello, {name}!")
 
+
 class CalculatorTool(Tool):
     @property
     def name(self) -> str:
         return "calculator"
-    
+
     @property
     def description(self) -> str:
         return "Perform basic arithemtic"
@@ -137,6 +141,233 @@ class CalculatorTool(Tool):
 
         return ToolResult(content=str(result))
 
+
+class WeatherTool(Tool):
+    @property
+    def name(self) -> str:
+        return "weather"
+
+    @property
+    def description(self) -> str:
+        return "Get current weather conditions for a location"
+
+    @property
+    def schema(self) -> dict[str, object]:
+        return {
+            "type": "object",
+            "properties": {
+                "location": {
+                    "type": "string",
+                    "description": ("The city or location to get weather for."),
+                },
+            },
+            "required": ["location"],
+        }
+
+    def execute(self, arguments: dict[str, object]) -> ToolResult:
+        location = arguments.get("location")
+
+        if not isinstance(location, str):
+            raise TypeError("location must be a string")
+
+        geocoding_response = requests.get(
+            "https://geocoding-api.open-meteo.com/v1/search",
+            params={
+                "name": location,
+                "count": 1,
+                "language": "en",
+                "format": "json",
+            },
+            timeout=10,
+        )
+        geocoding_response.raise_for_status()
+
+        geocoding_data = geocoding_response.json()
+        results = geocoding_data.get("results")
+
+        if not results:
+            raise ValueError(f"location not found: {location}")
+
+        place = results[0]
+
+        latitude = place["latitude"]
+        longitude = place["longitude"]
+        timezone_name = place["timezone"]
+
+        forecast_response = requests.get(
+            "https://api.open-meteo.com/v1/forecast",
+            params={
+                "latitude": latitude,
+                "longitude": longitude,
+                "hourly": (
+                    "temperature_2m,"
+                    "precipitation_probability,"
+                    "precipitation,"
+                    "weather_code,"
+                    "wind_speed_10m"
+                ),
+                "timezone": timezone_name,
+                "forecast_days": 2,
+            },
+            timeout=10,
+        )
+        forecast_response.raise_for_status()
+
+        forecast = forecast_response.json()
+        hourly = forecast["hourly"]
+
+        tomorrow = (
+            (datetime.now(UTC) + timedelta(days=1))
+            .astimezone(timezone(timedelta(seconds=forecast["utc_offset_seconds"])))
+            .date()
+        )
+
+        temperatures = []
+        precipitation_probabilities = []
+        precipitation = []
+        wind_speeds = []
+        weather_codes = []
+
+        for index, timestamp in enumerate(hourly["time"]):
+            forecast_date = datetime.fromisoformat(timestamp).date()
+
+            if forecast_date != tomorrow:
+                continue
+
+            temperatures.append(hourly["temperature_2m"][index])
+            precipitation_probabilities.append(
+                hourly["precipitation_probability"][index]
+            )
+            precipitation.append(hourly["precipitation"][index])
+            wind_speeds.append(hourly["wind_speed_10m"][index])
+            weather_codes.append(hourly["weather_code"][index])
+
+        if not temperatures:
+            raise ValueError("no forecast data available for tomorrow")
+        return ToolResult(
+            content=(
+                f"Location: {place['name']}, {place['country']}\n"
+                f"Forecast date: {tomorrow}\n"
+                f"Temperature: {min(temperatures):.1f}–"
+                f"{max(temperatures):.1f} °C\n"
+                f"Maximum precipitation probability: "
+                f"{max(precipitation_probabilities)}%\n"
+                f"Total precipitation: {sum(precipitation):.1f} mm\n"
+                f"Maximum wind speed: {max(wind_speeds):.1f} km/h\n"
+                f"Weather codes: {sorted(set(weather_codes))}"
+            )
+        )
+
+
+class FileSystemListTool(Tool):
+    @property
+    def name(self) -> str:
+        return "filesystem_list"
+
+    @property
+    def description(self) -> str:
+        return "List files and directories inside the agent's sandbox."
+
+    @property
+    def schema(self) -> dict[str, object]:
+        return {
+            "type": "object",
+            "properties": {
+                "path": {
+                    "type": "string",
+                    "description": (
+                        "Relative path inside the filesystem sandbox. "
+                        "Use '.' for the sandbox root."
+                    ),
+                },
+            },
+            "required": ["path"],
+        }
+
+    def execute(self, arguments: dict[str, object]) -> ToolResult:
+        path = arguments.get("path")
+
+        if not isinstance(path, str):
+            raise TypeError("path must be a string")
+
+        sandbox = Path.home() / "Projects" / "personal-agent" / "data"
+        target = (sandbox / path).resolve()
+
+        if not target.is_relative_to(sandbox.resolve()):
+            raise ValueError("path is outside the filesystem sandbox")
+
+        if not target.exists():
+            raise ValueError(f"path does not exist: {path}")
+
+        if not target.is_dir():
+            raise ValueError(f"path is not a directory: {path}")
+
+        entries = []
+
+        for entry in sorted(target.iterdir()):
+            kind = "directory" if entry.is_dir() else "file"
+            entries.append(f"{kind}: {entry.name}")
+
+        if not entries:
+            return ToolResult(content="Directory is empty.")
+
+        return ToolResult(content="\n".join(entries))
+
+
+class FileSystemReadTool(Tool):
+    MAX_FILE_SIZE = 1024 * 1024  # 1 MiB
+
+    @property
+    def name(self) -> str:
+        return "filesystem_read"
+
+    @property
+    def description(self) -> str:
+        return "Read the contents of a file inside the agent's sandbox."
+
+    @property
+    def schema(self) -> dict[str, object]:
+        return {
+            "type": "object",
+            "properties": {
+                "path": {
+                    "type": "string",
+                    "description": ("Relative path inside the filesystem sandbox."),
+                },
+            },
+            "required": ["path"],
+        }
+
+    def execute(self, arguments: dict[str, object]) -> ToolResult:
+        path = arguments.get("path")
+
+        if not isinstance(path, str):
+            raise TypeError("path must be a string")
+
+        sandbox = Path.home() / "Projects" / "personal-agent" / "data"
+        sandbox = sandbox.resolve()
+        target = (sandbox / path).resolve()
+
+        if not target.is_relative_to(sandbox):
+            raise ValueError("path is outside the filesystem sandbox")
+
+        if not target.exists():
+            raise ValueError(f"path does not exist: {path}")
+
+        if not target.is_file():
+            raise ValueError(f"path is not a file: {path}")
+
+        if target.stat().st_size > self.MAX_FILE_SIZE:
+            raise ValueError("file is too large to read")
+
+        try:
+            content = target.read_text(encoding="utf-8")
+        except UnicodeDecodeError as error:
+            raise ValueError("file is not valid UTF-8 text") from error
+
+        return ToolResult(content=content)
+
+
 if __name__ == "__main__":
     registry = ToolRegistry()