使用Python从阿里云物联网平台获取STM32温度数据

在物联网（IoT）应用中，设备数据的采集与监控至关重要。本文将详细介绍如何使用Python从阿里云物联网平台获取STM32设备的温度数据。我们将从已有的Java代码出发，逐步将其转换为Python，并处理在过程中遇到的问题，最终实现一个稳定且高效的Python脚本。

前言

在物联网应用中，设备数据的实时监控是不可或缺的一部分。阿里云物联网平台提供了丰富的API接口，方便开发者获取和管理设备数据。本文将通过将Java程序转写为Python，实现从阿里云物联网平台获取STM32设备的温度数据，并确保代码的稳定性和可维护性。

环境准备

在开始编码之前，需要确保以下环境和依赖已准备好：

1. Python 3.x：确保已安装Python 3.x版本。可以通过以下命令检查版本：

   python --version
   

2. 安装阿里云Python SDK：

   使用pip安装阿里云的Python SDK：

   pip install aliyun-python-sdk-core
   

3. 获取阿里云认证信息：

   您需要拥有阿里云的Access Key ID和Access Key Secret，用于认证API请求。

4. 设备信息：

   • ProductKey
   • DeviceName
   • IotInstanceId

   请确保这些信息已正确配置，并且设备已成功连接到阿里云物联网平台。

原始Java代码解析

本文最初的Java程序用于从阿里云物联网平台获取设备属性状态。主要步骤包括：

1. 初始化阿里云客户端：使用DefaultAcsClient，通过accessKeyId和secret进行认证。
2. 发送设备属性查询请求：使用QueryDevicePropertyStatusRequest构造请求，并发送给阿里云。
3. 处理响应：解析JSON响应，提取传感器数据并打印。

以下是关键的Java代码片段：

// 初始化客户端
IClientProfile profile = DefaultProfile.getProfile(regionId, accessKeyId, secret);
client = new DefaultAcsClient(profile);// 构造查询设备属性请求
QueryDevicePropertyStatusRequest request = new QueryDevicePropertyStatusRequest();
request.setRegionId(regionId);
request.setIotInstanceId(iotid);
request.setDeviceName(deviceName);
request.setProductKey(deviceProductkey);// 发送请求并获取响应
QueryDevicePropertyStatusResponse response = client.getAcsResponse(request);// 解析JSON响应并提取数据
String string = JSON.toJSONString(response);
JsonObject jsonObject = (JsonObject) parser.parse(string);
JsonObject data = (JsonObject) jsonObject.get("data");
JsonArray list = (JsonArray) data.get("list");
System.out.println("传感器数据：" + list.get(0).getAsJsonObject().get("value"));

Python代码转换与优化

将上述Java代码转换为Python之后，初步实现了设备数据的获取。然而，在实际运行中遇到了IndexError，这是因为响应中的数据列表为空。本文将逐步优化代码，确保其稳定性和准确性。

初始Python代码

初步的Python代码如下：

import time
import json
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkcore.acs_exception.exceptions import ClientException# ====================== 配置信息 ======================
ACCESS_KEY_ID = "YOUR_ACCESS_KEY_ID"  # 请替换为您的Access Key ID
SECRET = "YOUR_ACCESS_KEY_SECRET"     # 请替换为您的Access Key Secret
REGION_ID = "cn-shanghai"DEVICE_PRODUCTKEY = "hfuyHEEc31U"
DEVICE_NAME = "STM32_tem_hum_1215"IOT_INSTANCE_ID = "iot-06z00f0h5dpyp1f"
# =======================================================def get_device_property(client):"""查询设备属性状态并打印第一个传感器的数据值"""# 创建请求对象request = CommonRequest()request.set_accept_format('json')request.set_domain('iot.cn-shanghai.aliyuncs.com')  # 根据您的地域可能需要调整request.set_method('POST')request.set_version('2018-01-20')request.set_action_name('QueryDevicePropertyStatus')# 设置请求参数request.add_query_param('IotInstanceId', IOT_INSTANCE_ID)request.add_query_param('DeviceName', DEVICE_NAME)request.add_query_param('ProductKey', DEVICE_PRODUCTKEY)try:# 发送请求并获取响应response = client.do_action_with_exception(request)# 响应内容是字节，需要解码成字符串response_str = response.decode('utf-8')# 解析JSON响应response_json = json.loads(response_str)# 打印整个响应（用于调试，实际使用时可以移除）print("完整响应:", json.dumps(response_json, indent=4, ensure_ascii=False))# 提取传感器数据data = response_json.get('data', {})properties = data.get('list', [])if properties:# 假设您要获取列表中的第一个属性值first_property = properties[0]value = first_property.get('Value')print(f"传感器数据：{value}")else:print("未获取到传感器数据。")print(f"Data 内容: {data}")except ClientException as e:print(f"请求失败：{e}")def main():# 初始化AcsClientclient = AcsClient(ACCESS_KEY_ID, SECRET, REGION_ID)# 无限循环，每秒查询一次while True:get_device_property(client)time.sleep(1)  # 暂停1秒if __name__ == "__main__":main()

