128 lines
5.4 KiB
Python
128 lines
5.4 KiB
Python
from AWSIoTPythonSDK.MQTTLib import AWSIoTMQTTClient
|
|
import logging
|
|
import time
|
|
import argparse
|
|
import json
|
|
import os
|
|
from datetime import datetime
|
|
import urllib
|
|
import multiprocessing
|
|
import driver
|
|
import utilities
|
|
def main():
|
|
|
|
AllowedActions = ['both', 'publish', 'subscribe']
|
|
|
|
# Custom MQTT message callback
|
|
def customCallback(client, userdata, message):
|
|
print("Client: ")
|
|
print(client)
|
|
print("User Data: ")
|
|
print(userdata)
|
|
print("Received a new message: ")
|
|
print(message.payload)
|
|
print("from topic: ")
|
|
print(message.topic)
|
|
print("--------------\n\n")
|
|
|
|
|
|
# Read in command-line parameters
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("-e", "--endpoint", action="store", required=True, dest="host", help="Your AWS IoT custom endpoint")
|
|
parser.add_argument("-r", "--rootCA", action="store", required=True, dest="rootCAPath", help="Root CA file path")
|
|
parser.add_argument("-c", "--cert", action="store", dest="certificatePath", help="Certificate file path")
|
|
parser.add_argument("-k", "--key", action="store", dest="privateKeyPath", help="Private key file path")
|
|
parser.add_argument("-p", "--port", action="store", dest="port", type=int, help="Port number override")
|
|
parser.add_argument("-w", "--websocket", action="store_true", dest="useWebsocket", default=False,
|
|
help="Use MQTT over WebSocket")
|
|
parser.add_argument("-id", "--clientId", action="store", dest="clientId", default="basicPubSub",
|
|
help="Targeted client id")
|
|
parser.add_argument("-t", "--topic", action="store", dest="topic", default="dt/hpiot/", help="Targeted topic")
|
|
parser.add_argument("-m", "--mode", action="store", dest="mode", default="both",
|
|
help="Operation modes: %s"%str(AllowedActions))
|
|
parser.add_argument("-M", "--message", action="store", dest="message", default="Hello World!",
|
|
help="Message to publish")
|
|
|
|
args = parser.parse_args()
|
|
host = args.host
|
|
rootCAPath = args.rootCAPath
|
|
certificatePath = args.certificatePath
|
|
privateKeyPath = args.privateKeyPath
|
|
port = args.port
|
|
useWebsocket = args.useWebsocket
|
|
topic = args.topic
|
|
|
|
def jitp_registration():
|
|
#Attempt to connect to AWS IoT Core and start JITP for given certificate
|
|
myAWSIoTMQTTClient = None
|
|
myAWSIoTMQTTClient = AWSIoTMQTTClient(certificateID)
|
|
myAWSIoTMQTTClient.configureEndpoint(host, port)
|
|
myAWSIoTMQTTClient.configureCredentials(rootCAPath, './device1Cert.key', './device1CertAndCACert.pem')
|
|
while True:
|
|
try:
|
|
myAWSIoTMQTTClient.connect()
|
|
myAWSIoTMQTTClient.disconnect()
|
|
break
|
|
except Exception as e:
|
|
logger.info("Didn't connect trying again in 10 seconds: {}".format(e))
|
|
time.sleep(10)
|
|
#Get the config that should be in the database after JITP concludes
|
|
return json.load(urllib.urlopen('https://4ax24ru9ra.execute-api.us-east-1.amazonaws.com/Gamma/HPIoTgetConfig/?certificateID={}'.format(certificateID)))
|
|
|
|
# Configure logging
|
|
logger = logging.getLogger("AWSIoTPythonSDK.core")
|
|
logger.setLevel(logging.INFO)
|
|
streamHandler = logging.StreamHandler()
|
|
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
streamHandler.setFormatter(formatter)
|
|
logger.addHandler(streamHandler)
|
|
|
|
#Checking for main device certificate or making it if absent
|
|
if not os.path.isfile('./device1Cert.pem'):
|
|
os.system('openssl genrsa -out device1Cert.key 2048')
|
|
os.system('openssl req -config server.conf -new -key device1Cert.key -out device1Cert.pem')
|
|
os.system('openssl x509 -req -in device1Cert.pem -CA rootCA.pem -CAkey rootCA.key -CAcreateserial -out device1Cert.pem -days 365 -sha256')
|
|
|
|
if not os.path.isfile('./device1CertAndCACert.pem'):
|
|
os.system('cat device1Cert.pem rootCA.pem > device1CertAndCACert.pem')
|
|
|
|
|
|
certificateID = os.popen('openssl x509 -in device1Cert.pem -outform der | sha256sum').read()[:-4]
|
|
|
|
#Download the config from dynamodb with API call
|
|
logger.info("Attempting to download config file")
|
|
config = {}
|
|
try:
|
|
config = json.load(urllib.urlopen('https://4ax24ru9ra.execute-api.us-east-1.amazonaws.com/Gamma/HPIoTgetConfig/?certificateID={}'.format(certificateID)))
|
|
except Exception as e:
|
|
logger.error(e)
|
|
|
|
#No config in database probably haven't been registered attempt to connect and start JITP
|
|
if 'certificateID' not in config.keys():
|
|
config = jitp_registration()
|
|
|
|
#config = utilities.unmarshal_dynamodb_json(config)
|
|
|
|
|
|
print(config)
|
|
#Get all the device names from the config
|
|
devices = [ele for ele in config.keys() if('device' in ele)]
|
|
|
|
#Build a list of all processes, so that they can be terminated afterwards
|
|
all_processes = []
|
|
for device in devices:
|
|
driver.run(config[device],device,port, host, rootCAPath)
|
|
'''
|
|
process = multiprocessing.Process(target=driver.run, args=(config[device],device,port, host, rootCAPath), name="Process-{}".format(config[device]['locationID']))
|
|
process.start()
|
|
all_processes.append(process)
|
|
logger.info(all_processes)
|
|
for process in all_processes:
|
|
if process.exitcode:
|
|
process.terminate()
|
|
'''
|
|
if __name__ == '__main__':
|
|
main()
|
|
|
|
|