Uno strumento Python per recuperare dati sulla qualità dell’aria dalle API di Google Maps Air Quality

Uno strumento Python per ottenere dati sulla qualità dell'aria dalle API di Google Maps Air Quality

Scopri come recuperare dati sulla qualità dell’aria in tempo reale da tutto il mondo

Questo articolo spiega come utilizzare le API Google Maps Air Quality in Python per recuperare ed esplorare dati sulla qualità dell’aria in tempo reale, serie temporali e mappe. Dai un’occhiata al codice completo qui.

1. Contesto

Nel agosto 2023, Google ha annunciato l’aggiunta di un servizio di qualità dell’aria alla sua lista di API per la mappatura. Puoi leggere di più su questo 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 combina informazioni provenienti da molte fonti a diverse risoluzioni: sensori di inquinamento a terra, dati satellitari, informazioni sul traffico in tempo reale e previsioni da modelli numerici, per creare un insieme di dati sulla qualità dell’aria aggiornato dinamicamente in 100 paesi fino a una risoluzione di 500 m. Questo suona come un insieme di dati molto interessante e potenzialmente utile per tutte le applicazioni di mappatura, sanità e pianificazione!

Quando ho letto per la prima volta di questo, stavo pensando di provarlo in un’applicazione “parla con i tuoi dati”, utilizzando alcune delle cose apprese dalla costruzione di questo strumento di mapping dei viaggi. Forse un sistema che può tracciare una serie temporale delle concentrazioni di inquinamento atmosferico nella tua città preferita, o forse un tool per aiutare le persone a pianificare escursioni nella loro zona locale per evitare l’aria malsana?

• Identificazione dei punti caldi in aree urbane
• Emissioni di carbonio di una squadra di ingegneria ML
• Considerazioni pratiche nella progettazione dell’applicazione RAG

Ci sono tre strumenti delle API che possono aiutare qui: un servizio “condizioni attuali”, che fornisce valori dell’indice di qualità dell’aria attuale e concentrazioni di inquinanti in una determinata posizione; un servizio “condizioni storiche”, che fa lo stesso ma a intervalli di un’ora fino a 30 giorni nel passato, e un servizio “heatmap”, che fornisce condizioni attuali su un’area data come un’immagine.

In precedenza, avevo utilizzato l’eccellente pacchetto googlemaps per chiamare le API di Google Maps in Python, ma queste nuove API non sono ancora supportate. Sorprendentemente, oltre alla documentazione ufficiale, ho trovato pochi esempi di persone che utilizzano questi nuovi strumenti e nessun pacchetto Python preesistente progettato per chiamarli. Sarei felicemente corretto se qualcuno ne sa di diverso!

Ho quindi costruito alcuni strumenti rapidi di mia proprietà, e in questo post mostro come funzionano e come usarli. Spero che questo sia utile a chiunque voglia sperimentare con queste nuove API in Python e cerchi un punto di partenza. Tutto il codice per questo progetto può essere trovato qui, e probabilmente amplierò questo repository nel tempo man mano che aggiungo più funzionalità e costruisco qualche tipo di applicazione di mappatura con i dati sulla qualità dell’aria.

2. Ottieni la qualità dell’aria attuale in una determinata posizione

Cominciamo! In questa sezione spiegheremo come recuperare dati sulla qualità dell’aria in una determinata posizione con Google Maps. Per prima cosa avrai bisogno di una chiave API, che puoi generare tramite il tuo account Google Cloud. Hanno un periodo di prova gratuita di 90 giorni, dopodiché pagherai per i servizi API che utilizzi. Assicurati di abilitare l’API “Qualità dell’aria” e di essere consapevole delle politiche di prezzo prima di iniziare a effettuare molte chiamate!

Di solito memorizzo la mia chiave API in un file .env e lo carico con dotenv usando una funzione come questa

from dotenv import load_dotenvfrom 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 ispirazione dal pacchetto googlemaps per farlo in modo che possa essere generalizzato. Innanzitutto, creiamo una classe client che utilizzi requests per effettuare la chiamata. L’obiettivo è piuttosto semplice – vogliamo costruire un URL come quello riportato di seguito e includere tutte le opzioni di richiesta specifiche della query dell’utente.

https://airquality.googleapis.com/v1/currentConditions:lookup?key=YOUR_API_KEY