处理空列表错误

运行上述代码时，可能会遇到以下错误：

Traceback (most recent call last):File "c:\aliyuniot\main.py", line 71, in <module>main()File "c:\aliyuniot\main.py", line 67, in mainget_device_property(client)File "c:\aliyuniot\main.py", line 52, in get_device_propertyfirst_property = properties[0]
IndexError: list index out of range

这个错误表明在尝试访问properties[0]时，列表为空，导致IndexError。分析可能的原因：

1. JSON键名大小写不匹配：确保JSON响应中的键名与代码中一致。
2. API响应中确实没有数据：可能是请求参数配置错误，或设备未上报数据。
3. 认证或请求参数错误：验证AccessKeyId、Secret、IotInstanceId、DeviceName、ProductKey是否正确。

为解决上述问题，我们对代码进行了如下修改：

• 将JSON键名改为小写，确保与响应一致。
• 添加调试信息，打印完整的API响应。
• 添加对空列表的检查。

修改后的代码如下：

import time
import json
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkcore.acs_exception.exceptions import ClientException# ====================== 配置信息 ======================
ACCESS_KEY_ID = "YOUR_ACCESS_KEY_ID"  # 请替换为您的Access Key ID
SECRET = "YOUR_ACCESS_KEY_SECRET"     # 请替换为您的Access Key Secret
REGION_ID = "cn-shanghai"DEVICE_PRODUCTKEY = "hfuyHEEc31U"
DEVICE_NAME = "STM32_tem_hum_1215"IOT_INSTANCE_ID = "iot-06z00f0h5dpyp1f"
# =======================================================def get_device_property(client):"""查询设备属性状态并打印第一个传感器的数据值"""# 创建请求对象request = CommonRequest()request.set_accept_format('json')request.set_domain('iot.cn-shanghai.aliyuncs.com')  # 根据您的地域可能需要调整request.set_method('POST')request.set_version('2018-01-20')request.set_action_name('QueryDevicePropertyStatus')# 设置请求参数request.add_query_param('IotInstanceId', IOT_INSTANCE_ID)request.add_query_param('DeviceName', DEVICE_NAME)request.add_query_param('ProductKey', DEVICE_PRODUCTKEY)try:# 发送请求并获取响应response = client.do_action_with_exception(request)# 响应内容是字节，需要解码成字符串response_str = response.decode('utf-8')# 解析JSON响应response_json = json.loads(response_str)# 打印整个响应（用于调试，实际使用时可以移除）print("完整响应:", json.dumps(response_json, indent=4, ensure_ascii=False))# 提取传感器数据data = response_json.get('data', {})properties = data.get('list', [])if properties:# 假设您要获取列表中的第一个属性值first_property = properties[0]value = first_property.get('Value')print(f"传感器数据：{value}")else:print("未获取到传感器数据。")print(f"Data 内容: {data}")except ClientException as e:print(f"请求失败：{e}")except json.JSONDecodeError as e:print(f"JSON 解析失败：{e}")except Exception as e:print(f"发生未预期的错误：{e}")def main():# 初始化AcsClientclient = AcsClient(ACCESS_KEY_ID, SECRET, REGION_ID)# 无限循环，每秒查询一次while True:get_device_property(client)time.sleep(1)  # 暂停1秒if __name__ == "__main__":main()

