Update 3 files

- /lab4_emulator.py
- /create_test_thing.py
- /lambda_function.py

create_test_thing.py

+################################################### Connecting to AWS
+import boto3
+
+import json
+################################################### Create random name for things
+import random
+import string
+
+################################################### Parameters for Thing
+thingArn = ''
+thingId = ''
+thingName = ''.join([random.choice(string.ascii_letters + string.digits) for n in range(15)])
+print(thingName)
+defaultPolicyName = 'TestPolicy'
+###################################################
+
+def createThing():
+  global thingClient
+  global thingId
+  global thingArn
+  thingResponse = thingClient.create_thing(
+      thingName = thingName
+  )
+  data = json.loads(json.dumps(thingResponse, sort_keys=False, indent=4))
+  for element in data: 
+    if element == 'thingArn':
+        thingArn = data['thingArn']
+    elif element == 'thingId':
+        thingId = data['thingId']
+    createCertificate()
+
+def createCertificate():
+	global thingClient
+	certResponse = thingClient.create_keys_and_certificate(
+			setAsActive = True
+	)
+	data = json.loads(json.dumps(certResponse, sort_keys=False, indent=4))
+	for element in data: 
+			if element == 'certificateArn':
+					certificateArn = data['certificateArn']
+			elif element == 'keyPair':
+					PublicKey = data['keyPair']['PublicKey']
+					PrivateKey = data['keyPair']['PrivateKey']
+			elif element == 'certificatePem':
+					certificatePem = data['certificatePem']
+			elif element == 'certificateId':
+					certificateId = data['certificateId']
+							
+	with open('public.key', 'w') as outfile:
+			outfile.write(PublicKey)
+	with open('private.key', 'w') as outfile:
+			outfile.write(PrivateKey)
+	with open('cert.pem', 'w') as outfile:
+			outfile.write(certificatePem)
+
+	response = thingClient.attach_policy(
+			policyName = defaultPolicyName,
+			target = certificateArn
+	)
+	response = thingClient.attach_thing_principal(
+			thingName = thingName,
+			principal = certificateArn
+	)
+	response = thingClient.add_thing_to_thing_group(
+		thingGroupName = "TestThingGroup",
+		thingGroupArn = "arn:aws:iot:us-east-2:673050318892:thinggroup/TestThingGroup",
+		thingName = thingName,
+		thingArn = thingArn,
+	)
+
+thingClient = boto3.client('iot', region_name='us-east-2')
+createThing()
+createCertificate()
\ No newline at end of file

lab4_emulator.py