La classe Client prende la nostra chiave API come key e quindi costruisce l’request_url per la query. Accetta le opzioni di richiesta come un dizionario params e le inserisce nel corpo della richiesta JSON, che viene gestito dalla chiamata self.session.post().

import requestsimport 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 aiuta l’utente a assemblare opzioni di richiesta valide per l’API delle condizioni attuali e quindi utilizzare questa classe client per effettuare la richiesta. Anche questo è ispirato 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,):    """    Vedere la documentazione per questa API qui    https://developers.google.com/maps/documentation/air-quality/reference/rest/v1/currentConditions/lookup    """    params = {}    if isinstance(location, dict):        params["location"] = location    else:        raise ValueError(            "L'argomento location deve essere un dizionario contenente la latitudine e la longitudine"        )    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 abbastanza semplici. È necessario un dizionario con la longitudine e la latitudine del punto che si desidera investigare e facoltativamente può accettare vari altri argomenti che controllano quante informazioni vengono restituite. Vediamolo in azione con tutti gli argomenti impostati su True

# Imposta il clientclient = Client(key=GOOGLE_MAPS_API_KEY)# Una posizione a Los Angeles, CAlocation = {"longitude":-118.3,"latitude":34.1}# Una risposta JSONcurrent_conditions_data = current_conditions(  client,  location,  include_health_suggestion=True,  include_additional_pollutant_info=True)

Vengono restituite molte informazioni interessanti! Non solo abbiamo gli indici della qualità dell’aria dai diversi indici AQI universali e basati negli Stati Uniti, ma abbiamo anche le concentrazioni dei principali inquinanti, una descrizione di ognuno di essi e un insieme generale di raccomandazioni per la salute per la qualità dell’aria attuale.