提取STM32温度数据

根据用户提供的完整JSON响应结构，PropertyStatusInfo是一个包含多个属性的列表。我们需要遍历该列表，找到Name为“STM32温度”的条目，并提取其Value。

修改后的代码如下：

import time
import json
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkcore.acs_exception.exceptions import ClientException# ====================== 配置信息 ======================
ACCESS_KEY_ID = "YOUR_ACCESS_KEY_ID"  # 请替换为您的Access Key ID
SECRET = "YOUR_ACCESS_KEY_SECRET"     # 请替换为您的Access Key Secret
REGION_ID = "cn-shanghai"DEVICE_PRODUCTKEY = "hfuyHEEc31U"
DEVICE_NAME = "STM32_tem_hum_1215"IOT_INSTANCE_ID = "iot-06z00f0h5dpyp1f"
# =======================================================def get_stm32_temperature(client):"""查询设备属性状态并打印STM32温度的值"""# 创建请求对象request = CommonRequest()request.set_accept_format('json')request.set_domain('iot.cn-shanghai.aliyuncs.com')  # 根据您的地域可能需要调整request.set_method('POST')request.set_version('2018-01-20')request.set_action_name('QueryDevicePropertyStatus')# 设置请求参数request.add_query_param('IotInstanceId', IOT_INSTANCE_ID)request.add_query_param('DeviceName', DEVICE_NAME)request.add_query_param('ProductKey', DEVICE_PRODUCTKEY)try:# 发送请求并获取响应response = client.do_action_with_exception(request)# 响应内容是字节，需要解码成字符串response_str = response.decode('utf-8')# 解析JSON响应response_json = json.loads(response_str)# 打印整个响应（用于调试，实际使用时可以移除）print("完整响应:", json.dumps(response_json, indent=4, ensure_ascii=False))# 提取传感器数据data = response_json.get('Data', {})list_data = data.get('List', {})property_status_info = list_data.get('PropertyStatusInfo', [])if not property_status_info:print("未获取到 PropertyStatusInfo 数据。")print(f"Data 内容: {data}")return# 寻找名称为 "STM32温度" 的属性stm32_temp_value = Nonefor prop in property_status_info:if prop.get('Name') == "STM32温度":stm32_temp_value = prop.get('Value')breakif stm32_temp_value is not None:print(f"STM32温度：{stm32_temp_value}")else:print("未找到名称为 'STM32温度' 的属性。")except ClientException as e:print(f"请求失败：{e}")except json.JSONDecodeError as e:print(f"JSON 解析失败：{e}")except Exception as e:print(f"发生未预期的错误：{e}")def main():# 初始化AcsClientclient = AcsClient(ACCESS_KEY_ID, SECRET, REGION_ID)# 无限循环，每秒查询一次while True:get_stm32_temperature(client)time.sleep(1)  # 暂停1秒if __name__ == "__main__":main()

进一步优化

为了提高代码的健壮性和可维护性，我们进行了以下优化：

1. 使用logging模块替代print：方便管理日志级别和格式。
2. 环境变量管理敏感信息：避免将Access Key ID和Secret硬编码在脚本中。
3. 优雅地终止脚本：捕获键盘中断信号，使脚本可以通过Ctrl+C优雅地退出。

优化后的代码如下：

