AirQuility/AirQuilitySys/DayAir.py

import logging
import pymysql
import subprocess
import numpy as np
from matplotlib import pyplot as plt
from tkinter import Tk
from pyecharts import options as _opts
from pyecharts.charts import Bar

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def get_data(db_config):
    """
    从数据库中获取ptime和ci数据。
    :param db_config: 数据库配置字典
    :return: 包含ptime和ci的元组列表
    """
    ptime_list = []  # 存储ptime数据的列表
    ci_list = []  # 存储ci数据的列表
    so_list = []


    try:
        with pymysql.connect(**db_config) as conn:
            logging.info("数据库连接成功")
            with conn.cursor() as cursor:
                # 使用单个查询以提高性能
                sql_ptime = "SELECT ptime FROM May"
                cursor.execute(sql_ptime)
                data_ptime = cursor.fetchall()
                #print(data_ptime)
                # 处理查询结果
                #for row in data:
                    #for row in range(len(data)):
                        #if row == 0:
                            #continue
                   # print(row)

                ptime_list = [row['ptime'] for row in data_ptime if 'ptime' in row]
                #row_list=[val for sublist in row for val in sublist]

                sql_ci="SELECT ci FROM may"
                cursor.execute(sql_ci)
                data_ci=cursor.fetchall()
                #print(data_ci)
                ci_list = [row['ci'] for row in data_ci if 'ci' in row]

                sql_so = "SELECT so FROM may"
                cursor.execute(sql_so)
                data_so = cursor.fetchall()
                so_list = [row['so'] for row in data_so if 'so' in row]

                sql_no = "SELECT no FROM may"
                cursor.execute(sql_no)
                data_no = cursor.fetchall()
                no_list = [row['no'] for row in data_no if 'no' in row]

                sql_pm = "SELECT pm FROM may"
                cursor.execute(sql_pm)
                data_pm = cursor.fetchall()
                pm_list = [row['pm'] for row in data_pm if 'pm' in row]


            return ptime_list, ci_list, so_list, no_list, pm_list
    except pymysql.err.OperationalError as e:
        logging.error(f"数据库操作失败: {e}")
        # 根据具体场景，可以选择返回空列表、抛出异常或返回特定错误码
        return [], [], [], [], []
    #except Exception as e:
        #logging.error(f"未知错误: {e}")
        # 同上，根据具体场景进行处理
        #return [], []
def visualize_data(ptime_list, ci_list, so_list, no_list, pm_list):
    """
    可视化数据。
    :param ptime_list: ptime数据列表
    :param ci_list: ci数据列表
    :param column_to_plot: 需要绘制的列名
    """
    if not ptime_list or not ci_list or not so_list or not no_list or not pm_list:
        logging.error("数据为空，无法进行可视化")
        return
    #y_list=[]
    #for i in range(4,8):
     #   y_list.append(i)
    plt.figure(figsize=(15,10),dpi=80)

    plt.xlabel('Ptime')

    plt.title('CI over time')
    plt.plot(ptime_list, ci_list,label='CI')
    plt.plot(ptime_list, so_list,label='SO2')
    plt.plot(ptime_list, no_list, label='NO2')
    plt.plot(ptime_list, pm_list, label='PM2.5')
    plt.legend()
    plt.show()



if __name__ == "__main__":
    db_config = {
        "host": '127.0.0.1',
        "user": "root",
        "password": 'mysql>hyx123',
        "db": 'airquility',
        "charset": 'utf8',
        "cursorclass": pymysql.cursors.DictCursor
    }

    ptime_list, ci_list,so_list ,no_list, pm_list= get_data(db_config)
    visualize_data(ptime_list, ci_list, so_list, no_list, pm_list)

    print(pymysql.__version__)

'''
    window= Tk()
    window.title("每日空气质量")
    window.geometry("300x200")
#Label(window, text="图表已生成为air_quality_index.html").pack()

    window.mainloop()
    '''