Python项目93:智慧城市运行管理中心,大屏可视化(dash+plotly+pandas)

# 安装依赖：pip install dash plotly pandas numpy# -*- coding: utf-8 -*-# @Author : 小红牛# 微信公众号：wdPythonimport dashfrom dash import dcc, html, Input, Outputimport plotly.graph_objects as goimport plotly.express as pximport pandas as pdimport numpy as npfrom datetime import datetime, timedeltaimport dash_bootstrap_components as dbc# 初始化Dash应用app = dash.Dash(__name__, external_stylesheets=[dbc.themes.DARKLY])app.title = "智慧城市运营管理中心"# 生成模拟数据def generate_data():    np.random.seed(42)    # 交通流量数据    hours = list(range(24))    traffic_data = []    for i in range(7):  # 一周的数据        base = np.random.randint(200, 500)        for h in hours:            if 7 <= h <= 9 or 17 <= h <= 19:  # 高峰时段                traffic = base + np.random.randint(300, 600)            elif 22 <= h <= 5:  # 夜间                traffic = base + np.random.randint(0, 100)            else:  # 平峰时段                traffic = base + np.random.randint(100, 300)            traffic_data.append({                'date': (datetime.now() - timedelta(days=6-i)).strftime('%Y-%m-%d'),                'hour': h,                'traffic_volume': traffic            })    # 空气质量数据    districts = ['市中心', '高新区', '工业区', '居民区', '开发区']    air_quality_data = []    for district in districts:        for i in range(30):  # 30天数据            date = (datetime.now() - timedelta(days=29-i)).strftime('%Y-%m-%d')            pm25 = np.random.normal(30, 10)            pm10 = pm25 + np.random.normal(10, 5)            no2 = np.random.normal(20, 8)            air_quality_data.append({                'district': district,                'date': date,                'PM2.5': max(5, pm25),                'PM10': max(10, pm10),                'NO2': max(5, no2)            })    # 能源消耗数据    energy_data = []    energy_types = ['居民用电', '工业用电', '商业用电', '公共设施']    for i in range(12):  # 12个月        month = i + 1        for energy_type in energy_types:            if energy_type == '居民用电':                consumption = np.random.normal(500, 50) + 50 * np.sin(month * 0.5)            elif energy_type == '工业用电':                consumption = np.random.normal(800, 100) + 30 * np.cos(month * 0.3)            elif energy_type == '商业用电':                consumption = np.random.normal(600, 80) + 40 * np.sin(month * 0.4)            else:                consumption = np.random.normal(300, 30)            energy_data.append({                'month': month,                'energy_type': energy_type,                'consumption': max(100, consumption)            })    # 人口热力图数据    population_data = []    for lat in np.linspace(30.6, 30.8, 20):        for lon in np.linspace(104.0, 104.2, 20):            # 模拟市中心人口密度高            center_dist = np.sqrt((lat - 30.7)**2 + (lon - 104.1)**2)            density = max(10, 1000 * np.exp(-center_dist * 10))            population_data.append({                'lat': lat,                'lon': lon,                'density': density + np.random.normal(0, 100)            })    return {        'traffic': pd.DataFrame(traffic_data),        'air_quality': pd.DataFrame(air_quality_data),        'energy': pd.DataFrame(energy_data),        'population': pd.DataFrame(population_data)    }# 生成数据data = generate_data()# 应用布局app.layout = dbc.Container(    fluid=True,    children=[        # 标题栏        dbc.Row([            dbc.Col([                html.H1("智慧城市运营管理中心", className="text-center mb-4 mt-3"),                html.P("实时监控城市运行状态与关键指标", className="text-center text-muted mb-4")            ])        ]),        # 第一行：KPI指标        dbc.Row([            dbc.Col([                dbc.Card([                    dbc.CardBody([                        html.H4("交通指数", className="card-title"),                        html.H2("78.5", className="card-text text-success"),                        html.P("较昨日↑2.3%", className="card-text text-muted")                    ])                ], className="text-center shadow")            ], width=3),            dbc.Col([                dbc.Card([                    dbc.CardBody([                        html.H4("空气质量指数", className="card-title"),                        html.H2("42", className="card-text text-info"),                        html.P("优良", className="card-text text-muted")                    ])                ], className="text-center shadow")            ], width=3),            dbc.Col([                dbc.Card([                    dbc.CardBody([                        html.H4("能耗指数", className="card-title"),                        html.H2("65.2", className="card-text text-warning"),                        html.P("较上周↓1.5%", className="card-text text-muted")                    ])                ], className="text-center shadow")            ], width=3),            dbc.Col([                dbc.Card([                    dbc.CardBody([                        html.H4("安全指数", className="card-title"),                        html.H2("94.7", className="card-text text-success"),                        html.P("较上月↑0.8%", className="card-text text-muted")                    ])                ], className="text-center shadow")            ], width=3)        ], className="mb-4"),        # 第二行：主要图表        dbc.Row([            # 交通流量图            dbc.Col([                dbc.Card([                    dbc.CardHeader("交通流量实时监控"),                    dbc.CardBody([                        dcc.Graph(id='traffic-chart', style={'height': '300px'})                    ])                ], className="shadow")            ], width=6, className="mb-4"),            # 空气质量图            dbc.Col([                dbc.Card([                    dbc.CardHeader("区域空气质量"),                    dbc.CardBody([                        dcc.Graph(id='air-quality-chart', style={'height': '300px'})                    ])                ], className="shadow")            ], width=6, className="mb-4")        ]),        # 第三行：更多图表        dbc.Row([            # 能源消耗图            dbc.Col([                dbc.Card([                    dbc.CardHeader("能源消耗分析"),                    dbc.CardBody([                        dcc.Graph(id='energy-chart', style={'height': '300px'})                    ])                ], className="shadow")            ], width=6, className="mb-4"),            # 人口热力图            dbc.Col([                dbc.Card([                    dbc.CardHeader("人口密度热力图"),                    dbc.CardBody([                        dcc.Graph(id='population-chart', style={'height': '300px'})                    ])                ], className="shadow")            ], width=6, className="mb-4")        ]),        # 第四行：时间选择器        dbc.Row([            dbc.Col([                dbc.Card([                    dbc.CardBody([                        html.P("选择时间范围:", className="mb-2"),                        dcc.DatePickerRange(                            id='date-picker',                            min_date_allowed=(datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d'),                            max_date_allowed=datetime.now().strftime('%Y-%m-%d'),                            start_date=(datetime.now() - timedelta(days=7)).strftime('%Y-%m-%d'),                            end_date=datetime.now().strftime('%Y-%m-%d'),                            display_format='YYYY-MM-DD',                            className="mb-3"                        ),                        html.P("更新时间: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S"),                                id='update-time', className="text-muted text-end")                    ])                ])            ], width=12)        ])    ])# 回调函数更新图表@app.callback(    [Output('traffic-chart', 'figure'),     Output('air-quality-chart', 'figure'),     Output('energy-chart', 'figure'),     Output('population-chart', 'figure'),     Output('update-time', 'children')],    [Input('date-picker', 'start_date'),     Input('date-picker', 'end_date')])def update_charts(start_date, end_date):    # 更新时间显示    update_time = "更新时间: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S")    # 1. 交通流量图    traffic_fig = go.Figure()    # 获取最近一天的数据    latest_date = data['traffic']['date'].max()    day_traffic = data['traffic'][data['traffic']['date'] == latest_date]    traffic_fig.add_trace(go.Scatter(        x=day_traffic['hour'],        y=day_traffic['traffic_volume'],        mode='lines+markers',        name='交通流量',        line=dict(color='#00b4d8', width=3),        fill='tozeroy',        fillcolor='rgba(0, 180, 216, 0.2)'    ))    traffic_fig.update_layout(        template='plotly_dark',        plot_bgcolor='rgba(0,0,0,0)',        paper_bgcolor='rgba(0,0,0,0)',        xaxis_title="时间 (时)",        yaxis_title="车流量 (辆/小时)",        margin=dict(l=40, r=20, t=20, b=40)    )    # 2. 空气质量图    latest_air_date = data['air_quality']['date'].max()    latest_air_data = data['air_quality'][data['air_quality']['date'] == latest_air_date]    air_fig = go.Figure()    for district in latest_air_data['district'].unique():        district_data = latest_air_data[latest_air_data['district'] == district]        air_fig.add_trace(go.Bar(            x=['PM2.5', 'PM10', 'NO2'],            y=[district_data['PM2.5'].mean(), district_data['PM10'].mean(), district_data['NO2'].mean()],            name=district        ))    air_fig.update_layout(        template='plotly_dark',        plot_bgcolor='rgba(0,0,0,0)',        paper_bgcolor='rgba(0,0,0,0)',        barmode='group',        xaxis_title="污染物",        yaxis_title="浓度 (μg/m³)",        margin=dict(l=40, r=20, t=20, b=40)    )    # 3. 能源消耗图    energy_fig = px.line(        data['energy'],         x='month',         y='consumption',         color='energy_type',        markers=True    )    energy_fig.update_layout(        template='plotly_dark',        plot_bgcolor='rgba(0,0,0,0)',        paper_bgcolor='rgba(0,0,0,0)',        xaxis_title="月份",        yaxis_title="消耗量 (万kWh)",        legend_title="能源类型",        margin=dict(l=40, r=20, t=20, b=40)    )    # 4. 人口热力图    population_fig = go.Figure(go.Densitymapbox(        lat=data['population']['lat'],        lon=data['population']['lon'],        z=data['population']['density'],        radius=15,        colorscale='Viridis'    ))    population_fig.update_layout(        mapbox_style="carto-darkmatter",        mapbox_center={"lat": 30.7, "lon": 104.1},        mapbox_zoom=10,        margin=dict(l=20, r=20, t=20, b=20),        height=300    )    return traffic_fig, air_fig, energy_fig, population_fig, update_timeif __name__ == '__main__':    app.run(debug=True, port=8050)