Un outil Python pour récupérer les données de pollution de l’air à partir des API Google Maps Air Quality.

Un outil Python pour extraire les données de pollution de l'air depuis les API Google Maps Air Quality.

Apprenez comment obtenir des données de qualité de l’air en temps réel riches du monde entier

Cet article détaille comment nous pouvons utiliser les API de qualité de l’air de Google Maps en Python pour obtenir et explorer des données de pollution de l’air en direct, des séries chronologiques et des cartes. Consultez le code complet ici.

1. Contexte

En août 2023, Google a annoncé l’ajout d’un service de qualité de l’air à sa liste d’API de cartographie. Vous pouvez en lire plus à ce sujet ici. Il semble que ces informations soient maintenant également disponibles dans l’application Google Maps, bien que les données obtenables via les API se soient avérées beaucoup plus riches.

Conformément à l’annonce, Google combine des informations provenant de nombreuses sources à différentes résolutions, telles que des capteurs de pollution au sol, des données satellites, des informations sur le trafic en direct et des prévisions à partir de modèles numériques, pour produire un ensemble de données dynamiques de qualité de l’air dans 100 pays avec une résolution pouvant atteindre 500 mètres. Cela semble être un ensemble de données très intéressant et potentiellement utile pour toutes sortes d’applications de cartographie, de santé et de planification!

Lorsque j’ai lu cela pour la première fois, je prévoyais de l’essayer dans une application de “dialogue avec vos données”, en utilisant certaines des choses apprises lors de la construction de cet outil de planification de voyage. Peut-être un système capable de tracer une série chronologique des concentrations de pollution de l’air dans votre ville préférée, ou peut-être un outil pour aider les gens à planifier des randonnées dans leur région locale pour éviter l’air pollué?

Il existe trois outils API qui peuvent aider ici – un service de “conditions actuelles” qui fournit des valeurs d’indice de qualité de l’air actuelle et des concentrations de polluants à un emplacement donné ; un service de “conditions historiques” qui fait la même chose mais à des intervalles horaires jusqu’à 30 jours dans le passé, et un service de “heatmap” qui fournit les conditions actuelles sur une zone donnée sous la forme d’une image.

Précédemment, j’avais utilisé l’excellent package googlemaps pour appeler les APIs de Google Maps en Python, mais ces nouvelles APIs ne sont pas encore prises en charge. Étonnamment, en dehors de la documentation officielle, j’ai trouvé peu d’exemples de personnes utilisant ces outils nouveaux et aucun package Python préexistant conçu pour les appeler. J’accepterais volontiers une correction si quelqu’un sait autrement!

J’ai donc construit rapidement mes propres outils, et dans cet article, nous expliquerons comment ils fonctionnent et comment les utiliser. J’espère que cela sera utile à tous ceux qui souhaitent expérimenter avec ces nouvelles APIs en Python et qui cherchent un point de départ. Tout le code de ce projet peut être trouvé ici, et il est probable que j’étendrai ce dépôt au fil du temps en ajoutant de nouvelles fonctionnalités et en construisant une sorte d’application de cartographie avec les données de qualité de l’air.

2. Obtenez la qualité de l’air actuelle dans un emplacement donné

Commençons! Dans cette section, nous verrons comment obtenir des données sur la qualité de l’air dans un emplacement donné avec Google Maps. Vous aurez d’abord besoin d’une clé d’API que vous pouvez générer via votre compte Google Cloud. Ils offrent une période d’essai gratuite de 90 jours, après quoi vous devrez payer pour les services API que vous utilisez. Assurez-vous d’activer l’API “Qualité de l’air” et de prendre connaissance des politiques de tarification avant de commencer à effectuer un grand nombre d’appels!

Capture d'écran de la bibliothèque d'API Google Cloud, à partir de laquelle vous pouvez activer l'API de qualité de l'air. Image générée par l'auteur.

Je stocke généralement ma clé d’API dans un fichier .env et je le charge avec dotenv en utilisant une fonction comme celle-ci

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,    }

