如何用Python从AWS Timestream提取数据并通过PySimpleGUI实现温湿度可视化？

一、从Timestream拉取数据的核心流程 #

1. 配置AWS访问权限 #

用Python的boto3库连接Timestream，先搞定权限配置：

• 本地开发：在~/.aws/credentials文件填入access key和secret key，或者设置AWS_ACCESS_KEY_ID、AWS_SECRET_ACCESS_KEY环境变量
• AWS资源上运行（如EC2、EKS）：给实例/pod绑定有权限访问Timestream Query的IAM角色即可

2. 编写Timestream查询SQL #

Timestream支持标准SQL，查温湿度的示例语句（替换成你的库名和表名）：

SELECT time, temperature, humidity 
FROM "your_db_name"."your_table_name" 
WHERE time > ago(1h)  -- 拉取最近1小时数据，可按需调整时间范围
ORDER BY time ASC

3. 执行查询并解析结果 #

调用boto3客户端执行查询，把返回的结构化数据转成Python可直接使用的列表：

import boto3
from datetime import datetime

# 初始化Timestream查询客户端
client = boto3.client('timestream-query')
DB_NAME = "你的数据库名"
TABLE_NAME = "你的表名"

def fetch_temp_humi_data():
    query = f"""
        SELECT time, temperature, humidity 
        FROM "{DB_NAME}"."{TABLE_NAME}" 
        WHERE time > ago(1h)
        ORDER BY time ASC
    """
    response = client.query(QueryString=query)
    
    times, temps, humids = [], [], []
    # 跳过表头行，遍历数据行
    for row in response['Rows'][1:]:
        data = row['Data']
        # 转换时间格式与数值类型
        times.append(datetime.fromisoformat(data[0]['ScalarValue'].replace('Z', '+00:00')))
        temps.append(float(data[1]['ScalarValue']))
        humids.append(float(data[2]['ScalarValue']))
    
    return times, temps, humids

二、PySimpleGUI实时可视化实现 #

用matplotlib绘图并嵌入PySimpleGUI窗口，支持手动/自动刷新：

1. 安装依赖：pip install PySimpleGUI matplotlib boto3
2. 完整代码：

import PySimpleGUI as sg
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg

# 先粘贴上面的fetch_temp_humi_data函数

# 配置GUI布局
sg.theme('DefaultNoMoreNagging')
layout = [
    [sg.Text('温湿度实时监控', font=('微软雅黑', 16))],
    [sg.Canvas(key='-PLOT_CANVAS-', size=(800, 500))],
    [sg.Button('手动刷新'), sg.Button('退出'), sg.Checkbox('自动刷新(10秒)', default=True, key='-AUTO_REFRESH-')]
]

window = sg.Window('Timestream监控工具', layout, finalize=True)

# 初始化matplotlib图表
fig, (ax_temp, ax_humi) = plt.subplots(2, 1, figsize=(8, 5), sharex=True)
canvas = FigureCanvasTkAgg(fig, window['-PLOT_CANVAS-'].TKCanvas)
canvas.draw()
canvas.get_tk_widget().pack(fill='both', expand=1)

def update_chart():
    times, temps, humids = fetch_temp_humi_data()
    if not times:
        sg.popup('暂无数据', auto_close_duration=2)
        return
    
    # 清空旧图并重新绘制
    ax_temp.clear()
    ax_humi.clear()
    
    ax_temp.plot(times, temps, color='#ff4444', linewidth=2, label='温度(℃)')
    ax_temp.set_ylabel('温度')
    ax_temp.legend()
    ax_temp.grid(alpha=0.3)
    
    ax_humi.plot(times, humids, color='#33b5e5', linewidth=2, label='湿度(%)')
    ax_humi.set_xlabel('时间')
    ax_humi.set_ylabel('湿度')
    ax_humi.legend()
    ax_humi.grid(alpha=0.3)
    
    # 自动旋转时间标签避免重叠
    plt.setp(ax_humi.get_xticklabels(), rotation=45, ha='right')
    fig.tight_layout()
    canvas.draw()

# 初始加载数据
update_chart()

# 事件循环
while True:
    event, values = window.read(timeout=10000)  # 10秒超时触发自动刷新
    if event in (sg.WIN_CLOSED, '退出'):
        break
    if event == '手动刷新' or (values['-AUTO_REFRESH-'] and event == sg.TIMEOUT_EVENT):
        update_chart()

window.close()

三、可行替代方案 #

1. Streamlit快速搭建界面 #

不想写复杂GUI逻辑的话，Streamlit是首选，几行代码搞定实时可视化：

import streamlit as st
import matplotlib.pyplot as plt

st.title('Timestream温湿度监控')

# 自动刷新开关
auto_refresh = st.checkbox('自动刷新', value=True)

if auto_refresh or st.button('刷新数据'):
    times, temps, humids = fetch_temp_humi_data()
    fig, (ax1, ax2) = plt.subplots(2,1, figsize=(8,4))
    ax1.plot(times, temps, 'r-', label='温度')
    ax2.plot(times, humids, 'b-', label='湿度')
    st.pyplot(fig)

运行命令：streamlit run your_script.py，浏览器打开即可查看。

2. Grafana直接对接Timestream #

完全无需编码，Grafana支持Timestream作为数据源，拖拽配置折线图、仪表盘，还能添加告警规则，适合长期监控场景。

3. Plotly Dash构建交互式界面 #

如果需要复杂交互（如自定义时间范围、数据导出），Dash比PySimpleGUI更专业，基于React组件库，可构建企业级可视化界面。

内容的提问来源于stack exchange，提问作者Angad Bandal