{'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': 'Buona qualità dell'aria',   'dominantPollutant': 'pm10'},  {'code': 'usa_epa',   'displayName': 'AQI (US)',   'aqi': 39,   'aqiDisplay': '39',   'color': {'green': 0.89411765},   'category': 'Buona qualità dell'aria',   'dominantPollutant': 'pm10'}], 'pollutants': [{'code': 'co',   'displayName': 'CO',   'fullName': 'Monossido di carbonio',   'concentration': {'value': 292.61, 'units': 'PARTI_PER_MILIARDO'},   'additionalInfo': {'sources': 'Di solito proviene dalla combustione incompleta di carburanti a base di carbonio, come avviene nei motori delle automobili e nelle centrali elettriche.',    'effects': 'Se inalato, il monossido di carbonio può impedire al sangue di trasportare ossigeno. L'esposizione può causare vertigini, nausea e mal di testa. L'esposizione a concentrazioni elevate può portare alla perdita di conoscenza.'}},  {'code': 'no2',   'displayName': 'NO2',   'fullName': 'Biossido di azoto',   'concentration': {'value': 22.3, 'units': 'PARTI_PER_MILIARDO'},   'additionalInfo': {'sources': 'Le principali fonti sono i processi di combustione dei carburanti, come quelli utilizzati nell'industria e nei trasporti.',    'effects': 'L'esposizione può causare un aumento della reattività bronchiale nei pazienti affetti da asma, un declino della funzione polmonare nei pazienti affetti da malattia polmonare ostruttiva cronica (COPD) e un aumento del rischio di infezioni respiratorie, specialmente nei bambini piccoli.'}},  {'code': 'o3',   'displayName': 'O3',   'fullName': 'Ozono',   'concentration': {'value': 24.17, 'units': 'PARTI_PER_MILIARDO'},   'additionalInfo
3. Ottenere una serie temporale della qualità dell'aria in una determinata posizione
Non sarebbe bello poter ottenere una serie temporale di questi valori di AQI e inquinanti per una determinata posizione? Ciò potrebbe rivelare interessanti pattern come correlazioni tra gli inquinanti o fluttuazioni giornaliere causate dal traffico o da fattori meteorologici.
Possiamo farlo con un'altra richiesta POST all'API delle condizioni storiche, che ci fornirà una cronistoria oraria. Questo funziona allo stesso modo delle condizioni attuali, l'unica differenza principale è che poiché i risultati possono essere piuttosto lunghi, vengono restituiti come diverse pagine, che richiede un po' di logica extra per gestirli.
Modifichiamo il metodo request_post di Client per gestire questo.
  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)    # metti la prima pagina nel dizionario di risposta    pagina = 1    final_response = {        "pagina_{}".format(pagina) : response_body    }    # recupera tutte le pagine se necessario     while "nextPageToken" in response_body:      # chiamala nuovamente con il token della pagina successiva      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)      pagina += 1      final_response["pagina_{}".format(pagina)] = response_body    return final_response
Questo gestisce il caso in cui response_body contiene un campo chiamato nextPageToken, che è l'ID della pagina successiva dei dati che è stata generata ed è pronta per essere recuperata. Quando queste informazioni sono presenti, è sufficiente chiamare nuovamente l'API con un nuovo parametro chiamato pageToken, che la indirizza alla pagina corrispondente. Facciamo questo ripetutamente in un ciclo while finché non ci sono più pagine rimaste. Il nostro dizionario final_response contiene quindi un altro livello indicato dal numero di pagina. Per le chiamate a current_conditions ci sarà solo una pagina, ma per le chiamate a historical_conditions potrebbero esserci diverse.
Una volta risolto questo problema, possiamo scrivere una funzione historical_conditions 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,):    """    Vedere la documentazione di questa API qui: https://developers.google.com/maps/documentation/air-quality/reference/rest/v1/history/lookup    """    params = {}    if isinstance(location, dict):        params["location"] = location    else:        raise ValueError(            "L'argomento posizione deve essere un dizionario contenente latitudine e longitudine"        )    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):        # nota che il tempo deve essere nel formato "Zulu"        # ad esempio datetime.datetime.strftime(datetime.datetime.now(),"%Y-%m-%dT%H:%M:%SZ")        params["dateTime"] = specific_time    # periodi di ritardo in ore    elif lag_time and not specific_time and not isinstance(specific_period, dict):        params["hours"] = lag_time    else:        raise ValueError(            "Devi fornire specific_time, specific_period o lag_time come argomenti"        )    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    # dimensioni di pagina predefinite qui impostate a 100    params["pageSize"] = 100    # il token di pagina verrà compilato se necessario dal metodo request_post    params["pageToken"] = ""    return client.request_post("/v1/history:lookup", params)
Per definire il periodo storico, l'API può accettare un lag_time in ore, fino a 720 (30 giorni). Può anche accettare un specific_period dizionario, che definisce l'inizio e la fine dei tempi nel formato descritto nei commenti precedenti. Infine, per recuperare un'ora singola di dati, può accettare solo un timestamp, fornito da specific_time. Notare anche l'uso del parametro pageSize, che controlla quanti punti temporali vengono restituiti in ogni chiamata all'API. Il valore predefinito è 100.
Proviamolo.
# istanzia il clientclient = Client(key=GOOGLE_MAPS_API_KEY)# una posizione a Los Angeles, CAlocation = {"longitude":-118.3,"latitude":34.1}# una risposta JSONhistory_conditions_data = historical_conditions(    client,    location,    lag_time=720)
Dovremmo ottenere una lunga risposta JSON nidificata che contiene i valori dell'indice AQI e i valori specifici degli inquinanti a intervalli di 1 ora negli ultimi 720 ore. Ci sono molti modi per formattare questo in una struttura più adatta alla visualizzazione e all'analisi, e la funzione di seguito lo converterà in un dataframe pandas in formato "lungo", che funziona bene con seabornper la visualizzazione.
from itertools import chainimport 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)):    # bisogna fare questo controllo nel caso in cui l'uno dei timestamp non abbia dati, che può succedere a volte    if "indexes" in chained_pages[i]:      this_element = chained_pages[i]      # recupera l'ora      time = this_element["dateTime"]      # recupera tutti i valori dell'indice e aggiungi i metadati      all_indexes += [(time , x["code"],x["displayName"],"indice",x["aqi"],None) for x in this_element['indexes']]      # recupera tutti i valori degli inquinanti e aggiungi i metadati      all_pollutants += [(time , x["code"],x["fullName"],"inquinante",x["concentration"]["value"],x["concentration"]["units"]) for x in this_element['pollutants']]    all_results = all_indexes + all_pollutants  # genera un dataframe in formato "lungo"  res = pd.DataFrame(all_results,columns=["time","code","name","type","value","unit"])  res["time"]=pd.to_datetime(res["time"])  return res
Utilizzando questa funzione sull'output di historical_conditions, otterremo un dataframe le cui colonne sono formattate per una facile analisi.
df = historical_conditions_to_df(history_conditions_data)

