获取给定位置当前的空气质量
让我们开始吧!在本节中,我们将介绍如何使用 Google 地图获取给定位置的空气质量数据。您首先需要一个 API 密钥,可以通过您的 Google Cloud 帐户生成该密钥。他们有90 天的免费试用期,之后您将为您使用的 API 服务付费。在开始大量拨打电话之前,请确保启用“空气质量 API”,并了解定价政策!
Google Cloud API 库的屏幕截图,您可以在其中激活空气质量 API。图片由作者生成。
我通常将 API 密钥存储在一个文件中,并使用这样的函数.env
加载它dotenv
from dotenv import load_dotenv
from pathlib import Path
def load_secets ():
load_dotenv()
env_path = Path( "." ) / ".env"
load_dotenv(dotenv_path=env_path)
google_maps_key = os.getenv( "GOOGLE_MAPS_API_KEY" )
return {
"GOOGLE_MAPS_API_KEY" : google_maps_key,
}
获取当前条件需要 POST 请求,详细信息请参见此处。我们将从googlemaps包中获得灵感,以一种可以推广的方式来做到这一点。首先,我们构建一个用于requests
进行调用的客户端类。目标非常简单 - 我们希望构建一个如下所示的 URL,并包含特定于用户查询的所有请求选项。
https://airquality.googleapis.com/v1/currentConditions:lookup?key=YOUR_API_KEY
该类Client
接受我们的 API 密钥 as ,然后为查询key
构建。request_url
它接受请求选项作为params
字典,然后将它们放入 JSON 请求正文中,该self.session.post()
正文由调用处理。
import requests
import io
class Client(object):
DEFAULT_BASE_URL = "https://airquality.googleapis.com"
def __init__(self, key):
self.session = requests.Session()
self.key = key
def request_post(self, url, params):
request_url = self.compose_url(url)
request_header = self.compose_header()
request_body = params
response = self.session.post(
request_url,
headers=request_header,
json=request_body,
)
return self.get_body(response)
def compose_url(self, path):
return self.DEFAULT_BASE_URL + path + "?" + "key=" + self.key
@staticmethod
def get_body(response):
body = response.json()
if "error" in body:
return body["error"]
return body
@staticmethod
def compose_header():
return {
"Content-Type": "application/json",
}
现在我们可以创建一个函数来帮助用户为当前条件 API 组装有效的请求选项,然后使用此 Client 类来发出请求。同样,这也是受到 googlemaps 包设计的启发。
def current_conditions(
client,
location,
include_local_AQI=True,
include_health_suggestion=False,
include_all_pollutants=True,
include_additional_pollutant_info=False,
include_dominent_pollutant_conc=True,
language=None,
):
"""
See documentation for this API here
https://developers.google.com/maps/documentation/air-quality/reference/rest/v1/currentConditions/lookup
"""
params = {}
if isinstance(location, dict):
params["location"] = location
else:
raise ValueError(
"Location argument must be a dictionary containing latitude and longitude"
)
extra_computations = []
if include_local_AQI:
extra_computations.append("LOCAL_AQI")
if include_health_suggestion:
extra_computations.append("HEALTH_RECOMMENDATIONS")
if include_additional_pollutant_info:
extra_computations.append("POLLUTANT_ADDITIONAL_INFO")
if include_all_pollutants:
extra_computations.append("POLLUTANT_CONCENTRATION")
if include_dominent_pollutant_conc:
extra_computations.append("DOMINANT_POLLUTANT_CONCENTRATION")
if language:
params["language"] = language
params["extraComputations"] = extra_computations
return client.request_post("/v1/currentConditions:lookup", params)
该 API 的选项相对简单。它需要一个包含您想要调查的点的经度和纬度的字典,并且可以选择接受控制返回信息量的各种其他参数。让我们看看它的实际效果,所有参数都设置为True
# set up client
client = Client(key=GOOGLE_MAPS_API_KEY)
# a location in Los Angeles, CA
location = {"longitude":-118.3,"latitude":34.1}
# a JSON response
current_conditions_data = current_conditions(
client,
location,
include_health_suggestion=True,
include_additional_pollutant_info=True
)
返回了很多有趣的信息!我们不仅有来自通用和美国 AQI 指数的空气质量指数值,而且还有主要污染物的浓度、每种污染物的描述以及针对当前空气质量的一套整体健康建议。
{'dateTime': '2023-10-12T05:00:00Z',
'regionCode': 'us',
'indexes': [{'code': 'uaqi',
'displayName': 'Universal AQI',
'aqi': 60,
'aqiDisplay': '60',
'color': {'red': 0.75686276, 'green': 0.90588236, 'blue': 0.09803922},
'category': 'Good air quality',
'dominantPollutant': 'pm10'},
{'code': 'usa_epa',
'displayName': 'AQI (US)',
'aqi': 3