Cloud Foundry上SAP IoT服务的sdk。包装所有设备管理API以及两个云网关。
iot-services-sdk的Python项目详细描述
目录
说明
此包将SAP IoT服务、Cloud Foundry版本设备管理API以及其他API和MQTT云网关API封装在易于使用的Python代码中。您可以找到关于api here和here的更多信息。
功能包括：
- 包装设备管理api的所有功能
- 更好的错误反馈
- 围绕两个云网关的包装器
- rest和mqtt这两个客户机都进行了扩展，以便它们直接处理pem文件中的加密密钥
- 更好的rest和mqtt错误处理和错误反馈
- mqtt客户机有一个on_错误和一个on_命令回调，它直接返回已解析的消息
- rest客户机分析批量和单次上载的错误，并抛出一个错误，并显示相应的消息
- 两个客户端都返回失败消息，以使用户能够轻松重试
要求
您将需要一个IoT服务实例Cloud Foundry来使用此sdk。您可以找到有关如何激活服务here的更多信息。
安装
该套餐将在pip:
pip install sap_iot_services_sdk
如果要手动下载，请确保将paho-mqtt和requests安装到python环境中。
如何获得支持
请使用GitHub Issues来提交错误。
许可证
版权所有（c）2019 SAP SE或SAP关联公司。保留所有权利。 除非LICENSE file中另有说明，否则此文件是根据SAP示例代码许可证授权的。
用法
有关详细信息，请参阅pdf文件中的api文档。您可以检查测试的示例代码。
下面是一个显示以下功能的小示例：
- 创建功能、传感器类型、传感器和设备。
- 下载证书并创建mqtt客户机。
- 发布有效和无效度量以及发布和接收命令。
- 删除创建的每个实体。
importosfromsap_iot_services_sdkimportCapabilityServicefromsap_iot_services_sdkimportDeviceServicefromsap_iot_services_sdkimportGatewayServicefromsap_iot_services_sdkimportSensorTypeServicefromsap_iot_services_sdkimportSensorServicefromsap_iot_services_sdkimportMQTTClient# Initialize servicesdevice_service=DeviceService(instance=myinstance.eu10.cp.iot.sap,user=myuser,password=mypassword)capability_service=CapabilityService(instance=myinstance.eu10.cp.iot.sap,user=myuser,password=mypassword)gateway_service=GatewayService(instance=myinstance.eu10.cp.iot.sap,user=myuser,password=mypassword)sensor_type_service=SensorTypeService(instance=myinstance.eu10.cp.iot.sap,user=myuser,password=mypassword)sensor_service=SensorService(instance=myinstance.eu10.cp.iot.sap,user=myuser,password=mypassword)# Turn requests debug functionality oncapability_service.debug_requests_on()device_service.debug_requests_on()gateway_service.debug_requests_on()sensor_type_service.debug_requests_on()sensor_service.debug_requests_on()# Get Gatewayfilters=["status eq 'online'"]gateways_response=gateway_service.get_gateways(filters=filters)gateways=gateways_response.get_result()mqtt_gateway=next(gatewayforgatewayingatewaysifgateway['protocolId']=='mqtt')# Create Measure Capabilityproperties=[{"name":"sdk_test_temp","dataType":"double","unitOfMeasure":"Celcius"}]create_response=capability_service.create_capability('sdk_test_measure_id','sdk_test_measure_cap',properties)capability_measure_id=create_response.get_result().get('id')capability_measure_alternate_id=create_response.get_result().get('alternateId')# Create Command Capabilityproperties=[{"name":"sdk_test_cmd","dataType":"string"}]create_response=capability_service.create_capability('sdk_test_cmd_id','sdk_test_cmd_cap',properties)capability_cmd_id=create_response.get_result().get('id')capability_cmd_alternate_id=create_response.get_result().get('alternateId')# Create Devicecreate_response=device_service.create_device(mqtt_gateway['id'],'sdk_device')device=create_response.get_result()device_id=device['id']device_alternate_id=device['alternateId']# Create Sensor Typecapabilities=[{'id':capability_measure_id,'type':'measure'},{'id':capability_cmd_id,'type':'command'}]create_response=sensor_type_service.create_sensor_type('123456','sdk_mqtt_test_type',capabilities)sensor_type_id=create_response.get_result().get('id')sensor_type_alternate_id=create_response.get_result().get('alternateId')# Create Sensorcreate_response=sensor_service.create_sensor(device_id,'sdk-test-mqtt-sensor-id','sdk-test-mqtt-sensor-name',sensor_type_id)sensor_id=create_response.get_result()['id']sensor_alternate_id=create_response.get_result()['alternateId']# Download Certificateres=device_service.get_device_pem(device_id).get_result()pem=res.get('pem')secret=res.get('secret')pem_filepath='cert.pem'# Write Certificate to PEM filepem_file=open(pem_filepath,'w')pem_file.write(pem)pem_file.close()# Get MQTT Client for Devicemqtt_client=device_service.get_mqtt_client(device_alternate_id,pem_filepath,secret)mqtt_client.connect()mqtt_client.loop_start()# Publish a measuremeasures=[{'sdk_test_temp':30}]message_info_measures=mqtt_client.publish(capability_measure_alternate_id,sensor_alternate_id,measures,device_alternate_id)# Read measuresget_measures_response=device_service.get_measures(device_id)measures=get_measures_response.get_result()# Publish invalid measure and receive errordefon_error(client,userdata,report):print('ERROR RECEIVED: ')print(report)mqtt_client.on_error=on_errormeasures=[{'i_am_invalid':30}]message_info_measures=mqtt_client.publish(capability_measure_alternate_id,sensor_alternate_id,measures,device_alternate_id)# Publish and receive commanddefon_command(client,userdata,command):print('COMMAND RECEIVED: ')print(command)mqtt_client.on_command=on_commandmqtt_client.subscribe(device_alternate_id)command={'sdk_test_cmd':'I am a SDK!'}send_command_res=device_service.send_command_to_device(device_id,capability_cmd_id,sensor_id,command)# Stop MQTT loopmqtt_client.loop_stop()# Delete Sensordelete_sensor_res=sensor_service.delete_sensor(sensor_id)# Delete Devicedelete_response=device_service.delete_device(device_id)# Delete SensorTypedelete_sensor_type_response=sensor_type_service.delete_sensor_type(sensor_type_id)# Delete Measure Capabilitydelete_response=capability_service.delete_capability(capability_measure_id)# Delete Command Capabilitydelete_response=capability_service.delete_capability(capability_cmd_id)# Delete Certfiletry:os.remove(pem_filepath)exceptOSError:pass
测试
要运行测试，必须在根目录中放置config.ini。它必须包含以下信息：
[IOTS]
INSTANCE=myinstance.eu10.cp.iot.sap
BASE_URI=/iot/core/api/v1
GATEWAY_URI=/iot/gateway
USER=myuser
PASSWORD=mypassword
然后，运行
nosetests
在根目录中。测试将在它们完成之后进行清理-之后无需检查您的系统。
生成文档
使用sphinx直接从docstring构建文档：
cd docs
sphinx-apidoc -f -o source/ ../sap_iot_services_sdk
make latexpdf
将sapiotservicessdk.pdf文件复制到根目录中。
欢迎加入QQ群-->： 979659372
