添加绘图逻辑;

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 = {
+SemaMap = {
-  "startTimestamp": 1732032000000,
+    'apt_temp':      ('0305117001', 'adapter', "设备温度"),
-  "endTimestamp": 1732291199000,
+    'apt_volt_in':   ('0305118001', 'adapter', "输入电压"),
-  "deviceCode": "TTE0102DX2406240497",
+    'apt_curr_in':   ('0305119001', 'adapter', "输入电流"),
-  "mid": "0305120001",
+    'apt_volt_out':  ('0305120001', 'adapter', "输出电压"),
+    'apt_curr_out':  ('0305121001', 'adapter', "输出电流"),
+    'apt_power_out': ('0305122001', 'adapter', "输出功率"),
+}
+
+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()
 
-# 1. 读取本地HTML文件
+def adapter_status_graphs(device_id, times: tuple[int, int]):
-file_path = Path(r'D:\WorkingProject\LightStackAdapter\Log\设备测试数据记录-铁塔主站\南和县牧村\Untitled-1.html')
+    """ 获取数据, 绘制图表 """
-html_content = file_path.read_text()
+    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)
 
+    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)
 
-# 2. 解析HTML文件
+    # 图表绘制
-soup = BeautifulSoup(html_content, 'html.parser')
+    chart_apt(table_apt)
 
-# 3. 找到表格元素
+def chart_apt(table_apt):
-table = soup.find_all('table')  # 假设页面中只有一个表格，如果有多个表格，可能需要进一步筛选
+    """ 绘制适配器信息图表 """
+    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'))
 
-# 4. 提取表格数据
+    # 设置x轴的主要刻度定位器为自动日期定位器，这使得x轴上的刻度根据数据自动选择最合适的日期格式
-data = []
+    ax1.xaxis.set_major_locator(mdates.AutoDateLocator())
-headers = []
 
-# 提取表头
+    ax2 = ax1.twinx()
-header_row = table.find('thead').find('tr')
+    # 绘制斜线阴影
-for header in header_row.find_all('th'):
+    for i in range(len(table_apt) - 1):
-    headers.append(header.text.strip())
+        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)
 
-# 提取数据行
+    lines, labels    = ax1.get_legend_handles_labels()
-for row in table.find('tbody').find_all('tr'):
+    shadows, shadow_labels = ax2.get_legend_handles_labels()
-    row_data = []
+    ax1.legend(lines + shadows, labels + shadow_labels, loc='upper left')
-    for cell in row.find_all(['td', 'th']):
-        row_data.append(cell.text.strip())
-    data.append(row_data)
 
-# 5. 将数据保存为DataFrame
+    ax1.set_title('Device Data Visualization')
-df = pd.DataFrame(data, columns=headers)
+    ax1.set_xlabel('Time')
+    ax1.set_ylabel('Voltage (V)')
+    ax2.set_ylabel('Power (W)')
 
-# 6. 将DataFrame保存为CSV文件
+    plt.show()
-output_file = 'extracted_table.csv'
-df.to_csv(output_file, index=False, encoding='utf-8')
 
-print(f'表格数据已成功提取并保存到 {output_file}')
+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