88-python电网可视化项目-8-1

1. 可视化图表

# -*- coding: utf-8 -*-
"""
@Project : 01-python-learn
@File    : power_grid_system.py
@IDE     : PyCharm
@Author  : 刘庆东
@Date    : 2025/9/24 14:01核心业务类:刚开始认为技术最难其实 业务最难"""
# 1.导入各种包[模块]
from  database import db#数据库部分
import bcrypt #加密
"""
datetime 表示当前具体的时刻 2025年9月24日 下午02:06
timedelta 代表的是两个时间点之间的差值 比如说 1天  4小时23分钟后进行休息
"""
from datetime import datetime,timedelta
import logging #日志相关from pyecharts import options as opts"""
Line 上课学过使用场景:股票走势、气温变化、销售额度Bar 上课学过使用场景：月业绩排名、各个地区的人口数量Pie 大饼图使用场景:市场份额占比、预算分配Gauge 仪表盘使用场景: KPI、进度、完成率Grid 布局网格使用场景: 将一个折线图 和一个柱状图并排显示 ! """
from pyecharts.charts import Line,Bar,Pie,Gauge,Gridfrom pyecharts.globals import ThemeType#主题class PowerGridSystem:def __init__(self):#构造方法self.current_user=None #当前的用户#登录功能def login(self,username,password):try:query="""select user_id,username,password_hash,role,full_name from userswhere username=%s and STATUS='active'"""#查询返回结果result= db.execute_query(query,(username,))if result:#获取的是第一行数据user =result[0]if bcrypt.checkpw(password.encode('utf-8'),user['password_hash'].encode('utf-8')):#if password=='admin':self.current_user=user #将用户信息赋值给 current_user#返回之前进行日志的记录工作self.log_action("login","user",user['user_id'],f"用户{username}登录")return True#代表登录成功了except Exception as e:logging.error(f"登录失败{e}")return False#所有的操作 都需要进行日志的记录def log_action(self,action, target_type, target_id, details):query="""insert into system_logs (user_id, action, target_type, target_id, details)values (%s,%s,%s,%s,%s)"""#获取user_iduser_id= self.current_user['user_id'] if self.current_user else Nonedb.execute_update(query,(user_id,action,target_type,target_id,details))#登出的方法def login_out(self):if self.current_user:#用户不为空# 先进行日志的记录self.log_action("logout",'user',self.current_user['user_id'],f"用户{self.current_user['username']}登出")self.current_user = None#给当前的登录用置空#获取变电站信息def get_substations(self):#操作数据库的sql语句  任务query="""select s.*,u.full_name as operator_name from substations sleft join users u on s.operator_id = u.user_idorder by s.substation_name;    """return  db.execute_query(query)#通过变电站的id获取设备信息def get_equipment_by_substation_id(self,substation_id):sql="""select e.*,s.substation_name from equipment e join substations s on e.substation_id = s.substation_idwhere e.substation_id=%sorder by  e.equipment_name"""#执行通用的方法return db.execute_query(sql,(substation_id,))#获取工程师def get_users(self):query="""select user_id,username,password_hash,role,full_name from users"""return db.execute_query(query)"""order_id=system.create_work_order(titile=titile,#标题description=description,#描述equipment_id=selected_eq['equipment_id'],#关联设备IDsubstation_id=selected_sub['substation_id'],#关联变电站IDassigned_to=assigned_to,#assigned_topriority=priority,scheduled_start = datetime.now() + timedelta(hours=1),scheduled_end = datetime.now() + timedelta(hours=3))"""#创建工单def create_work_order(self,title,description,equipment_id,substation_id,assigned_to,priority,scheduled_start,scheduled_end):query="""insert into work_orders (title,description,equipment_id,substation_id,assigned_to,priority,scheduled_start,scheduled_end)values (%s,%s,%s,%s,%s,%s,%s,%s)"""return db.execute_update(query,(title,description,equipment_id,substation_id,assigned_to,priority,scheduled_start,scheduled_end))#获取最新的电力数据  2025-09-26 14:09def latest_power_data(self):query="""select e.equipment_id,e.equipment_name,p.timestamp,p.voltage_kV,p.current_A,p.active_power_kW from equipment ejoin power_data p on e.equipment_id=p.equipment_idwhere p.timestamp=(select max(timestamp) from power_datawhere equipment_id=e.equipment_id)order by e.equipment_name"""return db.execute_query(query)# 查询我的工单def get_work_orders(self,user_id):# 这个sql只是拿到了 设备的id  如果想要获取设备的名字 需要做多表关联查询query="""selectorder_id,title,equipment_id,status,priority,created_atfrom work_orders where assigned_to=%s"""return db.execute_query(query, (user_id,))#生成电压趋势图 2025-10-09PM 14:28   1 设备的编号 2.查看多少小时的数据 方法二def create_voltage_chart(self,equipment_id,hours=24):"""生成电压趋势图"""data = self.get_history_data(equipment_id, hours)if not data:return Nonetimestamps = [d['timestamp'].strftime('%H:%M') for d in data]voltages = [d['voltage_kV'] for d in data]line = (Line(init_opts=opts.InitOpts(theme=ThemeType.LIGHT, width="800px", height="400px")).add_xaxis(timestamps).add_yaxis("电压 (kV)", voltages, is_smooth=True).set_global_opts(title_opts=opts.TitleOpts(title="电压趋势"),tooltip_opts=opts.TooltipOpts(trigger="axis", axis_pointer_type="cross"),xaxis_opts=opts.AxisOpts(type_="category"),yaxis_opts=opts.AxisOpts(type_="value", name="kV"),))return line# 2025-10-09PM 14:28    这个方法要先写  方法一def get_history_data(self,equipment_id,hours=24):start_time = datetime.now() - timedelta(hours=hours)query = """SELECT timestamp, voltage_kV, current_A, active_power_kW, power_factor, frequency_HzFROM power_dataWHERE equipment_id = %s AND timestamp >= %sORDER BY timestamp"""return db.execute_query(query, (equipment_id, start_time))#测试
system= PowerGridSystem()