Python Edge SDK Manual

October 21, 2024

Environment

Runtime
- Python: 3.6
- Pip: 1.8
Python Package:
- EdgeSync360-EdgeHub-Edge-Python-SDK

Installation

Via the command line

pip install EdgeSync360-EdgeHub-Edge-Python-SDK

Upgrade the package version

pip install EdgeSync360-EdgeHub-Edge-Python-SDK --upgrade

Check the installed version

pip show EdgeSync360-EdgeHub-Edge-Python-SDK

Name: EdgeSync360_EdgeHub_Edge_Python_SDK
Version: 1.0.0
Summary: EdgeSync360_EdgeHub_Edge_Python_SDK package allows developers to write Python applications which access the EdgeSync360/EdgeHub Platform via MQTT protocol.
Home-page: https://github.com/EdgeHub-Repo/dc-edge-sdk-python-sample
Author: Paul Tseng
Author-email: edgehub.dev@gmail.com
License:
Location: /opt/homebrew/lib/python3.12/site-packages
Requires: paho-mqtt
Required-by:

FAQ

The system displays the following SSL certificate error messages:

```json
[SSL: CERTIFICATE_VERIFY_FAILED] certificate verification failed (_ssl.c:720

Windows
- Install Win 32/Win 64 OpenSSL.
Mac OS
- You have two options:

Run an install command shipped with Python 3.6.

cd /Applications/Python\ 3.6/
./Install\ Certificates.command

Install the certification package with

pip install certifi

EdgeAgent

Constructor (EdgeAgentOptions options)

New EdgeAgent object

options = EdgeAgentOptions(
  reconnectInterval = 1,                                # MQTT reconnect interval in seconds.
  nodeId = '5095cf13-f005-4c81-b6c9-68cf038e2b87',      # Get from portal
  deviceId = 'device1',                                 # If the type is Device, deviceId must be input.
  type = constant.EdgeType['Gateway'],                  # Configure the edge as a Gateway or Device. The default setting is Gateway.
  heartbeat = 60,                                       # The default is 60 seconds.
  dataRecover = True,                                   # Whether to recover data when disconnected
  connectType = constant.ConnectType['DCCS'],           # Connection type (DCCS, MQTT). The default setting is DCCS.
  MQTT = MQTTOptions(                                   # If the connectType is MQTT, the following options must be input:
    hostName = '127.0.0.1',
    port = 1883,
    userName = 'admin',
    password = 'admin',
    protocolType = constant.Protocol['TCP']             # MQTT protocol (TCP, WebSocket). The default is TCP.
  ),
  DCCS = DCCSOptions(
    APIURL = '',         # DCCS API URL
    credentialKey = ''  # Credential key
  ),
  AzureIotHub = AzureIotHub=AzureIotHubOptions(
      connectionString='',
  ),
)

edgeAgent = EdgeAgent(options = options);

Azure IoT Hub

Due to the connection limit of the Azure IoT Hub SDK, the C# SDK can only connect to the Azure IoT Hub using a connection string. Therefore, you can only connect to the Edge Hub using a EdgeLink device.

Event

EdgeAgent has three subscription events.

Connected: When EdgeAgent is connected to the IoTHub.
Disconnected: When EdgeAgent is disconnected from the IoTHub.
MessageReceived: When EdgeAgent receives an MQTT message from the cloud. The message types are as follows:
- WriteValue: Change the tag value from the cloud.
- WriteConfig: Change the config from the cloud.
- TimeSync: Returns the current time from the cloud.
- ConfigAck: Response to config uploads from the edge to the cloud.

edgeAgent.on_connected = edgeAgent_on_connected
edgeAgent.on_disconnected = edgeAgent_on_disconnected
edgeAgent.on_message = edgeAgent_on_message

def edgeAgent_on_connected(agent, isConnected):
  print('Connection success')

def edgeAgent_on_disconnected(agent, isDisconnected):
  print('Disconnected')

def edgeAgent_on_message(agent, messageReceivedEventArgs):
  # messageReceivedEventArgs format: Model.Event.MessageReceivedEventArgs
  type = messageReceivedEventArgs.type
  message = messageReceivedEventArgs.message
  if type == constant.MessageType['WriteValue']:
    # message format: Model.Edge.WriteValueCommand
    for device in message.deviceList:
      print('deviceId: &#123;0&#125;'.format(device.Id))
      for tag in device.tagList:
        print('tagName: &#123;0&#125;, Value: &#123;1&#125;'.format(tag.name, str(tag.value)))
  elif type == constant.MessageType['WriteConfig']:
    print('WriteConfig')
  elif type == constant.MessageType['TimeSync']:
    # message format: Model.Edge.TimeSyncCommand
    print(str(message.UTCTime))
  elif type == constant.MessageType['ConfigAck']:
    # message format: Model.Edge.ConfigAck
    print('Upload Config Result: &#123;0&#125;'&#125;.format(str(message.result)))

Connect ()

Connect to the IoTHub. When connected successfully, a connected event will be triggered.

edgeAgent.connect();

Disconnect ()

Disconnect from the IoTHub. When disconnected successfully, a disconnected event will be triggered.

edgeAgent.disconnect();

UploadConfig (ActionType action, EdgeConfig edgeConfig)

Upload Node/Device/Tag Config with Action Type (Create/Update/Delete).

config = EdgeConfig()
# set node config
# set device config
# set tag config

result = edgeAgent.uploadConfig(constant.ActionType['Create'], edgeConfig = config)

Node config setting

nodeConfig = NodeConfig(nodeType = constant.EdgeType['Gateway'])
config.node = nodeConfig

Device config setting

deviceConfig = DeviceConfig(
  id = 'Device1',
  name = 'Device 1',
  deviceType = 'Device Type',
  description = 'Description'
)
config.node.deviceList.append(deviceConfig)

Analog tag config setting

analogTag = AnalogTagConfig(
  name = 'ATag',
  description = 'AnalogTag',
  readOnly = True,
  arraySize = 0,
  spanHigh = 10,
  spanLow = 0,
  engineerUnit = 'cm',
  integerDisplayFormat = 2,
  fractionDisplayFormat = 4
)
config.node.deviceList[0].analogTagList.append(analogTag)

Discrete Tag config setting

discreteTag = DiscreteTagConfig(
  name = 'DTag',
  description = 'DiscreteTag',
  readOnly = False,
  arraySize = 0,
  state0 = '1',
  state1 = '0',
  state2 = None,
  state3 = None,
  state4 = None,
  state5 = None,
  state6 = None,
  state7 = None
)
config.node.deviceList[0].discreteTagList.append(discreteTag)

Text tag config setting

textTag = TextTagConfig(
  name = 'TTag',
  description = 'TextTag',
  readOnly = True,
  arraySize = 0
)
config.node.deviceList[0].textTagList.append(textTag)

Block Config

Create block config:

blockConfig = BlockConfig(
  blockType = 'Pump',
  analogTagList = [
    AnalogTagConfig(
      name = 'ATag',
      description = 'AnalogTag',
      readOnly = True,
      arraySize = 0,
      spanHigh = 10,
      spanLow = 0,
      engineerUnit = 'cm',
      integerDisplayFormat = 2,
      fractionDisplayFormat = 4
    )
  ],
  discreteTagList = [
    DiscreteTagConfig(
      name = 'DTag',
      description = 'DiscreteTag',
      readOnly = False,
      arraySize = 0,
      state0 = '1',
      state1 = '0',
      state2 = None,
      state3 = None,
      state4 = None,
      state5 = None,
      state6 = None,
      state7 = None
    )
  ],
  textTagList = [
    TextTagConfig(
      name = 'TTag',
      description = 'TextTag',
      readOnly = True,
      arraySize = 0
    )
  ]

)

This block config contain 3 tag, and the block type is `Pump`, using this block config create two block:

```python
deviceConfig.addBlock('Pump01',blockConfig)
deviceConfig.addBlock('Pump02',blockConfig)

