Nell’agosto 2023, Google ha annunciato l’aggiunta di un servizio di qualità dell’aria al suo elenco di API di mappatura. Puoi leggere di più a riguardo Qui. Sembra che queste informazioni siano ora disponibili anche all’interno dell’app Google Maps, anche se i dati ottenibili tramite le API si sono rivelati molto più ricchi.
Secondo l’annuncio, Google sta combinando informazioni provenienti da molte fonti a diverse risoluzioni – sensori di inquinamento terrestri, dati satellitari, informazioni sul traffico in tempo reale e previsioni da modelli numerici – per produrre un set di dati aggiornato dinamicamente sulla qualità dell’aria in 100 paesi con una risoluzione fino a 500 m. . Sembra un set di dati molto interessante e potenzialmente utile per tutti i tipi di applicazioni di mappatura, assistenza sanitaria e pianificazione!
Quando ho letto per la prima volta di questo, avevo intenzione di provarlo in un’applicazione “parla con i tuoi dati”, utilizzando alcune delle cose apprese dalla creazione di questo mappatore di viaggio attrezzo. Forse un sistema in grado di tracciare una serie temporale delle concentrazioni di inquinamento atmosferico nella tua città preferita, o forse uno strumento per aiutare le persone a pianificare escursioni nella loro zona per evitare l’aria cattiva?
Ci sono tre strumenti API che può aiutare qui: un servizio di “condizioni attuali”, che fornisce i valori attuali dell’indice di qualità dell’aria e le concentrazioni di inquinanti in un determinato luogo; un servizio “condizioni storiche”, che fa la stessa cosa ma a intervalli orari per un massimo di 30 giorni nel passato e un servizio “mappa termica”, che fornisce le condizioni attuali su una determinata area come immagine.
In precedenza avevo usato l’eccellente googlemapspacchetto per chiamare le API di Google Maps in Python, ma queste nuove API non sono ancora supportate. Sorprendentemente, al di là della documentazione ufficiale ho potuto trovare pochi esempi di persone che utilizzano questi nuovi strumenti e nessun pacchetto Python preesistente progettato per richiamarli. Sarei felice di essere corretto se qualcuno sapesse il contrario!
Ho quindi creato alcuni strumenti rapidi per conto mio e in questo post spieghiamo come funzionano e come utilizzarli. Spero che questo possa essere utile a chiunque voglia sperimentare queste nuove API in Python e cerchi un punto da cui iniziare. È possibile trovare tutto il codice per questo progetto Quie probabilmente espanderò questo repository nel tempo man mano che aggiungerò più funzionalità e creerò una sorta di applicazione di mappatura con i dati sulla qualità dell’aria.
Iniziamo! In questa sezione esamineremo come recuperare i dati sulla qualità dell’aria in una determinata posizione con Google Maps. Avrai prima bisogno di una chiave API, che puoi generare tramite il tuo account Google Cloud. Loro hanno un Periodo di prova gratuito di 90 giornidopodiché pagherai per i servizi API che utilizzi. Assicurati di abilitare l ‘”Air Quality API” e di essere consapevole delle politiche sui prezzi prima di iniziare a fare molte chiamate!
Di solito memorizzo la mia chiave API in un file .env file e caricarlo con dotenv utilizzando una funzione come questa
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,
}
Per ottenere le condizioni attuali è necessaria una richiesta POST come dettagliato Qui. Prenderemo spunto da Google Maps package per farlo in un modo che possa essere generalizzato. Innanzitutto, creiamo una classe client che utilizza requests per effettuare la chiamata. L’obiettivo è abbastanza semplice: vogliamo creare un URL come quello seguente e includere tutte le opzioni di richiesta specifiche per la query dell’utente.
https://airquality.googleapis.com/v1/currentConditions:lookup?key=YOUR_API_KEY
IL Clientaccetta la nostra chiave API come key e poi costruisce il request_url per la domanda. Accetta le opzioni di richiesta come a params dizionario e quindi li inserisce nel corpo della richiesta JSON, che è gestito da self.session.post() chiamata.
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",
}
Ora possiamo creare una funzione che aiuti l’utente ad assemblare opzioni di richiesta valide per l’API delle condizioni correnti e quindi utilizzare questa classe Client per effettuare la richiesta. Ancora una volta, questo si ispira al design del pacchetto 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)
Le opzioni per questa API sono relativamente semplici. Ha bisogno di un dizionario con la longitudine e la latitudine del punto che vuoi indagare e, facoltativamente, può includere vari altri argomenti che controllano la quantità di informazioni restituite. Vediamolo in azione con tutti gli argomenti impostati su 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
)
Vengono restituite molte informazioni interessanti! Non solo disponiamo dei valori dell’indice di qualità dell’aria degli indici AQI universale e statunitense, ma abbiamo anche le concentrazioni dei principali inquinanti, una descrizione di ciascuno e una serie generale di raccomandazioni sanitarie per l’attuale qualità dell’aria.
{'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.'}}
Non sarebbe bello poter recuperare una serie temporale di questi valori AQI e inquinanti per una determinata località? Ciò potrebbe rivelare modelli interessanti come le correlazioni tra gli inquinanti o le fluttuazioni giornaliere causate dal traffico o da fattori meteorologici.
Possiamo farlo con un’altra richiesta POST al file condizioni storiche API, che ci darà una storia oraria. Funziona più o meno allo stesso modo delle condizioni attuali, l’unica differenza sostanziale è che, poiché i risultati possono essere piuttosto lunghi, vengono restituiti come diversi pages che richiede un po’ di logica in più per essere gestita.
Modifichiamo il request_post metodo di Client per gestire questa cosa.
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
Questo gestisce il caso in cui response_body contiene un campo chiamato nextPageTokenche è l’ID della successiva pagina di dati che è stata generata ed è pronta per essere recuperata. Laddove esistono tali informazioni, dobbiamo solo richiamare nuovamente l’API con un nuovo parametro chiamato pageToken che lo indirizza alla pagina pertinente. Lo facciamo ripetutamente in un ciclo while finché non rimangono più pagine. Nostro final_response dizionario quindi ora contiene un altro livello indicato dal numero di pagina. Per chiamate a current_conditions ci sarà sempre e solo una pagina, ma per le chiamate a historical_conditions potrebbero essercene diversi.
Fatto ciò, possiamo scrivere a historical_conditions funzionare in uno stile molto simile a 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)
Per definire il periodo storico, l’API può accettare a lag_time in ore, fino a 720 (30 giorni). Può anche accettare a specific_perioddizionario, con definisce gli orari di inizio e fine nel formato descritto nei commenti sopra. Infine, per recuperare una singola ora di dati, può accettare solo un timestamp, fornito da specific_time . Da notare anche l’uso del pageSize parametro, che controlla il numero di punti temporali restituiti in ogni chiamata all’API. Il valore predefinito qui è 100.
Proviamolo.
# 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
)
Dovremmo ottenere una risposta JSON lunga e nidificata che contenga i valori dell’indice AQI e i valori degli inquinanti specifici con incrementi di 1 ora nelle ultime 720 ore. Esistono molti modi per formattarlo in una struttura più adatta alla visualizzazione e all’analisi e la funzione seguente lo convertirà in un dataframe panda in formato “lungo”, che funziona bene con seabornper tracciare.
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
Eseguendo questo sull’output di historical_conditions produrrà un dataframe le cui colonne sono formattate per una facile analisi.
df = historical_conditions_to_df(history_conditions_data)
E ora possiamo tracciare il risultato seaborn o qualche altro strumento di visualizzazione.
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)
Già questo è molto interessante! Esistono chiaramente diverse periodicità nelle serie temporali degli inquinanti ed è da notare che l’AQI statunitense è strettamente correlato con le concentrazioni di pm25 e pm10, come previsto. Ho molta meno familiarità con l’AQI universale che Google fornisce qui, quindi non posso spiegare perché appaia anti-correlato con pm25 e p10. Un UAQI più piccolo significa una migliore qualità dell’aria? Nonostante qualche ricerca in giro non sono riuscito a trovare una buona risposta.
Veniamo ora al caso d’uso finale dell’API Qualità dell’aria di Google Maps: la generazione di riquadri della mappa di calore. IL documentazione su questo argomento ce ne sono pochi, il che è un peccato perché queste tessere sono un potente strumento per visualizzare la qualità dell’aria attuale, soprattutto se combinate con un Folium carta geografica.
Li recuperiamo con una richiesta GET, che implica la creazione di un URL nel seguente formato, dove la posizione del riquadro è specificata da zoom , x E y
GET https://airquality.googleapis.com/v1/mapTypes/{mapType}/heatmapTiles/{zoom}/{x}/{y}
Cosa farezoom , x E y Significare? Possiamo rispondere a questa domanda apprendendo come Google Maps converte le coordinate di latitudine e longitudine in “coordinate delle tessere”, che è descritto in dettaglio Qui. In sostanza, Google Maps memorizza le immagini in griglie in cui ciascuna cella misura 256 x 256 pixel e le dimensioni reali della cella sono una funzione del livello di zoom. Quando effettuiamo una chiamata all’API, dobbiamo specificare da quale griglia attingere, che è determinata dal livello di zoom, e da dove attingere sulla griglia, che è determinato dal x E y coordinate della tessera. Ciò che ritorna è un array di byte che può essere letto da Python Imaging Library (PIL) o da un pacchetto simile di elaborazione delle immagini.
Avendo formato il ns url nel formato sopra, possiamo aggiungere alcuni metodi al file Client classe che ci permetterà di recuperare l’immagine corrispondente.
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
Questo è positivo, ma quello di cui abbiamo veramente bisogno è la capacità di convertire un insieme di coordinate di longitudine e latitudine in coordinate di riquadro. La documentazione spiega come: prima convertiamo le coordinate nel file Proiezione di Mercatoreda cui convertiamo in “coordinate pixel” utilizzando il livello di zoom specificato. Alla fine lo traduciamo nelle coordinate della tessera. Per gestire tutte queste trasformazioni, possiamo usare il file TileHelper classe sottostante.
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))
Possiamo vederlo location_to_tile_xy sta prendendo un dizionario di posizione e un livello di zoom e restituisce il riquadro in cui è possibile trovare quel punto. Un’altra funzione utile è tile_to_bounding_box che troverà le coordinate di delimitazione di una cella della griglia specificata. Ne abbiamo bisogno se vogliamo geolocalizzare la cella e tracciarla su una mappa.
Vediamo come funziona all’interno del file air_quality_tile funzione di seguito, che prenderà in considerazione il nostro client , location e una stringa che indica il tipo di tessera che vogliamo recuperare. Dobbiamo anche specificare un livello di zoom, che all’inizio può essere difficile da scegliere e richiede alcuni tentativi ed errori. Discuteremo di get_adjoining_tiles argomento a 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
Dalla lettura del codice, possiamo vedere che il flusso di lavoro è il seguente: innanzitutto, trova le coordinate della tessera della posizione di interesse. Questo specifica la cella della griglia che vogliamo recuperare. Quindi, trova le coordinate di delimitazione di questa cella della griglia. Se vogliamo recuperare le tessere circostanti, troviamo l’angolo più vicino del riquadro di delimitazione e quindi usiamolo per calcolare le coordinate delle tessere delle tre celle della griglia adiacenti. Quindi chiama l’API e restituisci ciascuno dei riquadri come immagine con il riquadro di delimitazione corrispondente.
Possiamo eseguirlo nel modo standard, come segue:
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)
E poi traccia con folium per una mappa zoomabile! Tieni presente che qui sto utilizzando leafmap, perché questo pacchetto può generare mappe Folium compatibili con gradio, un potente strumento per generare semplici interfacce utente per applicazioni Python. Dare un’occhiata a Questo articolo per un esempio.
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)
Forse in modo deludente, il riquadro contenente la nostra posizione a questo livello di zoom è principalmente mare, anche se è comunque bello vedere l’inquinamento atmosferico tracciato sopra una mappa dettagliata. Se ingrandisci, puoi vedere che le informazioni sul traffico stradale vengono utilizzate per informare i segnali sulla qualità dell’aria nelle aree urbane.
Collocamento get_adjoining_tiles=True ci dà una mappa molto più bella perché recupera le tre tessere più vicine e non sovrapposte a quel livello di zoom. Nel nostro caso questo aiuta molto a rendere la mappa più presentabile.
Personalmente preferisco le immagini generate quando pollutant=US_AQIma ci sono diverse opzioni. Sfortunatamente l’API non restituisce una scala di colori, anche se sarebbe possibile generarne una utilizzando i valori dei pixel nell’immagine e conoscendo il significato dei colori.
Grazie per essere arrivato fino alla fine! Qui abbiamo esplorato come utilizzare le API della qualità dell’aria di Google Maps per fornire risultati in Python, che potrebbero essere utilizzati in applicazioni interessanti. In futuro spero di dare seguito a un altro articolo su air_quality_mapper tool man mano che si evolve ulteriormente, ma spero che gli script discussi qui siano utili di per sé. Come sempre, qualsiasi suggerimento per ulteriori sviluppi sarà molto apprezzato!
Fonte: towardsdatascience.com
