我目前正在开发一个在本地运行的应用程序 - 我希望尽快部署它!
我正在使用:
这是我目前正在解决的问题:
我正在使用英国交通部 (DfT) 的开放公交车数据 API 来获取、处理和显示伦敦各地的实时公交车位置。
获取、处理和显示总线数据简单易行。
我目前正在使用 st_folium() 将公交车位置显示为标记集群 - 主要是因为大约有 8,300 辆公交车在行驶(绘制单个公交车可能会占用大量资源)。这很好,因为当我放大单个标记时会出现。
我通常使用 folium_static() 并已切换到 st_folium() 以开始在我的地图中使用更多交互性。
这是我的问题:
我想实现一个刷新按钮,重新加载总线数据并将用户带回最后一个已知的中心和缩放位置,这样就可以产生看到总线实时移动的效果(并不完美,但它应该可以解决问题)我需要它做什么)。
总线位置数据在主机端每 10-30 秒更新一次 (DfT)。
我知道 st_folium() 返回一个字典,其中包含缩放、中心、上次单击等变量。
我也知道访问会话状态对于利用这些值很重要。
但是,我已经尝试了各种方法来成功添加刷新功能,但似乎无法使其正常工作。
我遇到了以下问题:
有人有在地图中实现类似功能的经验吗?您是否遇到过类似的问题?
任何帮助都会很棒!
非常感谢,
克里斯
这是我当前的代码,目前在没有刷新功能的情况下工作正常:
import folium
from folium.plugins import MarkerCluster
import streamlit as st
from bod_api import fetch_and_save_bus_data
from xml_to_parquet import xml_to_parquet
from geo_prep import load_data, create_gdf
from streamlit_folium import st_folium
# Set page config and title
st.set_page_config(layout="wide")
st.markdown("# :bus: Live London Bus Map")
# Set up initial map location
CENTER_START = [51.5074, -0.1278]
ZOOM_START = 12
def load_and_prepare_data():
# Call DfT API for bus data and convert xml into a parquet file
# Create a geo dataframe from the parquet (CRS is 4328)
api_key = st.secrets["tfl_bus"]["api_key"]
api_url = st.secrets["tfl_bus"]["api_url"]
bounding_box = st.secrets["tfl_bus"]["bounding_box"]
fetch_and_save_bus_data(api_url, bounding_box, api_key, "bus_location_data.xml")
xml_to_parquet("bus_location_data.xml", "bus_location_data_combined.parquet")
data = load_data("bus_location_data_combined.parquet")
geo_df = create_gdf(data)
return geo_df
def initialize_session_state():
# Set session state variables
if "center" not in st.session_state:
st.session_state["center"] = CENTER_START
if "zoom" not in st.session_state:
st.session_state["zoom"] = ZOOM_START
if "map_data" not in st.session_state:
st.session_state["map_data"] = {}
def add_markers_to_cluster(geo_df, marker_cluster):
# Iterate over rows to create markers
for index, row in geo_df.iterrows():
# Set the colour based on OperatorRef
icon_color = 'red' if row['OperatorRef'] == 'TFLO' else 'blue'
# Create popup content
popup_content = f"""
<b>Bus Route:</b> {row['PublishedLineName']}<br>
<b>Direction:</b> {row['DirectionRef']}<br>
<b>Operator:</b> {row['OperatorRef']}
"""
# Add marker to marker cluster
folium.Marker(
location=[row.geometry.y, row.geometry.x],
icon=folium.Icon(color=icon_color, icon="bus", prefix="fa"),
popup=popup_content,
tooltip=row['PublishedLineName']
).add_to(marker_cluster)
def create_map(center, zoom):
# Create the map with the marker cluster
m = folium.Map(location=center, zoom_start=zoom)
marker_cluster = MarkerCluster().add_to(m)
return m, marker_cluster
# Load session state
initialize_session_state()
# Load and prepare data
geo_df = load_and_prepare_data()
# Display the map
m, marker_cluster = create_map(st.session_state["center"], st.session_state["zoom"])
add_markers_to_cluster(geo_df, marker_cluster)
map_data = st_folium(m, center=st.session_state["center"], zoom=st.session_state["zoom"],
width=1285, height=800, key='new', returned_objects=[])
我尝试实现这样的
def main()
函数,我认为这最终导致应用程序由于某种原因而无限刷新。
def main():
initialize_session_state()
# Add a refresh button
if st.button("Refresh"):
geo_df = load_and_prepare_data()
else:
if "geo_df" not in st.session_state:
st.session_state["geo_df"] = load_and_prepare_data()
geo_df = st.session_state["geo_df"]
m, marker_cluster = create_map(st.session_state["center"], st.session_state["zoom"])
add_markers_to_cluster(geo_df, marker_cluster)
# Handle map interactions
map_data = st_folium(m, center=st.session_state["center"], zoom=st.session_state["zoom"], width=1285, height=800, key='map')
if map_data:
if map_data.get("last_center"):
st.session_state["center"] = map_data["last_center"]
if map_data.get("last_zoom"):
st.session_state["zoom"] = map_data["last_zoom"]
if __name__ == "__main__":
main()
我在 Streamlit Folium 中也遇到了同样的无限刷新问题。您是否找到解决方法或向库作者提出问题?