import time
import json
import os
import logging
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkcore.acs_exception.exceptions import ClientException# ====================== 配置信息 ======================
ACCESS_KEY_ID = os.getenv('ALIYUN_ACCESS_KEY_ID')  # 从环境变量获取
SECRET = os.getenv('ALIYUN_SECRET')               # 从环境变量获取
REGION_ID = "cn-shanghai"DEVICE_PRODUCTKEY = "hfuyHEEc31U"
DEVICE_NAME = "STM32_tem_hum_1215"IOT_INSTANCE_ID = "iot-06z00f0h5dpyp1f"
# =======================================================# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')def get_stm32_temperature(client):"""查询设备属性状态并打印STM32温度的值"""# 创建请求对象request = CommonRequest()request.set_accept_format('json')request.set_domain('iot.cn-shanghai.aliyuncs.com')  # 根据您的地域可能需要调整request.set_method('POST')request.set_version('2018-01-20')request.set_action_name('QueryDevicePropertyStatus')# 设置请求参数request.add_query_param('IotInstanceId', IOT_INSTANCE_ID)request.add_query_param('DeviceName', DEVICE_NAME)request.add_query_param('ProductKey', DEVICE_PRODUCTKEY)try:# 发送请求并获取响应response = client.do_action_with_exception(request)# 响应内容是字节，需要解码成字符串response_str = response.decode('utf-8')# 解析JSON响应response_json = json.loads(response_str)# 打印整个响应（用于调试，实际使用时可以移除）logging.debug("完整响应: %s", json.dumps(response_json, indent=4, ensure_ascii=False))# 提取传感器数据data = response_json.get('Data', {})list_data = data.get('List', {})property_status_info = list_data.get('PropertyStatusInfo', [])if not property_status_info:logging.warning("未获取到 PropertyStatusInfo 数据。")logging.debug("Data 内容: %s", data)return# 寻找名称为 "STM32温度" 的属性stm32_temp_value = Nonefor prop in property_status_info:if prop.get('Name') == "STM32温度":stm32_temp_value = prop.get('Value')breakif stm32_temp_value is not None:logging.info(f"STM32温度：{stm32_temp_value}")else:logging.warning("未找到名称为 'STM32温度' 的属性。")except ClientException as e:logging.error(f"请求失败：{e}")except json.JSONDecodeError as e:logging.error(f"JSON 解析失败：{e}")except Exception as e:logging.error(f"发生未预期的错误：{e}")def main():# 检查环境变量是否设置if not ACCESS_KEY_ID or not SECRET:logging.error("未设置环境变量 ALIYUN_ACCESS_KEY_ID 和/或 ALIYUN_SECRET。请设置后重试。")return# 初始化AcsClientclient = AcsClient(ACCESS_KEY_ID, SECRET, REGION_ID)try:# 无限循环，每秒查询一次while True:get_stm32_temperature(client)time.sleep(1)  # 暂停1秒except KeyboardInterrupt:logging.info("脚本已手动终止。")if __name__ == "__main__":main()

完整优化后的Python代码

结合上述所有优化，以下是最终的Python脚本，用于从阿里云物联网平台获取STM32温度数据。

import time
import json
import os
import logging
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from aliyunsdkcore.acs_exception.exceptions import ClientException# ====================== 配置信息 ======================
ACCESS_KEY_ID = os.getenv('ALIYUN_ACCESS_KEY_ID')  # 从环境变量获取
SECRET = os.getenv('ALIYUN_SECRET')               # 从环境变量获取
REGION_ID = "cn-shanghai"DEVICE_PRODUCTKEY = "hfuyHEEc31U"
DEVICE_NAME = "STM32_tem_hum_1215"IOT_INSTANCE_ID = "iot-06z00f0h5dpyp1f"
# =======================================================# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')def get_stm32_temperature(client):"""查询设备属性状态并打印STM32温度的值"""# 创建请求对象request = CommonRequest()request.set_accept_format('json')request.set_domain('iot.cn-shanghai.aliyuncs.com')  # 根据您的地域可能需要调整request.set_method('POST')request.set_version('2018-01-20')request.set_action_name('QueryDevicePropertyStatus')# 设置请求参数request.add_query_param('IotInstanceId', IOT_INSTANCE_ID)request.add_query_param('DeviceName', DEVICE_NAME)request.add_query_param('ProductKey', DEVICE_PRODUCTKEY)try:# 发送请求并获取响应response = client.do_action_with_exception(request)# 响应内容是字节，需要解码成字符串response_str = response.decode('utf-8')# 解析JSON响应response_json = json.loads(response_str)# 打印整个响应（用于调试，实际使用时可以移除）logging.debug("完整响应: %s", json.dumps(response_json, indent=4, ensure_ascii=False))# 提取传感器数据data = response_json.get('Data', {})list_data = data.get('List', {})property_status_info = list_data.get('PropertyStatusInfo', [])if not property_status_info:logging.warning("未获取到 PropertyStatusInfo 数据。")logging.debug("Data 内容: %s", data)return# 寻找名称为 "STM32温度" 的属性stm32_temp_value = Nonefor prop in property_status_info:if prop.get('Name') == "STM32温度":stm32_temp_value = prop.get('Value')breakif stm32_temp_value is not None:logging.info(f"STM32温度：{stm32_temp_value}")else:logging.warning("未找到名称为 'STM32温度' 的属性。")except ClientException as e:logging.error(f"请求失败：{e}")except json.JSONDecodeError as e:logging.error(f"JSON 解析失败：{e}")except Exception as e:logging.error(f"发生未预期的错误：{e}")def main():# 检查环境变量是否设置if not ACCESS_KEY_ID or not SECRET:logging.error("未设置环境变量 ALIYUN_ACCESS_KEY_ID 和/或 ALIYUN_SECRET。请设置后重试。")return# 初始化AcsClientclient = AcsClient(ACCESS_KEY_ID, SECRET, REGION_ID)try:# 无限循环，每秒查询一次while True:get_stm32_temperature(client)time.sleep(1)  # 暂停1秒except KeyboardInterrupt:logging.info("脚本已手动终止。")if __name__ == "__main__":main()