E ora possiamo rappresentare il risultato in seaborn o in qualche altro strumento di visualizzazione.
import seaborn as snsg = 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)

Questo è già molto interessante! Ci sono chiaramente diverse periodicità nelle serie temporali degli inquinanti ed è notevole che l'AQI US sia strettamente correlato alle concentrazioni di pm25 e pm10, come ci si aspettava. Conosco molto meno l'AQI universale che Google fornisce qui, quindi non riesco a spiegare perché sembra essere anticorrelato con pm25 e pm10. Il valore UAQI minore significa una migliore qualità dell'aria? Nonostante qualche ricerca non sono riuscito a trovare una buona risposta.
4. Ottieni le piastrelle del mosaico della qualità dell'aria
Ora per l'ultima applicazione dell'API sulla qualità dell'aria di Google Maps: generare le piastrelle del mosaico del calore. La documentazione in merito è scarsa, il che è un peccato perché queste piastrelle sono uno strumento potente per visualizzare la qualità dell'aria attuale, soprattutto quando combinate con una mappa Folium.
Le recuperiamo attraverso una richiesta GET, che comporta la costruzione di un URL nel seguente formato, in cui la posizione della piastrella è specificata da zoom, x e y
GET https://airquality.googleapis.com/v1/mapTypes/{mapType}/heatmapTiles/{zoom}/{x}/{y}
Cosa significano zoom, x e y? Possiamo rispondere a questa domanda imparando come Google Maps converte le coordinate di latitudine e longitudine in "coordinate delle piastrelle", che è descritto dettagliatamente qui. Fondamentalmente, Google Maps memorizza le immagini in griglie in cui ogni cella misura 256 x 256 pixel e le dimensioni reali della cella dipendono dal livello di zoom. Quando effettuiamo una chiamata all'API, è necessario specificare da quale griglia prendere i dati, che è determinato dal livello di zoom, e da dove prendere i dati dalla griglia, che è determinato dalle coordinate delle piastrelle x e y. Ciò che viene restituito è una sequenza di byte che può essere letta da Python Imaging Library (PIL) o da un pacchetto di elaborazione delle immagini simile.
Dopo aver formato il nostro url nel formato sopra indicato, possiamo aggiungere alcuni metodi alla classe Client che ci consentiranno di recuperare l'immagine corrispondente.
  def request_get(self,url):    request_url = self.compose_url(url)    response = self.session.get(request_url)    # per le immagini provenienti dal servizio delle piastrelle del mosaico del calore    return self.get_image(response)  @staticmethod  def get_image(response):    if response.status_code == 200:      image_content = response.content      # nota l'uso di Image da PIL qui      # necessita di from PIL import Image      image = Image.open(io.BytesIO(image_content))      return image    else:      print("La richiesta GET per l'immagine ha restituito un errore")      return None
