<jupyter_start><jupyter_text>*This notebook contains material from [cbe61622](https://jckantor.github.io/cbe61622);content is available [on Github](https://github.com/jckantor/cbe61622.git).*     9.1 Publishing an Experiment via MQTTThis notebook is part of a proof concept study regarding use of the Internet of Things (IoT) as infrastructure for engineering teaching laboratories. This notebook uses the [SimPy](https://simpy.readthedocs.io/en/latest/) library a create real-time simulation of a hypothetical experiment. The experiment responds to inputs and publishes data to a remote user via an an MQTT broker. In actual use, the content of this notebook would be implemented in an attached device such as an Arduino, Particle, or Raspberry Pi, with attached sensors.The companion notebook ??? demonstrates how a remote client could interact with the experiment via MQTT.  9.1.1 InstallationsThe following installations are required for use on Google Colab.<jupyter_code>!pip install paho-mqtt
!pip install simpy<jupyter_output>Requirement already satisfied: paho-mqtt in /Users/jeff/opt/anaconda3/lib/python3.7/site-packages (1.5.1)
Requirement already satisfied: simpy in /Users/jeff/opt/anaconda3/lib/python3.7/site-packages (4.0.1)<jupyter_text>9.1.2 Publishing a real-time simulation via MQTTTopics:| topic | messages || :-- | :-- || `cbe-virtual-lab/command` | start and stop experiments || `cbe-virtual-lab/expt-name/data` | topic  9.1.2.1 Proof of ConceptHere we experiment with encapsulating the experiment as a standalone class. This is set up so that upon receiving an appropriate message from the remote user, a new instance of the experiment is created and run.<jupyter_code>%matplotlib inline
import paho.mqtt.client as mqtt
import paho.mqtt.publish as publish
import time
import matplotlib.pyplot as plt
import numpy as np
from IPython import display

# select experiment duration

# set up client to interact with cbe-virtual-laboratory

class CBEClient(mqtt.Client):
    
    def __init__(self, recv="", send=""):
        super().__init__()
        self.host = "mqtt.eclipse.org"
        self.recv = recv
        self.send = send

    def on_connect(self, client, userdata, flags, rc):
        print(f"Connected: {self.host} with return code {rc}")
        if self.recv:
            self.subscribe(self.recv, qos=2)
            print(f"Subscribed: {self.recv}")
        
    def on_message(self, client, userdata, msg):
        payload = json.loads(msg.payload.decode("utf-8"))
        print(f"Receieved: {payload} from {self.recv}")
    
    def connect(self):
        super().connect(host=self.host, port=1883, keepalive=60)
        
    def publish(self, payload):
        if self.send:
            super().publish(self.send, payload=payload)
            print(f"Sent: {payload} to {self.send}")
        else:
            print("No send topic has been specified.")
        
    def __enter__(self):
        self.connect()
        self.loop_start()
        time.sleep(0.5)
        print(f"Loop Started: {self}")
        return self
        
    def __exit__(self, exc_type, exc_value, exc_traceback):
        self.loop_stop()
        print(f"Loop Stopped: {self}")
        
def on_message(client, userdata, msg):
    payload = json.loads(msg.payload.decode("utf-8"))
    send = "/".join(["cbe-virtual-laboratory", payload["client_id"]])
    print(send)
    print(f"Message: {payload}")
    with CBEClient(send=send) as expt:
        for k in range(payload["duration"]):
            x = 0
            y = 0
            expt.publish(json.dumps({"time": k, "x" : x, "y" : y}))
            time.sleep(1)

# listen for command on the command channel
with CBEClient(recv = "cbe-virtual-laboratory/command/#") as cbe:
    cbe.on_message = on_message
    time.sleep(30)<jupyter_output>Connected: mqtt.eclipse.org with return code 0
Subscribed: cbe-virtual-laboratory/command/#
Loop Started: <__main__.CBEClient object at 0x7fe572670e10>
cbe-virtual-laboratory/12
Message: {'client_id': '12', 'duration': 10}
Connected: mqtt.eclipse.org with return code 0
Loop Started: <__main__.CBEClient object at 0x7fe5733a2910>
Sent: {"time": 0, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 1, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 2, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 3, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 4, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 5, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 6, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 7, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 8, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Sent: {"time": 9, "x": 0, "y": 0} to cbe-virtual-laboratory/12
Loop Stopped: <__main__.CBEClient object[...]<jupyter_text>9.1.3 Version 1T<jupyter_code>import time
import simpy
import paho.mqtt.publish as publish
import paho.mqtt.client as mqtt
import json

class Experiment():

    def __init__(self, topic, duration):
        self.topic = topic
        self.duration = duration
        self.env = simpy.rt.RealtimeEnvironment(factor=1)
        self.proc = self.env.process(self.process())
        self.client = mqtt.Client()
        self.client.on_connect = self.on_connect
        self.client.on_publish = self.on_publish

    def on_connect(self, client, userdata, flags, rc):
        print(f"Connected with result code {rc}")
        self.client.subscribe(self.topic)

    def on_publish(self, client, userdata, result):
        print(f"{client} published with result code {result}")

    def process(self):
        t_start = time.perf_counter()
        t = 0
        y = 2.0
        while True:
            msg = f"{round(t,2)},{y:5.2f}"
            self.client.publish(self.topic, msg)
            yield self.env.timeout(1 - (t - round(t, 0)))
            t = time.perf_counter() - t_start
            y -= 0.1*y

    def run(self, client):
        print(f"Experiment started by {client}")
        self.client.connect("mqtt.eclipse.org", 1883, 60)
        self.env.run(until=self.duration)
        self.client.disconnect()
        print("End experiment.")

# set up client to wait for messages on 
#     cbe-virtual-laboratory/command/#
# expect a message with a specified experiment duration

def on_connect(client, userdata, flags, rc):
    print(f"Connected with result code {rc}")
    client.subscribe("cbe-virtual-laboratory/command/#")

def on_message(client, userdata, msg):
    print(f"Received {msg.payload} from {msg.topic}")
    data = json.loads(msg.payload.decode("utf-8"))
    duration = data['duration']
    expt = Experiment("cbe-virtual-laboratory/expt", duration)
    expt.run(client)

# setup client
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

# connect client to broker
client.connect("mqtt.eclipse.org", 1883, 60)

# start a non-blocking thread to wait for messages
client.loop_start()

# prove the loop is non-blocking
for k in range(20):
    print(k)
    time.sleep(1)

# don't leave a zombie thread behind
client.loop_stop()<jupyter_output>0
Connected with result code 0
1
2
3
Received b'{"duration": 10}' from cbe-virtual-laboratory/command
Experiment started by <paho.mqtt.client.Client object at 0x7fe57268dd10>
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 1
4
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 2
5
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 3
6
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 4
7
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 5
8
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 6
9
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 7
10
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 8
11
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 9
12
<paho.mqtt.client.Client object at 0x7fe5726a36d0> published with result code 10
13
[...]
{'fraction_non_alphanumeric': 0.1208723509117792, 'fraction_numerical': 0.036347954657466736, 'mean_word_length': 3.5119510839355197, 'pattern_counts': {'":': 32, '<': 24, '<?xml version=': 0, '>': 23, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '3', 'dir': 'jupyter-structured-clean-dedup', 'id': '524506', 'license': 'MIT', 'n_tokens_mistral': 2865, 'n_tokens_neox': 2697, 'n_words': 782, 'path': 'docs/09.01-Publishing_an_Experiment_via_MQTT.ipynb', 'repo_name': 'jckantor/cbe61622', 'size': '8116'}