Obtenir les conditions actuelles nécessite une requête POST comme indiqué ici. Nous allons nous inspirer du package googlemaps pour le faire d’une manière généralisée. Tout d’abord, nous construisons une classe client qui utilise requests pour effectuer l’appel. L’objectif est assez simple – nous voulons construire une URL comme celle ci-dessous et inclure toutes les options de requête spécifiques à la requête de l’utilisateur.

https://airquality.googleapis.com/v1/currentConditions:lookup?key=VOTRE_CLÉ_API

La classe Client prend notre clé d’API en tant que clé et construit ensuite l’request_url pour la requête. Il accepte les options de requête sous forme d’un dictionnaire params puis les place dans le corps JSON de la requête, qui est géré par l’appel self.session.post().

import requestsimport ioclass Client(object):    BASE_URL_PAR_DÉFAUT = "https://airquality.googleapis.com"    def __init__(self, clé):        self.session = requests.Session()        self.clé = clé    def demande_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, chemin):        return self.BASE_URL_PAR_DÉFAUT + chemin + "?" + "key=" + self.clé    @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",        }

Maintenant, nous pouvons créer une fonction qui aide l’utilisateur à assembler des options de requête valides pour l’API des conditions actuelles, puis utilise cette classe Client pour effectuer la requête. Encore une fois, cela s’inspire de la conception du package googlemaps.

def conditions_actuelles(    client,    emplacement,    inclure_QAI_local=True,    inclure_suggestion_santé=False,    inclure_tous_les_polluants=True,    inclure_informations_polluants_supplémentaires=False,    inclure_concentration_polluant_prédominant=True,    langue=None,):    """    Voir la documentation pour cette API ici    https://developers.google.com/maps/documentation/air-quality/reference/rest/v1/currentConditions/lookup    """    params = {}    if isinstance(emplacenemt, dict):        params["emplacement"] = emplacement    else:        raise ValueError(            "L'argument d'emplacement doit être un dictionnaire contenant la latitude et la longitude"        )    calculs_supplémentaires = []    if inclure_QAI_local:        calculs_supplémentaires.append("QAI_LOCAL")    if inclure_suggestion_santé:        calculs_supplémentaires.append("RECOMMANDATIONS_SANTÉ")    if inclure_informations_polluants_supplémentaires:        calculs_supplémentaires.append("INFORMATIONS_SUPPLÉMENTAIRES_POLLUANT")    if inclure_tous_les_polluants:        calculs_supplémentaires.append("CONCENTRATION_POLLUANT")    if inclure_concentration_polluant_prédominant:        calculs_supplémentaires.append("CONCENTRATION_POLLUANT_PRÉDOMINANT")    if langue:        params["langue"] = langue    params["extraCalculs"] = calculs_supplémentaires    return client.demande_post("/v1/currentConditions:lookup", params)

Les options pour cette API sont relativement simples. Elle nécessite un dictionnaire avec la longitude et la latitude du point que vous souhaitez examiner, et peut éventuellement prendre en compte différents autres arguments qui contrôlent la quantité d’informations renvoyées. Voyons cela en action avec tous les arguments définis sur True

# mettre en place le clientclient = Client(clé=CLÉ_API_GOOGLE_MAPS)# un emplacement à Los Angeles, CAemplacement = {"longitude":-118.3,"latitude":34.1}# une réponse JSONdonnées_conditions_actuelles = conditions_actuelles(  client,  emplacement,  inclure_suggestion_santé=True,  inclure_informations_polluants_supplémentaires=True)

Beaucoup d’informations intéressantes sont renvoyées ! Non seulement nous avons les valeurs de l’indice de la qualité de l’air des indices AQI universels et basés aux États-Unis, mais nous avons également des concentrations des principaux polluants, une description de chacun d’eux et un ensemble global de recommandations sanitaires pour la qualité de l’air actuelle.

