python +pandas +flask +echarts 气象数据可视化

import json import pandas as pd from flask import Flask, jsonify, render_template, redirect, url_for, session, flash,request import pymysql from pyspark.sql.types import StringType from sqlalchemy import create_engine,text from forms import loginForm app = Flask(__name__) app.config['SECRET_KEY'] = '123456' @app.route("/") def index(): return render_template("index.html") @app.route("/echart1") def echart1(): # 连接到MySQL数据库 engine = create_engine('mysql+pymysql://root:[email protected]/tianqi') connection = engine.connect() # 从数据库中读取数据 df = pd.read_sql(sql=text('SELECT 城市,风向,风速,时间 FROM 24h where 城市="北京"'), con=connection) # 定义一个函数来将风向转换为文本 def get_wind_direction(x): if (x > 337.5) or (x <= 22.5): return "N" elif (x > 22.5) and (x <= 67.5): return "NE" elif (x > 67.5) and (x <= 112.5): return "E" elif (x > 112.5) and (x <= 157.5): return "SE" elif (x > 157.5) and (x <= 202.5): return "S" elif (x > 202.5) and (x <= 247.5): return "SW" elif (x > 247.5) and (x <= 292.5): return "W" elif (x > 292.5) and (x <= 337.5): return "NW" else: return None df['风向'] = df['风向'].apply(get_wind_direction) df['时间'] = df['时间'] print(df) # 将 '时间' 列转换为时间戳 df['时间'] = pd.to_datetime(df['时间']) df['时间'] = df['时间'].apply(lambda x: x.timestamp()) # 将三列数据组成一个数组 array = df[['时间', '风速', '风向']].to_numpy() print(array) # 渲染模板并传入图表对象 return jsonify(array.tolist()) @app.route("/echart2") def echart2(): # 连接到MySQL数据库 engine = create_engine('mysql+pymysql://root:[email protected]/tianqi') query = "SELECT 温度,相对湿度 FROM 24h WHERE 城市='北京'" df = pd.read_sql(sql=text(query), con=engine.connect()) array = df.values.tolist() print(array) return jsonify(array) @app.route("/echart3") def echart3(): # 连接到MySQL数据库 engine = create_engine('mysql+pymysql://root:[email protected]/tianqi') # 查询SQL语句 query = "SELECT 城市,日期,最高温度,最低温度 FROM tempchart where 城市='北京'" # 使用pandas读取数据 df = pd.read_sql(sql=text(query), con=engine.connect()) # 转换日期格式 df['日期'] = pd.to_datetime(df['日期'], format='%Y-%m-%d') # 筛选日期 df = df[df['日期'] < '2023-04-20'] # 提取数据 date = df['日期'].apply(lambda x: x.strftime('%Y-%m-%d')).tolist() array1 = df['最高温度'].tolist() array2 = df['最低温度'].tolist() return jsonify({'date':date,'array1':array1,'array2':array2}) @app.route("/echart4") def echart4(): # 连接到MySQL数据库 engine = create_engine('mysql+pymysql://root:[email protected]/tianqi') # 查询SQL语句 query = "SELECT 城市,日期,白天温度,夜晚温度 FROM predict where 城市='北京'" # 使用pandas读取数据 df = pd.read_sql(sql=text(query), con=engine.connect()) df['日期'] = pd.to_datetime(df['日期'], format='%Y-%m-%d') date = df['日期'].tolist() array1 = df['白天温度'].tolist() array2 = df['夜晚温度'].tolist() return jsonify({'date': date, 'array1': array1, 'array2': array2}) @app.route("/echart5") def echart5(): # 连接到MySQL数据库 engine = create_engine('mysql+pymysql://root:[email protected]/tianqi') # 从MySQL数据库中获取数据 query = "SELECT 城市,降雨量 FROM rain24h" df = pd.read_sql(sql=text(query), con=engine.connect()) # 选取并显示前十个降雨量最高的城市 top_10_rainfall = df.sort_values('降雨量', ascending=False).head(10) print(top_10_rainfall) # 将城市和降雨量分别存储在列表中 city = top_10_rainfall['城市'].tolist() array = top_10_rainfall['降雨量'].tolist() return jsonify({'array':array,'city':city}) @app.route("/echart6") def echart6(): # 连接到MySQL数据库 engine = create_engine('mysql+pymysql://root:[email protected]/tianqi') # 从MySQL数据库中获取数据 query = "SELECT 城市,降雨量 FROM rain1h" df = pd.read_sql(sql=text(query), con=engine.connect()) # 选取并显示前十个降雨量最高的城市 top_10_rainfall = df.sort_values('降雨量', ascending=False).head(10) print(top_10_rainfall) # 将城市和降雨量分别存储在列表中 city = top_10_rainfall['城市'].tolist() array = top_10_rainfall['降雨量'].tolist() return jsonify({'array':array,'city':city}) import cpca # 编写将城市名称转换为省份名称的函数 def get_province(city): location = cpca.transform([city]) return location.iloc[0]["省"] @app.route("/echart7") def echart7(): # 连接到MySQL数据库 engine = create_engine('mysql+pymysql://root:[email protected]/tianqi') # 从MySQL数据库中获取数据 query = "SELECT 城市,温度 FROM day" df = pd.read_sql(sql=text(query), con=engine.connect()) print(df) # 将 get_province() 函数应用于城市列，并添加省份列 df['province'] = df['城市'].apply(get_province) # 根据省份列进行分组，并计算每个省份的平均温度 df = df.groupby('province')['温度'].mean().reset_index() df.columns = ['province', 'avg(温度)'] # 将结果转换为键值对列表 array = df.to_dict('records') array = [{'name': i['province'], 'value': round(i['avg(温度)'], 2)} for i in array] print(array) return jsonify(array) if __name__ == "__main__": app.run(debug=True,port=5004)