+# Import SDK packages
+from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
+import time
+import json
+import pandas as pd
+import numpy as np
+
+
+#TODO 1: modify the following parameters
+#Starting and end index, modify this
+device_st = 0
+device_end = 5
+
+#Path to the dataset, modify this
+data_path = "vehicle{}/vehicle{}.csv"
+
+#Path to your certificates, modify this
+certificate_formatter = "vehicle{}/cert.pem"
+key_formatter = "vehicle{}/private.key"
+
+
+class MQTTClient:
+    def __init__(self, device_id, cert, key):
+        # For certificate based connection
+        self.device_id = str(device_id)
+        self.state = 0
+        self.client = AWSIoTMQTTClient(self.device_id)
+        #TODO 2: modify your broker address
+        self.client.configureEndpoint("a3l1br4lpomzf8-ats.iot.us-east-2.amazonaws.com", 8883)
+        self.client.configureCredentials("./keys/AmazonRootCA1.pem", key, cert)
+        self.client.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
+        self.client.configureDrainingFrequency(2)  # Draining: 2 Hz
+        self.client.configureConnectDisconnectTimeout(10)  # 10 sec
+        self.client.configureMQTTOperationTimeout(5)  # 5 sec
+        self.client.onMessage = self.customOnMessage
+        
+
+    def customOnMessage(self,message):
+        #TODO 3: fill in the function to show your received message
+        j = json.loads(message.payload)
+        print(f"Vehicle {self.device_id} max co2 is {j}")
+
+
+    # Suback callback
+    def customSubackCallback(self,mid, data):
+        #You don't need to write anything here
+        pass
+
+
+    # Puback callback
+    def customPubackCallback(self,mid):
+        #You don't need to write anything here
+        pass
+
+
+    def publish(self, topic="vehicle/emission/data"):
+    # Load the vehicle's emission data
+        df = pd.read_csv(data_path.format(self.device_id,self.device_id))
+        for index, row in df.iterrows():
+            # Create a JSON payload from the row data
+            payload = json.dumps(row.to_dict())
+            
+            # Publish the payload to the specified topic
+            # print(f"Publishing: {payload} to {topic}")
+            self.client.publishAsync(topic, payload, 0, ackCallback=self.customPubackCallback)
+            
+            # Sleep to simulate real-time data publishing
+            time.sleep(1)
+            
+
+
+
+# print("Loading vehicle data...")
+# data = []
+# for i in range(device_st, device_end):
+#     a = pd.read_csv(data_path.format(i))
+#     data.append(a)
+
+print("Initializing MQTTClients...")
+clients = []
+for device_id in range(device_st, device_end):
+    print(certificate_formatter.format(device_id,device_id),key_formatter.format(device_id,device_id))
+    client = MQTTClient(device_id,certificate_formatter.format(device_id,device_id),key_formatter.format(device_id,device_id))
+    client.client.connect()
+    client.client.subscribe(f"vehicle/veh{device_id}",0, lambda x,y,z: {})
+    clients.append(client)
+ 
+
+while True:
+    print("send now?")
+    x = input()
+    if x == "s":
+        for i,c in enumerate(clients):
+            c.publish()
+
+    elif x == "d":
+        for c in clients:
+            c.client.disconnect()
+        print("All devices disconnected")
+        exit()
+    else:
+        print("wrong key pressed")
+
+    time.sleep(3)
\ No newline at end of file

lambda_function.py

+# Import SDK packages
+from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
+import time
+import json
+import pandas as pd
+import numpy as np
+
+cert = "grass_cert.pem"
+key = "grass_private.key"
+
+class MQTTClient:
+    def __init__(self, device_id, cert, key):
+        # For certificate based connection
+        self.device_id = str(device_id)
+        self.state = 0
+        self.client = AWSIoTMQTTClient(self.device_id)
+        self.client.configureEndpoint("a3l1br4lpomzf8-ats.iot.us-east-2.amazonaws.com", 8883)
+        self.client.configureCredentials("./keys/AmazonRootCA1.pem", key, cert)
+        self.client.configureOfflinePublishQueueing(-1)  # Infinite offline Publish queueing
+        self.client.configureDrainingFrequency(2)  # Draining: 2 Hz
+        self.client.configureConnectDisconnectTimeout(10)  # 10 sec
+        self.client.configureMQTTOperationTimeout(5)  # 5 sec
+        self.client.onMessage = self.customOnMessage
+        self.co2_map = dict()
+        
+
+    def customOnMessage(self,message):
+        msg = json.loads(message.payload)
+        co2 = float(msg['vehicle_CO2'])
+        id = msg['vehicle_id']
+        max_co2 = self.co2_map.get(id, 0.0)
+        max_co2 = max(max_co2, co2)
+        self.co2_map[id] = max_co2
+        self.respond(id, max_co2)
+        print(f"Vehicle {id} max co2 is {max_co2}")
+        # print(f"client {self.device_id} received payload {message.__dict__} from topic {"topic"}")
+
+
+    # Suback callback
+    def customSubackCallback(self,mid, data):
+        #You don't need to write anything here
+        pass
+
+
+    # Puback callback
+    def customPubackCallback(self,mid):
+        #You don't need to write anything here
+        pass
+
+    def respond(self, vehicle_id, co2):
+        payload = json.dumps({'MaxCO2':co2})
+        self.client.publishAsync(f"vehicle/{vehicle_id}", payload, 0)
+
+    def publish(self, topic="vehicle/emission/data"):
+        pass
+            
+client = MQTTClient(255,cert,key)
+client.client.connect()
+client.client.subscribe("vehicle/emission/data",0, lambda x,y,z:{})
+ 
+
+while True:
+    time.sleep(3)
\ No newline at end of file