{'dateTime': '2023-10-12T05:00:00Z', 'codeDeRégion': 'us', 'indices': [{'code': 'uaqi',   'displayName': 'Indice AQI universel',   'aqi': 60,   'aqiDisplay': '60',   'couleur': {'rouge': 0.75686276, 'vert': 0.90588236, 'bleu': 0.09803922},   'catégorie': 'Bonne qualité de l'air',   'polluantDominant': 'pm10'},  {'code': 'usa_epa',   'displayName': 'AQI (États-Unis)',   'aqi': 39,   'aqiDisplay': '39',   'couleur': {'vert': 0.89411765},   'catégorie': 'Bonne qualité de l'air',   'polluantDominant': 'pm10'}], 'polluants': [{'code': 'co',   'displayName': 'CO',   'fullName': 'Monoxyde de carbone',   'concentration': {'valeur': 292,61, 'unités': 'PARTIES_PAR_MILLIARD'},   'informationsSupplémentaires': {'sources': 'Provenant généralement de la combustion incomplète des combustibles carbonés, telle que celle qui se produit dans les moteurs de voiture et les centrales électriques.',    'effets': 'Lorsqu'il est inhalé, le monoxyde de carbone peut empêcher le sang de transporter de l'oxygène. L'exposition peut provoquer des étourdissements, des nausées et des maux de tête. Une exposition à des concentrations extrêmes peut entraîner une perte de conscience.'}},  {'code': 'no2',   'displayName': 'NO2',   'fullName': 'Dioxyde d'azote',   'concentration': {'valeur': 22,3, 'unités': 'PARTIES_PAR_MILLIARD'},   'informationsSupplémentaires': {'sources': 'Les principales sources sont les processus de combustion des combustibles, tels que ceux utilisés dans l'industrie et les transports.',    'effets': 'L'exposition peut provoquer une réactivité bronchique accrue chez les patients asthmatiques, une diminution de la fonction pulmonaire chez les patients atteints de la maladie pulmonaire

3. Obtenir une série chronologique de la qualité de l'air à un emplacement donné

Ne serait-il pas agréable de pouvoir récupérer une série chronologique de ces valeurs de qualité de l'air et de polluants pour un emplacement donné ? Cela pourrait révéler des motifs intéressants, tels que des corrélations entre les polluants ou des fluctuations quotidiennes causées par la circulation ou les facteurs liés à la météo.

Nous pouvons faire cela avec une autre requête POST vers l'API des conditions historiques, qui nous donnera un historique horaire. Cela fonctionne de la même manière que les conditions actuelles, la seule différence majeure étant que, comme les résultats peuvent être assez longs, ils sont renvoyés sous la forme de plusieurs pages, ce qui nécessite une petite logique supplémentaire pour les gérer.

Modifions la méthode request_post de la classe Client pour gérer cela.

  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)    # mettez la première page dans le dictionnaire de réponse    page = 1    final_response = {        "page_{}".format(page) : response_body    }    # récupérez toutes les pages si nécessaire     while "nextPageToken" in response_body:      # appelez à nouveau avec le jeton de la page suivante      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

Cela gère le cas où le corps de la réponse contient un champ appelé nextPageToken, qui est l'identifiant de la page suivante des données générées et prêtes à être récupérées. Lorsque cette information existe, nous devons simplement appeler à nouveau l'API avec un nouveau paramètre appelé pageToken, qui le dirige vers la page correspondante. Nous faisons cela de manière répétée dans une boucle tant qu'il reste des pages. Notre dictionnaire final_response contient donc maintenant une autre couche indiquée par le numéro de page. Pour les appels à current_conditions, il n'y aura toujours qu'une seule page, mais pour les appels à historical_conditions, il peut y en avoir plusieurs.

Avec cela pris en charge, nous pouvons écrire une fonction historical_conditions de manière très similaire à 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,):    """    Voir la documentation de cette API ici 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'argument Location doit être un dictionnaire contenant la latitude et la 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):        # notez que le temps doit être au format "Zulu"        # par exemple datetime.datetime.strftime(datetime.datetime.now(),"%Y-%m-%dT%H:%M:%SZ")        params["dateTime"] = specific_time    # périodes de retard en heures    elif lag_time and not specific_time and not isinstance(specific_period, dict):        params["hours"] = lag_time    else:        raise ValueError(            "Doit fournir les arguments specific_time, specific_period ou lag_time"        )    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    # taille de la page paramétrée par défaut à 100 ici    params["pageSize"] = 100    # le jeton de page sera rempli si nécessaire par la méthode request_post    params["pageToken"] = ""    return client.request_post("/v1/history:lookup", params)

Pour définir la période historique, l'API peut accepter un lag_time en heures, jusqu'à 720 (30 jours). Elle peut également accepter un dictionnaire specific_period, qui définit les heures de début et de fin dans le format décrit dans les commentaires ci-dessus. Enfin, pour obtenir une seule heure de données, elle peut accepter un seul horodatage, fourni par specific_time. Notez également l'utilisation du paramètre pageSize, qui contrôle le nombre de points temporels renvoyés dans chaque appel à l'API. Par défaut, il est de 100.

Essayons.

# configurer le clientclient = Client(key=GOOGLE_MAPS_API_KEY)# un emplacement à Los Angeles, CAlocation = {"longitude":-118.3,"latitude":34.1}# une réponse JSONhistory_conditions_data = historical_conditions(    client,    location,    lag_time=720)

Nous devrions obtenir une longue réponse JSON imbriquée qui contient les valeurs de l'indice de la qualité de l'air (AQR) et les valeurs spécifiques des polluants par incréments d'une heure au cours des 720 dernières heures. Il existe de nombreuses façons de formater cela dans une structure plus propice à la visualisation et à l'analyse, et la fonction ci-dessous le convertira en un dataframe pandas au format "long", qui fonctionne bien avec seaborn pour les graphiques.

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)):    # besoin de cette vérification au cas où l'un des horodatages ne contient pas de données, ce qui peut parfois arriver    if "indexes" in chained_pages[i]:      this_element = chained_pages[i]      # récupérer l'heure      time = this_element["dateTime"]      # récupérer toutes les valeurs d'indice et ajouter les métadonnées      all_indexes += [(time , x["code"],x["displayName"],"index",x["aqi"],None) for x in this_element['indexes']]      # récupérer toutes les valeurs de polluant et ajouter les métadonnées      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  # générer un dataframe au format "long"  res = pd.DataFrame(all_results,columns=["time","code","name","type","value","unit"])  res["time"]=pd.to_datetime(res["time"])  return res

En exécutant cela sur la sortie de historical_conditions, on obtiendra un dataframe dont les colonnes sont formatées pour une analyse facile.

df = historical_conditions_to_df(history_conditions_data)
Exemple de dataframe de données historiques de l'indice de qualité de l'air (AQR), prêt pour l'affichage graphique

Et maintenant nous pouvons afficher le résultat dans seaborn ou un autre outil de visualisation.

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)
Indices AQR universel, AQR aux États-Unis, valeurs de pm25 et pm10 pour cet emplacement à Los Angeles sur une période de 30 jours. Image générée par l'auteur.

C'est déjà très intéressant ! Il y a clairement plusieurs périodicités dans les séries chronologiques des polluants et il est notable que l'AQR aux États-Unis est étroitement corrélé aux concentrations de pm25 et pm10, comme prévu. Je ne connais pas très bien l'AQR universel que Google fournit ici, donc je ne peux pas expliquer pourquoi il semble être en anti-corrélation avec le pm25 et p10. Est-ce que plus petit UAQI signifie une meilleure qualité de l'air ? Malgré quelques recherches, je n'ai pas pu trouver une bonne réponse.

4. Obtenir des tuiles de carte de la qualité de l'air

Maintenant pour le dernier cas d'utilisation de l'API Google Maps Air Quality - générer des tuiles de carte de chaleur. La documentation à ce sujet est sparse, ce qui est dommage car ces tuiles sont un outil puissant pour visualiser la qualité de l'air actuelle, surtout lorsqu'elles sont combinées à une carte Folium.

Nous les récupérons avec une requête GET, ce qui implique la construction d'une URL au format suivant, où l'emplacement de la tuile est spécifié par zoom, x et y

GET https://airquality.googleapis.com/v1/mapTypes/{mapType}/heatmapTiles/{zoom}/{x}/{y}

Que signifient zoom, x et y? Nous pouvons répondre à cela en apprenant comment Google Maps convertit les coordonnées en latitude et longitude en "coordonnées de tuiles", ce qui est décrit en détail ici. Essentiellement, Google Maps stocke des images dans des grilles où chaque cellule mesure 256 x 256 pixels et les dimensions réelles de la cellule sont une fonction du niveau de zoom. Lorsque nous faisons un appel à l'API, nous devons spécifier quelle grille utiliser, ce qui est déterminé par le niveau de zoom, et où sur la grille dessiner, ce qui est déterminé par les coordonnées de tuiles x et y. Ce qui revient est un tableau de bytes qui peut être lu par Python Imaging Library (PIL) ou un package de traitement d'images similaire.

Une fois notre url formée selon le format ci-dessus, nous pouvons ajouter quelques méthodes à la classe Client qui nous permettront de récupérer l'image correspondante.

  def request_get(self,url):    request_url = self.compose_url(url)    response = self.session.get(request_url)    # pour les images provenant du service de tuiles de carte de chaleur    return self.get_image(response)  @staticmethod  def get_image(response):    if response.status_code == 200:      image_content = response.content      # notez l'utilisation de l'Image de PIL ici      # needs from PIL import Image      image = Image.open(io.BytesIO(image_content))      return image    else:      print("La requête GET pour l'image a renvoyé une erreur")      return None

C'est bien, mais ce dont nous avons vraiment besoin, c'est la possibilité de convertir un ensemble de coordonnées en longitude et latitude en coordonnées de tuiles. La documentation explique comment faire - nous convertissons d'abord les coordonnées en projection de Mercator, à partir de laquelle nous convertissons en "coordonnées de pixels" en utilisant le niveau de zoom spécifié. Enfin, nous traduisons cela en coordonnées de tuiles. Pour gérer toutes ces transformations, nous pouvons utiliser la classe TileHelper ci-dessous.

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):    # Basé sur la fonction ici    # 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):    # voir https://developers.google.com/maps/documentation/javascript/coordinates    # pour les détails    box_north = self._tiletolat(ty,zoom_level)    # les numéros de tuile avancent vers le sud    box_south = self._tiletolat(ty+1,zoom_level)    box_west = self._tiletolon(tx,zoom_level)    # les numéros de tuile avancent vers l'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):      # le coin le plus proche est latmin, lonmin      direction = "sud-ouest"    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-ouest"    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-ouest":      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-ouest":      return [(tx-1,ty-1),(tx-1,ty),(tx,ty-1)]    else:      return [(tx+1,ty-1),(tx+1,ty),(tx,ty-1)]

Nous pouvons voir que location_to_tile_xy prend en compte un dictionnaire de localisation et un niveau de zoom, et renvoie la tuile dans laquelle ce point peut être trouvé. Une autre fonction utile est tile_to_bounding_box, qui trouvera les coordonnées de délimitation d'une cellule de grille spécifiée. Nous en avons besoin si nous allons géolocaliser la cellule et la représenter sur une carte.

Voyons comment cela fonctionne à l'intérieur de la fonction air_quality_tile ci-dessous, qui va prendre notre client, notre localisation et une chaîne indiquant le type de tuile que nous voulons récupérer. Nous devons également spécifier un niveau de zoom, ce qui peut être difficile à choisir au départ et nécessite un peu d'essais et d'erreurs. Nous discuterons de l'argument get_adjoining_tiles sous peu.

def air_quality_tile(client, location, pollutant="UAQI_INDIGO_PERSIAN", zoom=4, get_adjoining_tiles=True):  # voir 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"  ]  # contient des méthodes utiles pour manipuler les coordonnées des tuiles  helper = TileHelper()  # obtenir la tuile dans laquelle la localisation se trouve  world_coordinate, pixel_coord, tile_coord = helper.location_to_tile_xy(location, zoom_level=zoom)  # obtenir la boîte de délimitation de la tuile  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 = []  # obtenir toutes les tuiles adjacentes, ainsi que celle en 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])    )    # convertir l'image PIL en numpy    try:      image_response = np.array(image_response)    except:      image_response = None    tiles.append({        "bounds": bounding_box,        "image": image_response    })  return tiles

En lisant le code, nous pouvons voir que le flux de travail est le suivant : premièrement, trouver les coordonnées de la tuile de la localisation d'intérêt. Cela spécifie la cellule de grille que nous voulons récupérer. Ensuite, trouver les coordonnées de délimitation de cette cellule de grille. Si nous voulons récupérer les tuiles environnantes, trouver le coin le plus proche de la boîte de délimitation, puis utiliser cela pour calculer les coordonnées de tuile des trois cellules de grille adjacentes. Ensuite, appeler l'API et renvoyer chacune des tuiles en tant qu'image avec sa boîte de délimitation correspondante.

Nous pouvons l'exécuter de manière standard, comme suit :

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)

Et ensuite tracer avec folium une carte zoomable ! Notez que j'utilise leafmap ici, car ce paquet peut générer des cartes Folium compatibles avec gradio, un outil puissant pour générer des interfaces utilisateur simples pour les applications Python. Consultez cet article pour un exemple.

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)

Peut-être de manière décevante, la tuile contenant notre emplacement à ce niveau de zoom est principalement de la mer, bien qu'il soit quand même agréable de voir la pollution de l'air tracée sur une carte détaillée. Si vous zoomez, vous verrez que les informations sur la circulation routière sont utilisées pour informer les signaux de qualité de l'air dans les zones urbaines.

Tracé d'une tuile de carte thermique de la qualité de l'air sur une carte Folium. Image générée par l'auteur.

En définissant get_adjoining_tiles=True, nous obtenons une carte beaucoup plus agréable car elle récupère les trois tuiles les plus proches et non superposées à ce niveau de zoom. Dans notre cas, cela aide beaucoup à rendre la carte plus présentable.

Lorsque nous récupérons également les tuiles adjacentes, un résultat beaucoup plus intéressant est obtenu. Notez que les couleurs ici représentent l'indice CQA universel. Image générée par l'auteur.

Personnellement, je préfère les images générées lorsque pollutant=US_AQI, mais il existe plusieurs options différentes. Malheureusement, l'API ne renvoie pas d'échelle de couleurs, bien qu'il serait possible d'en générer une en utilisant les valeurs de pixels de l'image et la connaissance de ce que les couleurs signifient.

Les mêmes tuiles que ci-dessus colorées selon l'indice CQA américain. Cette carte a été générée le 10/12/2023 et le point rouge vif au centre de la Californie semble être un incendie prescrit dans les collines près de Coalinga, selon cet outil <a class=https://www.frontlinewildfire.com/california-wildfire-map/. Image générée par l'auteur." src="https://ai.miximages.com/miro.medium.com/v2/resize:fit:640/format:webp/1*lZ9azFqQ1plGa-MGKxfvwg.png"/>

Conclusion

Merci d'être arrivé jusqu'à la fin ! Nous avons ici exploré comment utiliser les API de qualité de l'air de Google Maps pour obtenir des résultats en Python, qui pourraient être utilisés dans différentes applications intéressantes. À l'avenir, j'espère pouvoir faire suite avec un autre article sur l'outil air_quality_mapper alors qu'il évolue, mais j'espère que les scripts discutés ici seront utiles en soi. Comme toujours, toute suggestion de développement ultérieur serait grandement appréciée !

We will continue to update IPGirl; if you have any questions or suggestions, please contact us!

Share:

Was this article helpful?

93 out of 132 found this helpful

Discover more

Science des données

Série chronologique pour le changement climatique Prévision de la demande d'origine-destination

L'exploitation des données de voitures flottantes est une tâche clé dans les systèmes de transport intelligents. Les ...

AI

Qu'est-ce qu'un générateur d'images IA ? Quelques-uns des meilleurs générateurs d'images IA en 2023

Un générateur d’images IA est une application de pointe de l’intelligence artificielle capable de créer d...

AI

Présentation d'OpenLLM Bibliothèque Open Source pour LLMs

Une plateforme conviviale pour exploiter des grands modèles de langage (LLM) en production, avec des fonctionnalités ...

AI

Les États-Unis traquent les logiciels malveillants chinois qui pourraient perturber les opérations militaires

Les responsables de la sécurité américaine affirment que la Maison Blanche est à la recherche de logiciels malveillan...

AI

La caméra arrête les deepfakes à l'instant du déclenchement

Les références de contenu intégrées vérifient l'authenticité des photos.

AI

Une nouvelle recherche sur l'apprentissage en profondeur identifie un médicament antipaludique comme possible traitement de l'ostéoporose

Le problème de l’ostéoporose, une affection caractérisée par une perte excessive de tissu osseux et un risque a...