Potenziare la tua app con la nuova funzionalità di connessioni di Streamlit e mappe interattive di Plotly

Potenziare la tua app con connessioni di Streamlit e mappe interattive di Plotly

Aeroa: Un’app per la visualizzazione della qualità dell’aria

Introduzione

Recentemente, Streamlit ha annunciato, al momento in cui questo articolo viene scritto, la sua nuova funzionalità, st.experimental_connection, e sono molto interessato a usarla e capire come funziona. Maggiori dettagli possono essere trovati nella loro documentazione ufficiale.

Allora, cos’è questa nuova funzionalità e cosa puoi fare con essa? Attraverso di essa, puoi creare una nuova connessione a un datastore o API o restituire una già esistente. Hai anche molte opzioni di configurazione, come credenziali, segreti, ecc., per le connessioni che vengono prese da varie fonti, come file di configurazione specifici della connessione e i file secrets.toml dell’app e gli argomenti passati a questa funzione. Se mi chiedi, per cose del genere, potresti costruire qualcosa da solo con Streamlit e il tuo codice (tempo richiesto), ma ora Streamlit ti offre migliori capacità con una funzionalità integrata.

Dettagli della classe di connessione

Quindi, vediamo alcuni dettagli in più sulla classe principale che questa funzionalità utilizza. Streamlit ti dà la possibilità di creare la tua classe di connessione e chiamarla all’interno della tua app. Ci sono già alcune classi di connessione integrate per SQL e Snowpark in Snowflake. È molto facile usarle, come nell’esempio per SQL qui sotto:

import streamlit as stconn = st.experimental_connection("sql")

Puoi anche fare cose più complesse, ma le discuteremo di seguito nel prossimo esempio specifico.

• Riorganizzazione delle conferenze di football universitario – Analisi esplorativa dei dati in Python
• AudioCraft di Meta una rivoluzione nell’audio e nella musica generati da intelligenza artificiale
• Simulazione 101 Trasferimento di Calore Conduttivo

Crea la tua classe di connessione

Streamlit ha annunciato il suo nuovo hackathon per costruire app che ti permettono di creare le tue classi di connessione. Così ho deciso di partecipare e creare un’app semplice a causa delle restrizioni di tempo. Questa app utilizzerà dati sulla qualità dell’aria e alcune informazioni meteorologiche fornite da un’API aperta chiamata OpenAQ. Fornisce diverse informazioni per quasi ogni paese del mondo basandosi su sensori installati in aree specifiche.

Per utilizzare la suddetta API, dobbiamo creare una nuova classe di connessione. Questa classe includerà la nuova sessione della libreria requests, una query che ottiene i paesi (necessita di un piccolo codice personalizzato), una query principale che ottiene i dati specifici del paese scelto e… questo è tutto. La parte sottostante sarà inclusa in un file “connection.py”.

from streamlit.connections import ExperimentalBaseConnectionimport requestsimport streamlit as stclass OpenAQConnection(ExperimentalBaseConnection[requests.Session]):    def __init__(self, *args, **kwargs):        super().__init__(*args, **kwargs)        self._resource = self._connect(**kwargs)    def _connect(self, **kwargs) -> requests.Session:        session = requests.Session()        return session    def cursor(self):        return self._resource    def query_countries(        self, limit=100, page=1, sort="asc", order_by="name", ttl: int = 3600    ):        @st.cache_data(ttl=ttl)        def _query_countries(limit, page, sort, order_by):            params = {                "limit": limit,                "page": page,                "sort": sort,                "order_by": order_by,            }            with self._resource as s:                response = s.get("https://api.openaq.org/v2/countries", params=params)            return response.json()        return _query_countries(limit, page, sort, order_by)    def query(        self,        country_id,        limit=1000,        page=1,        offset=0,        sort="desc",        radius=1000,        order_by="lastUpdated",        dumpRaw="false",        ttl: int = 3600,    ):        @st.cache_data(ttl=ttl)        def _get_locations_measurements(            country_id, limit, page, offset, sort, radius, order_by, dumpRaw        ):            params = {                "limit": limit,                "page": page,                "offset": offset,                "sort": sort,                "radius": radius,                "order_by": order_by,                "dumpRaw": dumpRaw,            }            if country_id is not None:                params["country_id"] = country_id            with self._resource as s:                response = s.get("https://api.openaq.org/v2/locations", params=params)            return response.json()        return _get_locations_measurements(            country_id, limit, page, offset, sort, radius, order_by, dumpRaw        )

Ovviamente, all’interno di questa connessione, utilizzo @st.cache_data(ttl=ttl) per memorizzare nella cache gli output. Per capire meglio gli argomenti utilizzati per la chiamata ai diversi endpoint, consulta la documentazione API corrispondente.

Crea la funzione di visualizzazione

Per la visualizzazione, viene utilizzata la libreria plotly e in particolare la classe Scattermapbox della classe go. (la funzione di seguito è molto grande per motivi di layout e potrebbe essere suddivisa in più parti, ma perdonami):