This will add 6 tag to your device, the first 3 tag will have a prefix of Pump01 and the last 3 tag will have a prefix of Pump02, for example:

SendData (EdgeData data)

Send the tag values to the cloud.

edgeData = EdgeData()
for i in range(1, 3):
  for j in range(1, 5):
    deviceId = 'Device' + str(i)
    tagName = 'ATag' + str(j)
    value = random.uniform(0, 100)
    tag = EdgeTag(deviceId, tagName, value)
    edgeData.tagList.append(tag)
  for j in range(1, 5):
    deviceId = 'Device' + str(i)
    tagName = 'DTag' + str(j)
    value = random.randint(0, 99) % 2
    tag = EdgeTag(deviceId, tagName, value)
    edgeData.tagList.append(tag)
  for j in range(1, 5):
    deviceId = 'Device' + str(i)
    tagName = 'TTag' + str(j)
    value = 'TEST ' + j
    tag = EdgeTag(deviceId, tagName, value)
    edgeData.tagList.append(tag)

edgeData.timestamp = datetime.datetime.now()
#edgeData.timestamp = datetime.datetime(2020,8,24,6,10,8)  	# You can specify the time(local time) of data

result = edgeAgent.sendData(data = edgeData)

An array tag value must use Dictionary, T is defined according to the tag type. (Analog: double, Discrete: int, Text: string).

