diff --git a/source/data_analysis.py b/source/data_analysis.py index e5fc380..5c5ec7c 100644 --- a/source/data_analysis.py +++ b/source/data_analysis.py @@ -1,7 +1,12 @@ +import time +import datetime import requests import pandas as pd from pathlib import Path from bs4 import BeautifulSoup +import pandas as pd +import matplotlib.pyplot as plt +import matplotlib.dates as mdates API_URL = "https://energy-iot.chinatowercom.cn/api/device/device/historyPerformance" @@ -25,49 +30,107 @@ headers = { "sec-ch-ua-platform": "Windows", } -body = { - "startTimestamp": 1732032000000, - "endTimestamp": 1732291199000, - "deviceCode": "TTE0102DX2406240497", - "mid": "0305120001", - "businessType": "7", - "pageNum": 2, - "pageSize": 5, - "total": 0 +SemaMap = { + 'apt_temp': ('0305117001', 'adapter', "设备温度"), + 'apt_volt_in': ('0305118001', 'adapter', "输入电压"), + 'apt_curr_in': ('0305119001', 'adapter', "输入电流"), + 'apt_volt_out': ('0305120001', 'adapter', "输出电压"), + 'apt_curr_out': ('0305121001', 'adapter', "输出电流"), + 'apt_power_out': ('0305122001', 'adapter', "输出功率"), } -# 1. 读取本地HTML文件 -file_path = Path(r'D:\WorkingProject\LightStackAdapter\Log\设备测试数据记录-铁塔主站\南和县牧村\Untitled-1.html') -html_content = file_path.read_text() +def get_history_data(device_id, data_type, times: tuple[int, int]): + """ 读取信号量历史数据, 返回接口json数据 """ + body = { + "startTimestamp": times[0], + "endTimestamp": times[1], + "deviceCode": f"{device_id}", + "mid": f"{data_type}", + "businessType": "7", + "pageNum": 2, + "pageSize": 5, + "total": 0 + } + req = requests.post(API_URL, data=body, headers=headers) + return req.json() +def adapter_status_graphs(device_id, times: tuple[int, int]): + """ 获取数据, 绘制图表 """ + data_volt_in = get_history_data(device_id, SemaMap["apt_volt_in"], times) + data_curr_in = get_history_data(device_id, SemaMap["apt_curr_in"], times) + data_volt_out = get_history_data(device_id, SemaMap["apt_volt_out"], times) + data_power_out = get_history_data(device_id, SemaMap["apt_power_out"], times) -# 2. 解析HTML文件 -soup = BeautifulSoup(html_content, 'html.parser') + data_apt = [] + for item in zip(data_volt_in['data'], data_volt_out['data'], data_curr_in['data'], data_power_out['data']): + print(item) + piont_time = time.mktime(time.strptime(item[0]['collectTime'], r"%Y-%m-%d %H:%M:%S")) + point_apt = { + 'time': int(piont_time), + 'volt_in': item[0]['value'], + 'volt_out': item[1]['value'], + 'curr_in': item[2]['value'], + 'power_out': item[3]['value'], + } + data_apt.append(point_apt) + table_apt = pd.DataFrame(data_apt) -# 3. 找到表格元素 -table = soup.find_all('table') # 假设页面中只有一个表格,如果有多个表格,可能需要进一步筛选 + # 图表绘制 + chart_apt(table_apt) -# 4. 提取表格数据 -data = [] -headers = [] +def chart_apt(table_apt): + """ 绘制适配器信息图表 """ + fig, ax1 = plt.subplots(figsize=(12, 6)) + ax1.plot(table_apt['time'], table_apt['volt_in'], color='green', label='Input Voltage') + ax1.plot(table_apt['time'], table_apt['volt_out'], color='red', label='Output Voltage') + # ax1.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d %H:%M:%S')) -# 提取表头 -header_row = table.find('thead').find('tr') -for header in header_row.find_all('th'): - headers.append(header.text.strip()) + # 设置x轴的主要刻度定位器为自动日期定位器,这使得x轴上的刻度根据数据自动选择最合适的日期格式 + ax1.xaxis.set_major_locator(mdates.AutoDateLocator()) -# 提取数据行 -for row in table.find('tbody').find_all('tr'): - row_data = [] - for cell in row.find_all(['td', 'th']): - row_data.append(cell.text.strip()) - data.append(row_data) + ax2 = ax1.twinx() + # 绘制斜线阴影 + for i in range(len(table_apt) - 1): + ax1.fill_between( + [table_apt['time'].iloc[i], table_apt['time'].iloc[i + 1]], + [table_apt['power_out'].iloc[i], table_apt['power_out'].iloc[i + 1]], + color='red', alpha=0.5) -# 5. 将数据保存为DataFrame -df = pd.DataFrame(data, columns=headers) + lines, labels = ax1.get_legend_handles_labels() + shadows, shadow_labels = ax2.get_legend_handles_labels() + ax1.legend(lines + shadows, labels + shadow_labels, loc='upper left') -# 6. 将DataFrame保存为CSV文件 -output_file = 'extracted_table.csv' -df.to_csv(output_file, index=False, encoding='utf-8') + ax1.set_title('Device Data Visualization') + ax1.set_xlabel('Time') + ax1.set_ylabel('Voltage (V)') + ax2.set_ylabel('Power (W)') -print(f'表格数据已成功提取并保存到 {output_file}') \ No newline at end of file + plt.show() + +def sim_data_apt(times:tuple[int, int]): + """ 模拟数据 """ + t_start = time.mktime(time.strptime(times[0], r"%Y-%m-%d %H:%M:%S")) + t_end = time.mktime(time.strptime(times[1], r"%Y-%m-%d %H:%M:%S")) + count_data = (t_end - t_start) / (10 * 60) + time_list = range(int(t_start), int(t_end), 20 * 60) + time_list = tuple(map(lambda x: time.strftime(r"%Y-%m-%d %H:%M:%S", time.localtime(x)), time_list)) + data = { + 'time': time_list, + 'volt_in': 10 + 10 * np.random.random(len(time_list)), + 'curr_in': 1 + 2 * np.random.random(len(time_list)), + 'volt_out': 54 + 2 * np.random.random(len(time_list)), + } + data['power_out'] = tuple(map(lambda x: x[0] * x[1], zip(data['volt_in'],data['curr_in']))) + + return pd.DataFrame(data) + +if __name__=='__main__': + import numpy as np + + data = sim_data_apt(('2024-10-1 00:00:00', '2024-10-1 12:00:00')) + + chart_apt(data) + + plt.waitforbuttonpress() + + pass \ No newline at end of file