import plotly.graph_objects as godef visualize_variable_on_map(data_dict, variable):    is_day = is_daytime()    mapbox_style = "carto-darkmatter" if not is_day else "open-street-map"    # Inizializza le liste per memorizzare i dati di più posizioni    latitudes = []    longitudes = []    values = []    display_names = []    last_updated = []    # Scorri i risultati ed estrai i dati rilevanti per ogni posizione    for result in data_dict.get("results", []):        measurements = result.get("parameters", [])        for measurement in measurements:            if measurement["parameter"] == variable:                value = measurement["lastValue"]                display_name = measurement["displayName"]                latitude = result["coordinates"]["latitude"]                longitude = result["coordinates"]["longitude"]                last_updated_value = result["lastUpdated"]                latitudes.append(latitude)                longitudes.append(longitude)                values.append(value)                display_names.append(display_name)                last_updated.append(last_updated_value)    if not latitudes or not longitudes or not values:        print(f"I dati di {variable} non sono stati trovati.")        return create_custom_markdown_card(            f"I dati di {variable} non sono stati trovati per il paese selezionato."        )    # Crea la visualizzazione    fig = go.Figure()    marker = [        custom_markers["humidity"]        if variable == "humidity"        else custom_markers["others"]    ]    # Aggiungi un'unica traccia di mappa con tutte le posizioni    fig.add_trace(        go.Scattermapbox(            lat=latitudes,            lon=longitudes,            mode="markers+text",            marker=dict(                size=20,                color=values,                colorscale="Viridis",  # Puoi scegliere altre scale di colore anche                colorbar=dict(title=f"{variable.capitalize()}"),            ),            text=[                f"{marker[0]} {display_name}: {values[i]}<br>Ultimo aggiornamento: {last_updated[i]}"                for i, display_name in enumerate(display_names)            ],            hoverinfo="text",        )    )    # Aggiorna il layout della mappa    fig.update_layout(        mapbox=dict(            style=mapbox_style,  # Scegli lo stile di mappa desiderato            zoom=5,  # Regola il livello di zoom iniziale come necessario            center=dict(                lat=sum(latitudes) / len(latitudes),                lon=sum(longitudes) / len(longitudes),            ),        ),        margin=dict(l=0, r=0, t=0, b=0),    )    create_custom_markdown_card(information)    st.plotly_chart(fig, use_container_width=True)

Crea l’applicazione

Il codice seguente è incluso nel nostro file “app.py”:

import streamlit as stfrom connection import OpenAQConnectionfrom utils import * # una parte di utilità personalizzata con funzioni di supportost.set_page_config(page_title="OpenAQ Connection", layout="wide")conn = st.experimental_connection("openaq", type=OpenAQConnection)# nel caso in cui tu abbia un file readme tomlreadme = load_config("config_readme.toml")# Infost.title("Dati sulla qualità dell'aria")with st.expander("Che cos'è questa app?", expanded=False):    st.write(readme["app"]["app_intro"])    st.write("")st.write("")st.sidebar.image(load_image("logo.png"), use_column_width=True)display_links(readme["links"]["repo"], readme["links"]["other_link"])with st.spinner("Caricamento dei paesi disponibili..."):    # I paesi esistono nelle prime 2 pagine    countries = []    for page in [1, 2]:        try:            countries_request = conn.query_countries(page=page)["results"]            countries = countries + countries_request        except Exception:            countries_error = True    transformed_countries = {        country["name"]: {            "code": country["code"],            "parameters": country["parameters"],            "locations": country["locations"],            "lastUpdated": country["lastUpdated"],        }        for country in countries    }    # Aggiungi un valore globale predefinito quando l'app viene inizializzata    transformed_countries["Global"] = {        "code": None,        "parameters": general_parameters,        "locations": None,        "lastUpdated": None,    }# Parametrist.sidebar.title("Selezione")selected_country = st.sidebar.selectbox(    "Seleziona il paese desiderato",    transformed_countries,    placeholder="Paese",    index=len(transformed_countries) - 1,  # Ottiene l'ultimo "Global"    help=readme["tooltips"]["country"],)selected_viariable = st.sidebar.selectbox(    "Seleziona la variabile desiderata",    transformed_countries[selected_country]["parameters"],    placeholder="Variabile",    index=1,    help=readme["tooltips"]["variable"],)radius = st.sidebar.slider(    "Seleziona un raggio",    min_value=100,    max_value=25000,    step=100,    value=1000,    help=readme["tooltips"]["radius"],)total_locations = transformed_countries[selected_country]["locations"]last_time = transformed_countries[selected_country]["lastUpdated"]information = f"Il paese selezionato è {selected_country}. Le posizioni trovate in totale sono {total_locations} con gli ultimi aggiornamenti a {last_time}."code = transformed_countries[selected_country]["code"]locations_response = conn.query(code, radius)st.title("Mappa")visualize_variable_on_map(locations_response, selected_viariable)

Quindi, dopo aver eseguito la nostra app “streamlit run app.py”, abbiamo la nostra app in esecuzione.

Ho chiamato l’app “AEROA” e puoi trovarla implementata nella community cloud di streamlit qui. Puoi anche trovare il codice sorgente su Github e giocarci secondo le tue preferenze.

Conclusioni

In questo breve tutorial, abbiamo mostrato la nuova funzionalità st.experimental_connection di streamlit e l’abbiamo utilizzata per stabilire una connessione con un’API aperta che fornisce dati sulla qualità dell’aria. Inoltre, abbiamo sviluppato anche una nuova app che mostra i risultati in una mappa di plotly.