# analog array tag
dicVal = {
  &#39;0&#39;: 0.5,
  &#39;1&#39;: 1.42
};
tag = EdgeTag(deviceId = 'Device1', tagName = 'ATag', value = dicVal)

# discrete array tag
dicVal = {
  &#39;0&#39;: 0,
  &#39;1&#39;: 1
};
tag = EdgeTag(deviceId = 'Device1', tagName = 'DTag', value = dicVal)

# text array tag
dicVal = {
  &#39;0&#39;: &#39;zero&#39;,
  &#39;1&#39;: &#39;one&#39;
};
tag = EdgeTag(deviceId = 'Device1', tagName = 'TTag', value = dicVal)

Send the tag values in batch to the cloud.

array = []
for n in range(1, 10):
  edgeData = EdgeData()
  for i in range(1, 3):
    for j in range(1, 5):
      deviceId = 'Device' + str(i)
      tagName = 'ATag' + str(j)
      value = random.uniform(0, 100)
      tag = EdgeTag(deviceId, tagName, value)
      edgeData.tagList.append(tag)
    for j in range(1, 5):
      deviceId = 'Device' + str(i)
      tagName = 'DTag' + str(j)
      value = random.randint(0, 99) % 2
      tag = EdgeTag(deviceId, tagName, value)
      edgeData.tagList.append(tag)
    for j in range(1, 5):
      deviceId = 'Device' + str(i)
      tagName = 'TTag' + str(j)
      value = 'TEST ' + j
      tag = EdgeTag(deviceId, tagName, value)
      edgeData.tagList.append(tag)
  edgeData.timestamp = datetime.datetime.now()
  array.append(edgeData)

result = edgeAgent.sendData(array)

SendDeviceStatus (EdgeDeviceStatus deviceStatus)

Send the device status to the cloud when the status is changed.

deviceStatus = EdgeDeviceStatus()
for i in range(1, 3):
  id = 'Device' + str(i)
  status = EdgeStatus(id = id, status = constant.Status['Online'])

  deviceStatus.deviceList.append(status)

result = edgeAgent.sendDeviceStatus(deviceStatus)

Property

Property Name	Data Type	Description
isConnected	Boolean	Connection Status (read only)

Sample Code

GitHub
- dc-edge-sdk-python-sample

Release Note

[1.0.2] - 2024-10-21
- Add
  - Support Azure IoT Hub device connection which is created from EdgeHub
[1.0.1] - 2024-09-09
- Add
  - Support BlockType that is defined in WISE-PaaS MQTT v1.0.16
[1.0.0] - 2024-08-08
- Add
  - EdgeSync360 EdgeHub EdgeSDK Project initialized, derived from WISE-PaaS DataHub EdgeSDK

Release Note (Archived: WISE-PaaS DataHub EdgeSDK)

[1.1.7] - 2021-09-29
- Added
  - Support send data in batch for high frequency data
[1.1.6] - 2021-09-09
- Changed
  - Remove the mechanism for splitting huge data packages into many small packages
[1.1.5] - 2020-11-26
- Added
  - Add logger for debugging
- Fixed
  - Fix the problem that error log will appear when user doesn’t implement on_message event
- Removed
  - TimeSync and WriteConfig of message received event
[1.1.4] - 2020-09-15
- Fixed
  - The array size of TagConfig set invalid issue
[1.1.3] - 2020-08-30
- Fixed
  - Using datetime.now() instead of datetime.utcnow() to specify EdgeData’s timestamp
[1.1.2] - 2020-08-24
- Fixed
  - Fix the default timestamp is in local time problem
[1.1.1] - 2020-07-10
- Support to customize the time of data
[1.1.0] - 2020-02-15
- Added
  - Support the “Delsert” action of UploadConfig
  - Add the “RetentionPolicyName” property to DeviceConfig. This will automatically bind the retention policy to a device on the cloud.
- Changed
  - Remove the deprecation properties of NodeConfig: name and description
[1.0.10] - 2018-10-15
[1.0.9] - 2018-10-15

Environment

Installation​

FAQ​

EdgeAgent

Constructor (EdgeAgentOptions options)​

Azure IoT Hub​

Event​

Connect ()​

Disconnect ()​

UploadConfig (ActionType action, EdgeConfig edgeConfig)​

Block Config​

SendData (EdgeData data)​

SendDeviceStatus (EdgeDeviceStatus deviceStatus)​

Property​