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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125 | class OpenWeatherMapToolSpec(BaseToolSpec):
"""打开天气工具规范。"""
spec_functions = ["weather_at_location", "forecast_tommorrow_at_location"]
def __init__(self, key: str, temp_units: str = "celsius") -> None:
"""使用参数进行初始化。"""
try:
from pyowm import OWM
except ImportError:
raise ImportError(
"The OpenWeatherMap tool requires the pyowm package to be installed. "
"Please install it using `pip install pyowm`."
)
self.key = key
self.temp_units = temp_units
self._owm = OWM(self.key)
self._mgr = self._owm.weather_manager()
def _format_current_temp(self, temperature: Any, temp_unit: str) -> str:
return (
f" - Current: {temperature['temp']}{temp_unit}\n"
f" - High: {temperature['temp_max']}{temp_unit}\n"
f" - Low: {temperature['temp_min']}{temp_unit}\n"
f" - Feels like: {temperature['feels_like']}{temp_unit}"
)
def _format_forecast_temp(self, temperature: Any, temp_unit: str) -> str:
return (
f" - High: {temperature['max']}{temp_unit}\n"
f" - Low: {temperature['min']}{temp_unit}"
)
def _format_weather(self, place: str, temp_str: str, w: Any) -> str:
"""格式化OpenWeatherMap的天气响应。
感谢langchain/utilities/openweathermap.py中的函数。
"""
detailed_status = w.detailed_status
wind = w.wind()
humidity = w.humidity
rain = w.rain
heat_index = w.heat_index
clouds = w.clouds
return (
f"In {place}, the current weather is as follows:\n"
f"Detailed status: {detailed_status}\n"
f"Wind speed: {wind['speed']} m/s, direction: {wind['deg']}°\n"
f"Humidity: {humidity}%\n"
"Temperature: \n"
f"{temp_str}\n"
f"Rain: {rain}\n"
f"Heat index: {heat_index!s}\n"
f"Cloud cover: {clouds}%"
)
def weather_at_location(self, location: str) -> List[Document]:
"""找到指定位置的当前天气。
Args:
place (str):
要查询天气的地点。
应为城市名和国家。
"""
from pyowm.commons.exceptions import NotFoundError
try:
observation = self._mgr.weather_at_place(location)
except NotFoundError:
return [Document(text=f"Unable to find weather at {location}.")]
w = observation.weather
temperature = w.temperature(self.temp_units)
temp_unit = "°C" if self.temp_units == "celsius" else "°F"
temp_str = self._format_current_temp(temperature, temp_unit)
weather_text = self._format_weather(location, temp_str, w)
return [Document(text=weather_text, metadata={"weather from": location})]
def forecast_tommorrow_at_location(self, location: str) -> List[Document]:
"""找到明天某地的天气预报。
Args:
location (str):
要查找明天天气的地点。
应该是城市名和国家。
"""
from pyowm.commons.exceptions import NotFoundError
from pyowm.utils import timestamps
try:
forecast = self._mgr.forecast_at_place(location, "daily")
except NotFoundError:
return [Document(text=f"Unable to find weather at {location}.")]
tomorrow = timestamps.tomorrow()
w = forecast.get_weather_at(tomorrow)
temperature = w.temperature(self.temp_units)
temp_unit = "°C" if self.temp_units == "celsius" else "°F"
temp_str = self._format_forecast_temp(temperature, temp_unit)
weather_text = self._format_weather(location, temp_str, w)
return [
Document(
text=weather_text,
metadata={
"weather from": location,
"forecast for": tomorrow.strftime("%Y-%m-%d"),
},
)
]
|