Streamlit仪表盘状态持久化到JSON：Pydantic与回调机制实践

import os
import json
from typing import List, Optional
from pydantic import BaseModel, Field

# 定义状态文件路径
STATE_PATH = os.path.join(os.getcwd(), 'application_state.json')

class SelectCameraState(BaseModel):
    selected_cameras: List[str] = Field(default_factory=list)

class CropState(BaseModel):
    crop_type: str = "Anchor"  # Anchor / Fixed
    bbox: List[int] = Field(default_factory=lambda: [0, 0, 100, 100])
    anchor_class: str = "default"
    anchor_position: List[int] = Field(default_factory=lambda: [50, 50])

class ProcessState(BaseModel):
    feature_extractor: str = "ResNet"
    embedding_processor: str = "PCA"
    outlier_detector: str = "IsolationForest"

class ApplicationState(BaseModel):
    camera_select_state: SelectCameraState = Field(default_factory=SelectCameraState)
    crop_state: CropState = Field(default_factory=CropState)
    process_state: ProcessState = Field(default_factory=ProcessState)
    # 可以在这里添加其他应用级别的状态

    class Config:
        validate_assignment = True # 开启赋值验证

# 示例：将Pydantic模型实例序列化并保存到文件
def save_state_to_json(state_model: ApplicationState, path: str = STATE_PATH):
    """将ApplicationState模型序列化为JSON并保存到文件。"""
    try:
        with open(path, 'w', encoding='utf-8') as f:
            f.write(state_model.model_dump_json(indent=2)) # indent=2 使JSON格式更易读
        print(f"应用状态已保存到: {path}")
    except IOError as e:
        print(f"保存状态文件失败: {e}")

# 示例使用
# app_state_instance = ApplicationState()
# # ... 修改 app_state_instance 的属性 ...
# save_state_to_json(app_state_instance)

import streamlit as st

def load_initial_state(path: str = STATE_PATH) -> ApplicationState:
    """从JSON文件加载应用状态，如果失败则返回默认状态。"""
    if os.path.exists(path):
        try:
            with open(path, 'r', encoding='utf-8') as f:
                data = json.load(f)
            return ApplicationState.model_validate(data) # Pydantic v2+ 使用 model_validate
        except (json.JSONDecodeError, FileNotFoundError, Exception) as e:
            st.warning(f"加载状态文件失败或文件损坏 ({e})，将使用默认状态。")
            return ApplicationState()
    else:
        st.info("状态文件不存在，将使用默认状态。")
        return ApplicationState()

# 在Streamlit应用启动时加载或初始化状态
if 'app_state' not in st.session_state:
    st.session_state.app_state = load_initial_state()

# 定义保存状态的回调函数
def on_state_change():
    """当Streamlit组件值改变时，更新session_state并保存应用状态。"""
    save_state_to_json(st.session_state.app_state)
    st.toast("应用状态已自动保存！", icon="✅")

# Streamlit应用示例
st.title("Streamlit 应用状态持久化演示")

# 1. 摄像头选择状态
st.header("摄像头选择")
selected_cameras_input = st.multiselect(
    "选择摄像头",
    options=["Camera A", "Camera B", "Camera C", "Camera D"],
    default=st.session_state.app_state.camera_select_state.selected_cameras,
    key="camera_select",
    on_change=on_state_change # 绑定回调函数
)
# 更新Pydantic模型
st.session_state.app_state.camera_select_state.selected_cameras = selected_cameras_input


# 2. 裁剪状态
st.header("裁剪设置")
crop_type_option = st.radio(
    "裁剪类型",
    options=["Anchor", "Fixed"],
    index=0 if st.session_state.app_state.crop_state.crop_type == "Anchor" else 1,
    key="crop_type",
    on_change=on_state_change
)
st.session_state.app_state.crop_state.crop_type = crop_type_option

bbox_values = st.slider(
    "边界框 (x, y, width, height)",
    min_value=0, max_value=500, value=st.session_state.app_state.crop_state.bbox,
    key="bbox_slider",
    on_change=on_state_change
)
st.session_state.app_state.crop_state.bbox = list(bbox_values) # slider返回tuple，需转为list


# 3. 处理状态
st.header("处理流程")
feature_extractor_option = st.selectbox(
    "特征提取器",
    options=["ResNet", "VGG", "EfficientNet"],
    index=["ResNet", "VGG", "EfficientNet"].index(st.session_state.app_state.process_state.feature_extractor),
    key="feature_extractor_select",
    on_change=on_state_change
)
st.session_state.app_state.process_state.feature_extractor = feature_extractor_option


# 显示当前状态（调试用）
st.sidebar.header("当前应用状态")
st.sidebar.json(st.session_state.app_state.model_dump()) # 显示Pydantic模型字典表示