Questo va bene, ma ciò di cui abbiamo veramente bisogno è la capacità di convertire un insieme di coordinate in longitudine e latitudine in coordinate di piastrelle. La documentazione spiega come fare - prima convertiamo le coordinate nella proiezione Mercator, da cui convertiamo in "coordinate dei pixel" utilizzando il livello di zoom specificato. Infine, traduciamo ciò nelle coordinate delle piastrelle. Per gestire tutte queste trasformazioni, possiamo utilizzare la classe TileHelper di seguito.
import mathimport 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):    # Basato sulla funzione qui    # 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):    # vedere https://developers.google.com/maps/documentation/javascript/coordinates    # per i dettagli    box_north = self._tiletolat(ty,zoom_level)    # i numeri delle piastrelle avanzano verso sud    box_south = self._tiletolat(ty+1,zoom_level)    box_west = self._tiletolon(tx,zoom_level)    # i numeri delle piastrelle avanzano verso est    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):      # il più vicino è latmin, lonmin      direction = "sud-ovest"    elif (corner_lat_idx == 0) and (corner_lon_idx == 1):      direction = "sud-est"    elif (corner_lat_idx == 1) and (corner_lon_idx == 0):      direction = "nord-ovest"    else:      direction = "nord-est"    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 == "sud-ovest":      return [(tx-1,ty),(tx-1,ty+1),(tx,ty+1)]    elif direction == "sud-est":      return [(tx+1,ty),(tx+1,ty-1),(tx,ty-1)]    elif direction == "nord-ovest":      return [(tx-1,ty-1),(tx-1,ty),(tx,ty-1)]    else:      return [(tx+1,ty-1),(tx+1,ty),(tx,ty-1)]
Possiamo vedere che location_to_tile_xy prende in input un dizionario di posizione e livello di zoom e restituisce la cella in cui si trova quel punto. Un'altra funzione utile è tile_to_bounding_box, che troverà le coordinate di delimitazione di una cella di griglia specificata. Ne abbiamo bisogno se vogliamo geolocalizzare la cella e tracciarla su una mappa.
Vediamo come funziona all'interno della funzione air_quality_tile di seguito, che prenderà in input il nostro client, location e una stringa che indica il tipo di tile che vogliamo recuperare. Dobbiamo anche specificare un livello di zoom, che può essere difficile da scegliere all'inizio e richiede un po' di prova ed errore. Discuteremo presto l'argomento get_adjoining_tiles.
def air_quality_tile(client, location, pollutant="UAQI_INDIGO_PERSIAN", zoom=4, get_adjoining_tiles=True): # vedi 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" ]  # contiene metodi utili per gestire le coordinate dei tile  helper = TileHelper()  # ottieni il tile in cui si trova la posizione  world_coordinate, pixel_coord, tile_coord = helper.location_to_tile_xy(location,zoom_level=zoom)  # ottieni il bounding box del 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 = []  # ottieni tutti i tile adiacenti, oltre a quello considerato  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])    )    # converti l'immagine PIL in 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 workflow è il seguente: prima, trovare le coordinate del tile della posizione di interesse. Questo specifica la cella di griglia che vogliamo recuperare. Quindi, trovare le coordinate di delimitazione di questa cella di griglia. Se vogliamo recuperare i tile circostanti, trovare il punto più vicino del bounding box e poi utilizzarlo per calcolare le coordinate dei tile delle tre celle di griglia adiacenti. Quindi chiamare l'API e restituire ciascun tile come immagine con il relativo bounding box.
Possiamo eseguire questo nel modo standard, come segue:
client = Client(key=GOOGLE_MAPS_API_KEY)location = {"longitude":-118.3,"latitude":34.1}zoom = 7tiles = air_quality_tile(client, location, pollutant="UAQI_INDIGO_PERSIAN", zoom=zoom, get_adjoining_tiles=False)
E quindi tracciarlo con folium per una mappa zoomabile! Nota che sto usando leafmap qui, perché questo pacchetto può generare mappe Folium che sono compatibili con gradio, uno strumento potente per generare interfacce utente semplici per applicazioni Python. Dai un'occhiata a questo articolo per un esempio.
import leafmap.foliumap as leafmapimport 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 che contiene la nostra posizione a questo livello di zoom è principalmente mare, anche se è comunque bello vedere l'inquinamento atmosferico rappresentato su una mappa dettagliata. Se effettui uno zoom, puoi vedere che le informazioni sul traffico stradale vengono utilizzate per fornire segnali sulla qualità dell'aria nelle aree urbane.

Impostando get_adjoining_tiles=True otteniamo una mappa molto più gradevole, perché scarica le tre tessere più vicine e non sovrapposte a quel livello di zoom. Nel nostro caso, ciò aiuta molto a rendere la mappa più presentabile.

Personalmente preferisco le immagini generate quando pollutant=US_AQI, ma ci sono diverse opzioni disponibili. Purtroppo, 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.
Conclusione

Grazie per aver raggiunto la fine! Qui abbiamo esplorato come utilizzare le API di qualità dell'aria di Google Maps per ottenere risultati in Python, che potrebbero essere utilizzati in interessanti applicazioni. In futuro spero di seguire con un altro articolo sullo strumento air_quality_mapper man mano che evolve ulteriormente, ma spero che gli script qui discussi siano utili anche da soli. Come sempre, ogni suggerimento per un ulteriore sviluppo sarà molto apprezzato!


                            
                            

                                Air Quality,Geospatial Data,Google Map Api,Hands On Tutorials,Timeseries