运行与测试

在运行脚本之前，请确保完成以下步骤：

1. 设置环境变量：

   为了增强安全性，不将Access Key ID和Secret硬编码在脚本中。请通过环境变量设置这些信息。

   • 在Linux或macOS上：

     export ALIYUN_ACCESS_KEY_ID="您的Access Key ID"
     export ALIYUN_SECRET="您的Access Key Secret"
     

   • 在Windows上：

     set ALIYUN_ACCESS_KEY_ID=您的Access Key ID
     set ALIYUN_SECRET=您的Access Key Secret
     

2. 安装必要的Python库：

   确保已安装aliyun-python-sdk-core：

   pip install aliyun-python-sdk-core
   

3. 运行脚本：

   通过命令行运行Python脚本：

   python main.py
   

4. 检查输出：

   如果设备正确上报数据，您将看到类似以下的输出：

   2023-10-01 12:00:00,000 - INFO - STM32温度：36
   

   若发生错误或未找到相关属性，您将看到相应的警告或错误信息。

总结与注意事项

本文详细介绍了如何使用Python从阿里云物联网平台获取STM32设备的温度数据，包括从Java代码的转换、错误处理、日志记录、安全管理等多个方面。以下是一些关键的注意事项：

1. 安全性：

   • 保护认证信息：避免将Access Key ID和Secret硬编码在代码中，建议使用环境变量或安全的配置管理工具。
   • 权限管理：确保阿里云账号具备必要的权限，避免过度授权。

2. 错误处理：

   • 捕获异常：通过try-except块捕获并处理可能的异常，如请求失败、JSON解析错误等。
   • 日志记录：使用logging模块记录信息、警告和错误，方便后续调试和维护。

3. 代码优化：

   • 模块化：将功能模块化，便于维护和扩展。
   • 配置管理：将配置信息集中管理，便于修改和统一管理。

4. 性能与稳定性：

   • 合理的查询间隔：当前设置为每秒查询一次，根据实际需求调整查询频率，避免过度请求导致API限制或费用增加。
   • 优雅终止：通过捕获KeyboardInterrupt信号，实现脚本的优雅终止，确保资源的正确释放。

5. 调试与测试：

   • 打印完整响应：在调试阶段，打印完整的API响应，了解数据结构，便于提取所需信息。
   • 测试设备连接：确保设备已正确连接至阿里云物联网平台，并能正常上报数据。

通过本文的指导，您可以轻松地使用Python从阿里云物联网平台获取STM32设备的温度数据，并在此基础上扩展更多功能，如数据存储、实时监控、报警系统等。如有任何疑问或需要进一步的帮助，欢迎在评论区留言讨论！

---

免责声明：本文中的Access Key ID和Secret已经被替换为占位符YOUR_ACCESS_KEY_ID和YOUR_ACCESS_KEY_SECRET。请勿在实际代码中使用本文示例中的敏感信息。务必确保您的认证信息安全，避免泄露。