sparkplug/send_command_test.py at master · rpiper/sparkplug · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
"""*******************************************************************************
 * Copyright (c) 2021 Ian Craggs
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v2.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *    https://www.eclipse.org/legal/epl-2.0/
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial API and implementation and/or initial documentation
 *******************************************************************************"""


import paho.mqtt.client as mqtt
import time, sys

"""


"""
broker = "localhost"
port = 1883
host_id = "myhost"
edge_node_id = "TCK_edge"
device_id = "TCK_device"

subscribed = False
host_created = False

finished = False
messages = []
def control_on_message(client, userdata, msg):
    messages.append(msg)
    global host_created, finished
    if msg.topic == "SPARKPLUG_TCK/LOG":
        payload = msg.payload.decode("ascii")
        print("***", payload)
        if payload == "Host "+host_id+" successfully created":
            host_created = True
            print("Host created")
    elif msg.topic == "SPARKPLUG_TCK/CONSOLE_PROMPT":
        payload = msg.payload.decode("ascii")
        print(">>>", payload)
    elif msg.topic == "SPARKPLUG_TCK/RESULT":
        print("*** Result ***",  msg.payload)
        finished = True

def control_on_connect(client, userdata, flags, rc):
    print("Control client connected with result code "+str(rc))
    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    control_client.subscribe("SPARKPLUG_TCK/#")

def control_on_subscribe(client, userdata, mid, granted_qos):
    print("Control client subscribed")
    global subscribed
    subscribed = True

publish_count = 0
def control_on_publish(client, userdata, mid):
    print("Control client published")
    global publish_count
    publish_count += 1

control_client = mqtt.Client("sparkplug_control")
control_client.on_connect = control_on_connect
control_client.on_subscribe = control_on_subscribe
control_client.on_publish = control_on_publish
control_client.on_message = control_on_message
control_client.connect(broker, port)
control_client.loop_start()

def publish(topic, payload):
    global publish_count
    publish_count = 0
    print(topic, payload)
    control_client.publish(topic, payload)
    # wait for publish to complete
    while publish_count == 0:
        time.sleep(0.1)

while not subscribed:
    time.sleep(0.1)

# Ensure host application to test is running and connected.  We are using the host simulator here
subscribed = False
messages = []
control_client.subscribe("STATE/"+host_id)
while not subscribed:
    time.sleep(0.1)
count = 0
online = False
while True:
    count += 1
    if count == 10:
        break
    time.sleep(0.1)
    if len(messages) > 0:
        msg = messages[-1]
        payload = msg.payload.decode("ascii")
        print("message", msg.topic, payload)
        if msg.topic == "STATE/"+host_id and payload == "ONLINE":
            online = True
            break

if not online:
    publish("SPARKPLUG_TCK/HOST_CONTROL", "New host "+host_id)
    while not host_created:
        time.sleep(0.1)

publish("SPARKPLUG_TCK/TEST_CONTROL", "New host SendCommandTest "+host_id+" "+edge_node_id+" "+device_id)

while not finished:
    time.sleep(0.2)

print("Stopping MQTT client")
control_client.loop_stop()