在物联网(IoT)应用中,设备数据的采集与监控至关重要。本文将详细介绍如何使用Python从阿里云物联网平台获取STM32设备的温度数据。我们将从已有的Java代码出发,逐步将其转换为Python,并处理在过程中遇到的问题,最终实现一个稳定且高效的Python脚本。
前言
在物联网应用中,设备数据的实时监控是不可或缺的一部分。阿里云物联网平台提供了丰富的API接口,方便开发者获取和管理设备数据。本文将通过将Java程序转写为Python,实现从阿里云物联网平台获取STM32设备的温度数据,并确保代码的稳定性和可维护性。
环境准备
在开始编码之前,需要确保以下环境和依赖已准备好:
-
Python 3.x:确保已安装Python 3.x版本。可以通过以下命令检查版本:
python --version
-
安装阿里云Python SDK:
使用
pip
安装阿里云的Python SDK:pip install aliyun-python-sdk-core
-
获取阿里云认证信息:
您需要拥有阿里云的
Access Key ID
和Access Key Secret
,用于认证API请求。 -
设备信息:
ProductKey
DeviceName
IotInstanceId
请确保这些信息已正确配置,并且设备已成功连接到阿里云物联网平台。
原始Java代码解析
本文最初的Java程序用于从阿里云物联网平台获取设备属性状态。主要步骤包括:
- 初始化阿里云客户端:使用
DefaultAcsClient
,通过accessKeyId
和secret
进行认证。 - 发送设备属性查询请求:使用
QueryDevicePropertyStatusRequest
构造请求,并发送给阿里云。 - 处理响应:解析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():
# 初始化AcsClient
client = 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 main
get_device_property(client)
File "c:\aliyuniot\main.py", line 52, in get_device_property
first_property = properties[0]
IndexError: list index out of range
这个错误表明在尝试访问properties[0]
时,列表为空,导致IndexError
。分析可能的原因:
- JSON键名大小写不匹配:确保JSON响应中的键名与代码中一致。
- API响应中确实没有数据:可能是请求参数配置错误,或设备未上报数据。
- 认证或请求参数错误:验证
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():
# 初始化AcsClient
client = 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 = None
for prop in property_status_info:
if prop.get('Name') == "STM32温度":
stm32_temp_value = prop.get('Value')
break
if 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():
# 初始化AcsClient
client = AcsClient(ACCESS_KEY_ID, SECRET, REGION_ID)
# 无限循环,每秒查询一次
while True:
get_stm32_temperature(client)
time.sleep(1) # 暂停1秒
if __name__ == "__main__":
main()
进一步优化
为了提高代码的健壮性和可维护性,我们进行了以下优化:
- 使用
logging
模块替代print
:方便管理日志级别和格式。 - 环境变量管理敏感信息:避免将
Access Key ID
和Secret
硬编码在脚本中。 - 优雅地终止脚本:捕获键盘中断信号,使脚本可以通过
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 = None
for prop in property_status_info:
if prop.get('Name') == "STM32温度":
stm32_temp_value = prop.get('Value')
break
if 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
# 初始化AcsClient
client = 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 = None
for prop in property_status_info:
if prop.get('Name') == "STM32温度":
stm32_temp_value = prop.get('Value')
break
if 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
# 初始化AcsClient
client = 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()
运行与测试
在运行脚本之前,请确保完成以下步骤:
-
设置环境变量:
为了增强安全性,不将
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
-
-
安装必要的Python库:
确保已安装
aliyun-python-sdk-core
:pip install aliyun-python-sdk-core
-
运行脚本:
通过命令行运行Python脚本:
python main.py
-
检查输出:
如果设备正确上报数据,您将看到类似以下的输出:
2023-10-01 12:00:00,000 - INFO - STM32温度:36
若发生错误或未找到相关属性,您将看到相应的警告或错误信息。
总结与注意事项
本文详细介绍了如何使用Python从阿里云物联网平台获取STM32设备的温度数据,包括从Java代码的转换、错误处理、日志记录、安全管理等多个方面。以下是一些关键的注意事项:
-
安全性:
- 保护认证信息:避免将
Access Key ID
和Secret
硬编码在代码中,建议使用环境变量或安全的配置管理工具。 - 权限管理:确保阿里云账号具备必要的权限,避免过度授权。
- 保护认证信息:避免将
-
错误处理:
- 捕获异常:通过
try-except
块捕获并处理可能的异常,如请求失败、JSON解析错误等。 - 日志记录:使用
logging
模块记录信息、警告和错误,方便后续调试和维护。
- 捕获异常:通过
-
代码优化:
- 模块化:将功能模块化,便于维护和扩展。
- 配置管理:将配置信息集中管理,便于修改和统一管理。
-
性能与稳定性:
- 合理的查询间隔:当前设置为每秒查询一次,根据实际需求调整查询频率,避免过度请求导致API限制或费用增加。
- 优雅终止:通过捕获
KeyboardInterrupt
信号,实现脚本的优雅终止,确保资源的正确释放。
-
调试与测试:
- 打印完整响应:在调试阶段,打印完整的API响应,了解数据结构,便于提取所需信息。
- 测试设备连接:确保设备已正确连接至阿里云物联网平台,并能正常上报数据。
通过本文的指导,您可以轻松地使用Python从阿里云物联网平台获取STM32设备的温度数据,并在此基础上扩展更多功能,如数据存储、实时监控、报警系统等。如有任何疑问或需要进一步的帮助,欢迎在评论区留言讨论!
免责声明:本文中的Access Key ID
和Secret
已经被替换为占位符YOUR_ACCESS_KEY_ID
和YOUR_ACCESS_KEY_SECRET
。请勿在实际代码中使用本文示例中的敏感信息。务必确保您的认证信息安全,避免泄露。