使用Streamlit的新连接功能和交互式Plotly地图增强您的应用程序
使用Streamlit的新连接功能和交互式Plotly地图增强应用程序
Aeroa：一个用于空气质量可视化的应用程序
介绍
最近，Streamlit宣布了其新功能st.experimental_connection，我对使用它并了解其工作原理非常感兴趣。更多细节可以在他们的官方文档中找到。
那么，这个新功能是什么，以及你可以用它做什么？通过它，你可以创建一个新的与数据存储或API的连接，或者返回一个已存在的连接。你还有很多配置选项，如凭据、密钥等，这些选项从各种来源获取，例如任何与连接相关的配置文件、应用程序的secrets.toml文件以及传递给此函数的kwargs参数。如果你问我，对于这种情况，你可以使用Streamlit和你自己的代码（需要时间）构建一些东西，但现在Streamlit通过内置功能为你提供了更好的能力。
连接类的详细信息
那么，让我们看一下这个功能使用的主要类的更多细节。Streamlit允许你创建自己的连接类并在应用程序中调用它。已经为SQL和Snowflake中的Snowpark创建了一些内置的连接类。使用它们非常简单，如下面的SQL示例：
import streamlit as stconn = st.experimental_connection("sql")你也可以做更复杂的事情，但我们将在下一个具体的示例中讨论。
构建自己的连接类
Streamlit宣布了一个新的黑客马拉松活动，目的是构建允许你创建自己的连接类的应用程序。因此，我决定参与并创建一个简单的应用程序，因为时间有限。这个应用程序将使用由一个名为OpenAQ的开放API提供的空气质量和一些天气数据。它基于特定区域安装的传感器为世界上几乎每个国家提供了几个数据。
为了使用上述API，我们必须创建一个新的连接类。这个类将包括requests库的新会话、一个获取国家的查询（它需要一小段自定义代码）、一个获取选择的国家的特定数据的主要查询，以及…就是这样。下面的部分将包含在一个名为“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 )当然，在这个连接中，我使用了@st.cache_data(ttl=ttl)以便缓存输出结果。为了更好地理解调用不同端点时使用的参数，请查看相应的API文档。
创建可视化函数
对于可视化，使用的是plotly库，具体来说是来自go类的Scattermapbox。（下面的函数由于布局原因很长，可以分成更多部分，但请原谅我）：
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" # 初始化用于存储多个位置数据的列表 latitudes = [] longitudes = [] values = [] display_names = [] last_updated = [] # 循环遍历结果并提取每个位置的相关数据 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"{variable}数据未找到。") return create_custom_markdown_card( f"未找到所选择的国家的{variable}数据。" ) # 创建可视化 fig = go.Figure() marker = [ custom_markers["humidity"] if variable == "humidity" else custom_markers["others"] ] # 添加一个带有所有位置的单个散点图 fig.add_trace( go.Scattermapbox( lat=latitudes, lon=longitudes, mode="markers+text", marker=dict( size=20, color=values, colorscale="Viridis", # 也可以选择其他颜色比例尺 colorbar=dict(title=f"{variable.capitalize()}"), ), text=[ f"{marker[0]} {display_name}: {values[i]}<br>最后更新时间：{last_updated[i]}" for i, display_name in enumerate(display_names) ], hoverinfo="text", ) ) # 更新地图布局 fig.update_layout( mapbox=dict( style=mapbox_style, # 选择所需的地图样式 zoom=5, # 根据需要调整初始缩放级别 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)创建应用
下面的代码包含在我们的“app.py”文件中：
import streamlit as stfrom connection import OpenAQConnectionfrom utils import * # 一个包含支持函数的自定义工具部分st.set_page_config(page_title="OpenAQ Connection", layout="wide")conn = st.experimental_connection("openaq", type=OpenAQConnection)# 如果有readme toml文件readme = load_config("config_readme.toml")# 信息st.title("空气质量数据")with st.expander("这个应用是什么？", 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("正在加载可用国家..."): # 国家存在于前两页 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 } # 在应用初始化时添加一个默认的全球选项 transformed_countries["全球"] = { "code": None, "parameters": general_parameters, "locations": None, "lastUpdated": None, }# 参数st.sidebar.title("选择")selected_country = st.sidebar.selectbox( "选择所需的国家", transformed_countries, placeholder="国家", index=len(transformed_countries) - 1, # 获取最后一个"全球" help=readme["tooltips"]["country"],)selected_viariable = st.sidebar.selectbox( "选择所需的变量", transformed_countries[selected_country]["parameters"], placeholder="变量", index=1, help=readme["tooltips"]["variable"],)radius = st.sidebar.slider( "选择半径", 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"所选的国家是{selected_country}。总共找到{total_locations}个位置，最后更新时间为{last_time}。"code = transformed_countries[selected_country]["code"]locations_response = conn.query(code, radius)st.title("地图")visualize_variable_on_map(locations_response, selected_viariable)所以在运行我们的应用程序“streamlit run app.py”之后，我们的应用程序正在运行。
我将这个应用程序称为“AEROA”，您可以在streamlit社区云中找到它的部署位置。您还可以在Github上找到源代码，并根据自己的喜好进行调试。
结论
在这个快速教程中，我们展示了从streamlit中使用的新功能st.experimental_connection，并使用它与提供空气质量数据的开放API建立了连接。除此之外，我们还开发了一个漂亮的新应用程序，在plotly地图中显示结果。
