En agosto de 2023, Google anunció la incorporación de un servicio de calidad del aire a su lista de API de mapas. Puedes leer más sobre eso. aquí. Parece que esta información ahora también está disponible desde la aplicación Google Maps, aunque los datos que se pueden obtener a través de las API resultaron ser mucho más completos.
Según el anuncio, Google está combinando información de muchas fuentes en diferentes resoluciones (sensores de contaminación terrestres, datos satelitales, información de tráfico en vivo y predicciones de modelos numéricos) para producir un conjunto de datos actualizado dinámicamente sobre la calidad del aire en 100 países con una resolución de hasta 500 m. . ¡Esto suena como un conjunto de datos muy interesante y potencialmente útil para todo tipo de aplicaciones de mapeo, atención médica y planificación!
Cuando leí esto por primera vez, estaba planeando probarlo en una aplicación de “habla con tus datos”, usando algunas de las cosas aprendidas al crear esto. mapeador de viajes herramienta. ¿Quizás un sistema que pueda trazar una serie temporal de concentraciones de contaminación del aire en su ciudad favorita, o quizás una herramienta para ayudar a las personas a planificar caminatas en su área local para evitar el mal aire?
Hay tres herramientas API eso puede ayudar aquí: un servicio de “condiciones actuales”, que proporciona valores actuales del índice de calidad del aire y concentraciones de contaminantes en un lugar determinado; un servicio de “condiciones históricas”, que hace lo mismo pero en intervalos de una hora durante hasta 30 días en el pasado y un servicio de “mapa de calor”, que proporciona las condiciones actuales en un área determinada como una imagen.
Anteriormente había usado el excelente googlemapspaquete para llamar a las API de Google Maps en Python, pero estas nuevas API aún no son compatibles. Sorprendentemente, más allá de la documentación oficial pude encontrar pocos ejemplos de personas que utilizan estas nuevas herramientas y ningún paquete de Python preexistente diseñado para llamarlas. ¡Sin embargo, me corregirían felizmente si alguien supiera lo contrario!
Por lo tanto, creé algunas herramientas rápidas por mi cuenta y en esta publicación explicamos cómo funcionan y cómo usarlas. Espero que esto sea útil para cualquiera que quiera experimentar con estas nuevas API en Python y busque un lugar por donde comenzar. Todo el código para este proyecto se puede encontrar. aquíy probablemente ampliaré este repositorio con el tiempo a medida que agregue más funciones y cree algún tipo de aplicación de mapeo con los datos de calidad del aire.
¡Empecemos! En esta sección veremos cómo obtener datos sobre la calidad del aire en una ubicación determinada con Google Maps. Primero necesitará una clave API, que puede generar a través de su cuenta de Google Cloud. Ellos tienen un Período de prueba gratuito de 90 días, después de lo cual pagará por los servicios API que utilice. ¡Asegúrese de habilitar la “API de calidad del aire” y tenga en cuenta las políticas de precios antes de comenzar a hacer muchas llamadas!
Normalmente guardo mi clave API en un .env archivo y cargarlo con dotenv usando una función como esta
from dotenv import load_dotenv
from pathlib import Pathdef 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,
}
Obtener las condiciones actuales requiere una solicitud POST como se detalla aquí. Nos inspiraremos en el mapas de Google paquete para hacer esto de una manera que pueda generalizarse. Primero, construimos una clase de cliente que usa requests para hacer la llamada. El objetivo es bastante sencillo: queremos crear una URL como la siguiente e incluir todas las opciones de solicitud específicas para la consulta del usuario.
https://airquality.googleapis.com/v1/currentConditions:lookup?key=YOUR_API_KEY
El ClientLa clase toma nuestra clave API como key y luego construye el request_url para la consulta. Acepta opciones de solicitud como params diccionario y luego los coloca en el cuerpo de la solicitud JSON, que es manejado por el self.session.post() llamar.
import requests
import ioclass 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",
}
Ahora podemos crear una función que ayude al usuario a reunir opciones de solicitud válidas para la API de condiciones actuales y luego usar esta clase de Cliente para realizar la solicitud. Nuevamente, esto está inspirado en el diseño del paquete 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)
Las opciones para esta API son relativamente sencillas. Necesita un diccionario con la longitud y latitud del punto que desea investigar y, opcionalmente, puede aceptar varios otros argumentos que controlan la cantidad de información que se devuelve. Veámoslo en acción con todos los argumentos establecidos en 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
)
¡Se devuelve mucha información interesante! No solo tenemos los valores del índice de calidad del aire de los índices AQI universal y estadounidense, sino que también tenemos las concentraciones de los principales contaminantes, una descripción de cada uno y un conjunto general de recomendaciones de salud para la calidad del aire actual.
{'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': 39,
'aqiDisplay': '39',
'color': {'green': 0.89411765},
'category': 'Good air quality',
'dominantPollutant': 'pm10'}),
'pollutants': ({'code': 'co',
'displayName': 'CO',
'fullName': 'Carbon monoxide',
'concentration': {'value': 292.61, 'units': 'PARTS_PER_BILLION'},
'additionalInfo': {'sources': 'Typically originates from incomplete combustion of carbon fuels, such as that which occurs in car engines and power plants.',
'effects': 'When inhaled, carbon monoxide can prevent the blood from carrying oxygen. Exposure may cause dizziness, nausea and headaches. Exposure to extreme concentrations can lead to loss of consciousness.'}},
{'code': 'no2',
'displayName': 'NO2',
'fullName': 'Nitrogen dioxide',
'concentration': {'value': 22.3, 'units': 'PARTS_PER_BILLION'},
'additionalInfo': {'sources': 'Main sources are fuel burning processes, such as those used in industry and transportation.',
'effects': 'Exposure may cause increased bronchial reactivity in patients with asthma, lung function decline in patients with Chronic Obstructive Pulmonary Disease (COPD), and increased risk of respiratory infections, especially in young children.'}},
{'code': 'o3',
'displayName': 'O3',
'fullName': 'Ozone',
'concentration': {'value': 24.17, 'units': 'PARTS_PER_BILLION'},
'additionalInfo': {'sources': 'Ozone is created in a chemical reaction between atmospheric oxygen, nitrogen oxides, carbon monoxide and organic compounds, in the presence of sunlight.',
'effects': 'Ozone can irritate the airways and cause coughing, a burning sensation, wheezing and shortness of breath. Additionally, ozone is one of the major components of photochemical smog.'}},
{'code': 'pm10',
'displayName': 'PM10',
'fullName': 'Inhalable particulate matter (<10µm)',
'concentration': {'value': 44.48, 'units': 'MICROGRAMS_PER_CUBIC_METER'},
'additionalInfo': {'sources': 'Main sources are combustion processes (e.g. indoor heating, wildfires), mechanical processes (e.g. construction, mineral dust, agriculture) and biological particles (e.g. pollen, bacteria, mold).',
'effects': 'Inhalable particles can penetrate into the lungs. Short term exposure can cause irritation of the airways, coughing, and aggravation of heart and lung diseases, expressed as difficulty breathing, heart attacks and even premature death.'}},
{'code': 'pm25',
'displayName': 'PM2.5',
'fullName': 'Fine particulate matter (<2.5µm)',
'concentration': {'value': 11.38, 'units': 'MICROGRAMS_PER_CUBIC_METER'},
'additionalInfo': {'sources': 'Main sources are combustion processes (e.g. power plants, indoor heating, car exhausts, wildfires), mechanical processes (e.g. construction, mineral dust) and biological particles (e.g. bacteria, viruses).',
'effects': 'Fine particles can penetrate into the lungs and bloodstream. Short term exposure can cause irritation of the airways, coughing and aggravation of heart and lung diseases, expressed as difficulty breathing, heart attacks and even premature death.'}},
{'code': 'so2',
'displayName': 'SO2',
'fullName': 'Sulfur dioxide',
'concentration': {'value': 0, 'units': 'PARTS_PER_BILLION'},
'additionalInfo': {'sources': 'Main sources are burning processes of sulfur-containing fuel in industry, transportation and power plants.',
'effects': 'Exposure causes irritation of the respiratory tract, coughing and generates local inflammatory reactions. These in turn, may cause aggravation of lung diseases, even with short term exposure.'}}),
'healthRecommendations': {'generalPopulation': 'With this level of air quality, you have no limitations. Enjoy the outdoors!',
'elderly': 'If you start to feel respiratory discomfort such as coughing or breathing difficulties, consider reducing the intensity of your outdoor activities. Try to limit the time you spend near busy roads, construction sites, open fires and other sources of smoke.',
'lungDiseasePopulation': 'If you start to feel respiratory discomfort such as coughing or breathing difficulties, consider reducing the intensity of your outdoor activities. Try to limit the time you spend near busy roads, industrial emission stacks, open fires and other sources of smoke.',
'heartDiseasePopulation': 'If you start to feel respiratory discomfort such as coughing or breathing difficulties, consider reducing the intensity of your outdoor activities. Try to limit the time you spend near busy roads, construction sites, industrial emission stacks, open fires and other sources of smoke.',
'athletes': 'If you start to feel respiratory discomfort such as coughing or breathing difficulties, consider reducing the intensity of your outdoor activities. Try to limit the time you spend near busy roads, construction sites, industrial emission stacks, open fires and other sources of smoke.',
'pregnantWomen': 'To keep you and your baby healthy, consider reducing the intensity of your outdoor activities. Try to limit the time you spend near busy roads, construction sites, open fires and other sources of smoke.',
'children': 'If you start to feel respiratory discomfort such as coughing or breathing difficulties, consider reducing the intensity of your outdoor activities. Try to limit the time you spend near busy roads, construction sites, open fires and other sources of smoke.'}}
¿No sería bueno poder obtener una serie temporal de estos valores de ICA y de contaminantes para una ubicación determinada? Esto podría revelar patrones interesantes, como correlaciones entre los contaminantes o fluctuaciones diarias causadas por el tráfico o factores relacionados con el clima.
Podemos hacer esto con otra solicitud POST al condiciones históricas API, que nos dará un historial horario. Esto funciona de manera muy similar a las condiciones actuales, la única diferencia importante es que, dado que los resultados pueden ser bastante largos, se devuelven como varios pages que requiere un poco de lógica adicional para manejarlo.
Modifiquemos el request_post método de Client para manejar esto.
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,
)
response_body = self.get_body(response)
# put the first page in the response dictionary
page = 1
final_response = {
"page_{}".format(page) : response_body
}
# fetch all the pages if needed
while "nextPageToken" in response_body:
# call again with the next page's token
request_body.update({
"pageToken":response_body("nextPageToken")
})
response = self.session.post(
request_url,
headers=request_header,
json=request_body,
)
response_body = self.get_body(response)
page += 1
final_response("page_{}".format(page)) = response_body
return final_response
Esto maneja el caso donde response_body contiene un campo llamado nextPageToken, que es la identificación de la siguiente página de datos que se generó y está lista para recuperarse. Cuando exista esa información, solo necesitamos llamar a la API nuevamente con un nuevo parámetro llamado pageToken , que lo dirige a la página correspondiente. Hacemos esto repetidamente en un bucle while hasta que no queden más páginas. Nuestro final_response Por lo tanto, el diccionario ahora contiene otra capa indicada por el número de página. Para llamadas a current_conditions solo habrá una página, pero para llamadas a historical_conditions puede haber varios.
Una vez solucionado esto, podemos escribir un historical_conditions funcionar en un estilo muy similar al current_conditions .
def historical_conditions(
client,
location,
specific_time=None,
lag_time=None,
specific_period=None,
include_local_AQI=True,
include_health_suggestion=False,
include_all_pollutants=True,
include_additional_pollutant_info=False,
include_dominant_pollutant_conc=True,
language=None,
):
"""
See documentation for this API here https://developers.google.com/maps/documentation/air-quality/reference/rest/v1/history/lookup
"""
params = {}if isinstance(location, dict):
params("location") = location
else:
raise ValueError(
"Location argument must be a dictionary containing latitude and longitude"
)
if isinstance(specific_period, dict) and not specific_time and not lag_time:
assert "startTime" in specific_period
assert "endTime" in specific_period
params("period") = specific_period
elif specific_time and not lag_time and not isinstance(specific_period, dict):
# note that time must be in the "Zulu" format
# e.g. datetime.datetime.strftime(datetime.datetime.now(),"%Y-%m-%dT%H:%M:%SZ")
params("dateTime") = specific_time
# lag periods in hours
elif lag_time and not specific_time and not isinstance(specific_period, dict):
params("hours") = lag_time
else:
raise ValueError(
"Must provide specific_time, specific_period or lag_time arguments"
)
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_dominant_pollutant_conc:
extra_computations.append("DOMINANT_POLLUTANT_CONCENTRATION")
if language:
params("language") = language
params("extraComputations") = extra_computations
# page size default set to 100 here
params("pageSize") = 100
# page token will get filled in if needed by the request_post method
params("pageToken") = ""
return client.request_post("/v1/history:lookup", params)
Para definir el período histórico, la API puede aceptar un lag_time en horas, hasta 720 (30 días). También puede aceptar un specific_perioddiccionario, que define las horas de inicio y finalización en el formato descrito en los comentarios anteriores. Finalmente, para recuperar una sola hora de datos, puede aceptar solo una marca de tiempo, proporcionada por specific_time . Tenga en cuenta también el uso de la pageSize parámetro, que controla cuántos puntos de tiempo se devuelven en cada llamada a la API. El valor predeterminado aquí es 100.
Probémoslo.
# 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
history_conditions_data = historical_conditions(
client,
location,
lag_time=720
)
Deberíamos obtener una respuesta JSON larga y anidada que contenga los valores del índice AQI y los valores de contaminantes específicos en incrementos de 1 hora durante las últimas 720 horas. Hay muchas formas de formatear esto en una estructura que sea más fácil de visualizar y analizar, y la siguiente función lo convertirá en un marco de datos de pandas en formato “largo”, que funciona bien con seabornpara trazar.
from itertools import chain
import pandas as pddef historical_conditions_to_df(response_dict):
chained_pages = list(chain(*(response_dict(p)("hoursInfo") for p in (*response_dict))))
all_indexes = ()
all_pollutants = ()
for i in range(len(chained_pages)):
# need this check in case one of the timestamps is missing data, which can sometimes happen
if "indexes" in chained_pages(i):
this_element = chained_pages(i)
# fetch the time
time = this_element("dateTime")
# fetch all the index values and add metadata
all_indexes += ((time , x("code"),x("displayName"),"index",x("aqi"),None) for x in this_element('indexes'))
# fetch all the pollutant values and add metadata
all_pollutants += ((time , x("code"),x("fullName"),"pollutant",x("concentration")("value"),x("concentration")("units")) for x in this_element('pollutants'))
all_results = all_indexes + all_pollutants
# generate "long format" dataframe
res = pd.DataFrame(all_results,columns=("time","code","name","type","value","unit"))
res("time")=pd.to_datetime(res("time"))
return res
Ejecutando esto en la salida de historical_conditions producirá un marco de datos cuyas columnas están formateadas para facilitar el análisis.
df = historical_conditions_to_df(history_conditions_data)
Y ahora podemos trazar el resultado en seaborn o alguna otra herramienta de visualización.
import seaborn as sns
g = sns.relplot(
x="time",
y="value",
data=df(df("code").isin(("uaqi","usa_epa","pm25","pm10"))),
kind="line",
col="name",
col_wrap=4,
hue="type",
height=4,
facet_kws={'sharey': False, 'sharex': False}
)
g.set_xticklabels(rotation=90)
¡Esto ya es muy interesante! Claramente hay varias periodicidades en las series temporales de contaminantes y es notable que el ICA de EE. UU. esté estrechamente correlacionado con las concentraciones de pm25 y pm10, como se esperaba. Estoy mucho menos familiarizado con el AQI universal que Google proporciona aquí, por lo que no puedo explicar por qué aparece anticorrelacionado con pm25 y p10. ¿Un UAQI más pequeño significa una mejor calidad del aire? A pesar de algunas búsquedas, no he podido encontrar una buena respuesta.
Ahora, veamos el caso de uso final de la API de calidad del aire de Google Maps: generar mosaicos de mapas de calor. El documentación sobre esto es escaso, lo cual es una pena porque estos mosaicos son una herramienta poderosa para visualizar la calidad del aire actual, especialmente cuando se combinan con un Folium mapa.
Los recuperamos con una solicitud GET, que implica crear una URL en el siguiente formato, donde la ubicación del mosaico se especifica mediante zoom , x y y
GET https://airquality.googleapis.com/v1/mapTypes/{mapType}/heatmapTiles/{zoom}/{x}/{y}
Qué hacerzoom , x y y ¿significar? Podemos responder a esto aprendiendo cómo Google Maps convierte las coordenadas de latitud y longitud en “coordenadas en mosaico”, que se describe en detalle. aquí. Básicamente, Google Maps almacena imágenes en cuadrículas donde cada celda mide 256 x 256 píxeles y las dimensiones reales de la celda son una función del nivel de zoom. Cuando hacemos una llamada a la API, necesitamos especificar desde qué cuadrícula dibujar (que está determinado por el nivel de zoom) y en qué parte de la cuadrícula dibujar (que está determinado por el nivel de zoom). x y y coordenadas de mosaico. Lo que devuelve es una matriz de bytes que puede ser leída por Python Imaging Library (PIL) o un paquete de procesamiento de imágenes similar.
Habiendo formado nuestro url En el formato anterior, podemos agregar algunos métodos al Client clase que nos permitirá recuperar la imagen correspondiente.
def request_get(self,url):request_url = self.compose_url(url)
response = self.session.get(request_url)
# for images coming from the heatmap tiles service
return self.get_image(response)
@staticmethod
def get_image(response):
if response.status_code == 200:
image_content = response.content
# note use of Image from PIL here
# needs from PIL import Image
image = Image.open(io.BytesIO(image_content))
return image
else:
print("GET request for image returned an error")
return None
Esto es bueno, pero lo que realmente necesitamos es la capacidad de convertir un conjunto de coordenadas de longitud y latitud en coordenadas de mosaico. La documentación explica cómo: primero convertimos a coordenadas en el Proyección de Mercator, desde el cual convertimos a “coordenadas de píxeles” usando el nivel de zoom especificado. Finalmente lo traducimos a las coordenadas del mosaico. Para manejar todas estas transformaciones, podemos usar el TileHelper clase a continuación.
import math
import numpy as npclass TileHelper(object):
def __init__(self, tile_size=256):
self.tile_size = tile_size
def location_to_tile_xy(self,location,zoom_level=4):
# Based on function here
# https://developers.google.com/maps/documentation/javascript/examples/map-coordinates#maps_map_coordinates-javascript
lat = location("latitude")
lon = location("longitude")
world_coordinate = self._project(lat,lon)
scale = 1 << zoom_level
pixel_coord = (math.floor(world_coordinate(0)*scale), math.floor(world_coordinate(1)*scale))
tile_coord = (math.floor(world_coordinate(0)*scale/self.tile_size),math.floor(world_coordinate(1)*scale/self.tile_size))
return world_coordinate, pixel_coord, tile_coord
def tile_to_bounding_box(self,tx,ty,zoom_level):
# see https://developers.google.com/maps/documentation/javascript/coordinates
# for details
box_north = self._tiletolat(ty,zoom_level)
# tile numbers advance towards the south
box_south = self._tiletolat(ty+1,zoom_level)
box_west = self._tiletolon(tx,zoom_level)
# time numbers advance towards the east
box_east = self._tiletolon(tx+1,zoom_level)
# (latmin, latmax, lonmin, lonmax)
return (box_south, box_north, box_west, box_east)
@staticmethod
def _tiletolon(x,zoom):
return x / math.pow(2.0,zoom) * 360.0 - 180.0
@staticmethod
def _tiletolat(y,zoom):
n = math.pi - (2.0 * math.pi * y)/math.pow(2.0,zoom)
return math.atan(math.sinh(n))*(180.0/math.pi)
def _project(self,lat,lon):
siny = math.sin(lat*math.pi/180.0)
siny = min(max(siny,-0.9999), 0.9999)
return (self.tile_size*(0.5 + lon/360), self.tile_size*(0.5 - math.log((1 + siny) / (1 - siny)) / (4 * math.pi)))
@staticmethod
def find_nearest_corner(location,bounds):
corner_lat_idx = np.argmin((
np.abs(bounds(0)-location("latitude")),
np.abs(bounds(1)-location("latitude"))
))
corner_lon_idx = np.argmin((
np.abs(bounds(2)-location("longitude")),
np.abs(bounds(3)-location("longitude"))
))
if (corner_lat_idx == 0) and (corner_lon_idx == 0):
# closests is latmin, lonmin
direction = "southwest"
elif (corner_lat_idx == 0) and (corner_lon_idx == 1):
direction = "southeast"
elif (corner_lat_idx == 1) and (corner_lon_idx == 0):
direction = "northwest"
else:
direction = "northeast"
corner_coords = (bounds(corner_lat_idx),bounds(corner_lon_idx+2))
return corner_coords, direction
@staticmethod
def get_ajoining_tiles(tx,ty,direction):
if direction == "southwest":
return ((tx-1,ty),(tx-1,ty+1),(tx,ty+1))
elif direction == "southeast":
return ((tx+1,ty),(tx+1,ty-1),(tx,ty-1))
elif direction == "northwest":
return ((tx-1,ty-1),(tx-1,ty),(tx,ty-1))
else:
return ((tx+1,ty-1),(tx+1,ty),(tx,ty-1))
Podemos ver eso location_to_tile_xy está tomando un diccionario de ubicación y un nivel de zoom y devolviendo el mosaico en el que se puede encontrar ese punto. Otra función útil es tile_to_bounding_box , que encontrará las coordenadas delimitadoras de una celda de cuadrícula especificada. Necesitamos esto si vamos a geolocalizar la celda y trazarla en un mapa.
Veamos cómo funciona esto dentro del air_quality_tile función a continuación, que incluirá nuestro client , location y una cadena que indica qué tipo de mosaico queremos recuperar. También debemos especificar un nivel de zoom, que puede resultar difícil de elegir al principio y requiere algo de prueba y error. Discutiremos el get_adjoining_tiles argumento en breve.
def air_quality_tile(
client,
location,
pollutant="UAQI_INDIGO_PERSIAN",
zoom=4,
get_adjoining_tiles = True):
# see https://developers.google.com/maps/documentation/air-quality/reference/rest/v1/mapTypes.heatmapTiles/lookupHeatmapTile
assert pollutant in (
"UAQI_INDIGO_PERSIAN",
"UAQI_RED_GREEN",
"PM25_INDIGO_PERSIAN",
"GBR_DEFRA",
"DEU_UBA",
"CAN_EC",
"FRA_ATMO",
"US_AQI"
)
# contains useful methods for dealing the tile coordinates
helper = TileHelper()
# get the tile that the location is in
world_coordinate, pixel_coord, tile_coord = helper.location_to_tile_xy(location,zoom_level=zoom)
# get the bounding box of the tile
bounding_box = helper.tile_to_bounding_box(tx=tile_coord(0),ty=tile_coord(1),zoom_level=zoom)
if get_adjoining_tiles:
nearest_corner, nearest_corner_direction = helper.find_nearest_corner(location, bounding_box)
adjoining_tiles = helper.get_ajoining_tiles(tile_coord(0),tile_coord(1),nearest_corner_direction)
else:
adjoining_tiles = ()
tiles = ()
#get all the adjoining tiles, plus the one in question
for tile in adjoining_tiles + (tile_coord):
bounding_box = helper.tile_to_bounding_box(tx=tile(0),ty=tile(1),zoom_level=zoom)
image_response = client.request_get(
"/v1/mapTypes/" + pollutant + "/heatmapTiles/" + str(zoom) + '/' + str(tile(0)) + '/' + str(tile(1))
)
# convert the PIL image to numpy
try:
image_response = np.array(image_response)
except:
image_response = None
tiles.append({
"bounds":bounding_box,
"image":image_response
})
return tiles
Al leer el código, podemos ver que el flujo de trabajo es el siguiente: Primero, busque las coordenadas del mosaico de la ubicación de interés. Esto especifica la celda de la cuadrícula que queremos recuperar. Luego, encuentre las coordenadas delimitadoras de esta celda de la cuadrícula. Si queremos recuperar los mosaicos circundantes, busque la esquina más cercana del cuadro delimitador y luego utilícela para calcular las coordenadas de los mosaicos de las tres celdas de la cuadrícula adyacentes. Luego llame a la API y devuelva cada uno de los mosaicos como una imagen con su cuadro delimitador correspondiente.
Podemos ejecutar esto de la forma estándar, de la siguiente manera:
client = Client(key=GOOGLE_MAPS_API_KEY)
location = {"longitude":-118.3,"latitude":34.1}
zoom = 7
tiles = air_quality_tile(
client,
location,
pollutant="UAQI_INDIGO_PERSIAN",
zoom=zoom,
get_adjoining_tiles=False)
¡Y luego trace con folio un mapa ampliable! Tenga en cuenta que estoy usando leafmap aquí, porque este paquete puede generar mapas Folium que son compatibles con gradio, una poderosa herramienta para generar interfaces de usuario simples para aplicaciones Python. Echa un vistazo a Este artículo para un ejemplo.
import leafmap.foliumap as leafmap
import foliumlat = location("latitude")
lon = location("longitude")
map = leafmap.Map(location=(lat, lon), tiles="OpenStreetMap", zoom_start=zoom)
for tile in tiles:
latmin, latmax, lonmin, lonmax = tile("bounds")
AQ_image = tile("image")
folium.raster_layers.ImageOverlay(
image=AQ_image,
bounds=((latmin, lonmin), (latmax, lonmax)),
opacity=0.7
).add_to(map)
Quizás de manera decepcionante, el mosaico que contiene nuestra ubicación en este nivel de zoom es principalmente mar, aunque aún es agradable ver la contaminación del aire trazada en la parte superior de un mapa detallado. Si hace zoom, puede ver que la información del tráfico rodado se utiliza para informar las señales de calidad del aire en las zonas urbanas.
Configuración get_adjoining_tiles=True nos brinda un mapa mucho mejor porque busca los tres mosaicos más cercanos y que no se superponen en ese nivel de zoom. En nuestro caso eso ayuda mucho a hacer el mapa más presentable.
Personalmente prefiero las imágenes generadas cuando pollutant=US_AQI, pero hay varias opciones diferentes. Lamentablemente, la API no devuelve una escala de colores, aunque sería posible generar una utilizando los valores de píxeles de la imagen y el conocimiento de lo que significan los colores.
¡Gracias por llegar hasta el final! Aquí exploramos cómo utilizar las API de calidad del aire de Google Maps para ofrecer resultados en Python, que podrían usarse en aplicaciones interesantes. En el futuro espero seguir con otro artículo sobre el mapeador_de_calidad_del_aire herramienta a medida que evoluciona, pero espero que los scripts discutidos aquí sean útiles por derecho propio. Como siempre, cualquier sugerencia para un mayor desarrollo será muy apreciada.
