mosquitto_ex: mosq_feed_ex.py

#!/usr/bin/env python3

# -*- coding: utf-8 -*-

import paho.mqtt.client as paho
from subprocess import Popen, PIPE
import os
import json
import logging
import queue
import threading

############################################
#    Initial global variable               #
############################################

q = queue.Queue()
num_threads = 3
ping_warn = 800
ping_crit = 900
port_warn = 1
port_crit = 2

############################################
#    Initial logging to file               #
############################################
# Gets or creates a logger
logger = logging.getLogger("mosq_feed_actDas_7_queue.py")
# set log level
logger.setLevel(logging.DEBUG)
# define file handler and set formatter
file_handler = logging.FileHandler('/tmp/mosq_feed_actDas_7_queue.log')
formatter = logging.Formatter(
    '%(asctime)s : %(levelname)s : %(name)s : %(message)s')
file_handler.setFormatter(formatter)
# add file handler to logger
logger.addHandler(file_handler)
logger.info("log start")

############################################
#    function utility                      #
############################################

def is_json(myjson):
    print("is_json()")
    try:
        json_object = json.loads(myjson)
    #except ValueError as e:
    except:
        print("is_json() false")
        return False
    #print("is_json() True")
    return True

def do_stuff(q, thread_no):
    while True:
        task = q.get()
        #print( "*********** g.get()=", q.get() )
        process_id(task)
        q.task_done()
        #print(f'Thread #{thread_no} is doing task #{task} in the queue.')
        print(f'Thread #{thread_no} is doing task in the queue.')

def on_subscribe(client, userdata, mid, granted_qos):
    print("Subscribed: "+str(mid)+" "+str(granted_qos))

def check_cmd2( id, service, value ):
    print("check_cmd2()")
    print("id = {}".format(id))
    #logger.info("check_cmd2()")
    myCmd = ""

    #print("delay time=%d" % (delayT))
    if ( service == "pass_ping" ):
         #print ("pass_ping=%s" % (value) )
         code_status = check_ping( value )
         myCmd = "./send_gearman_temperature.sh %s pass_ping '%s msec. | pass_ping=%s;%s;%s;0;100' %d" % (id, value, value, ping_warn, ping_crit, code_status)
         #output = os.popen(myCmd).read()

    elif( service == "rssi" ):
         #print ("Temp=%s" % (value) )
         code_status = check_dist( value )
         #myCmd = "./send_gearman_temperature.sh %s Temp '%s C. | Temp=%s;%s;%s;0;100' %d" % (id, value, value, delay_warn, delay_crit, code_status)
         myCmd = "curl -X POST -H \"Content-Type: application/json\" \"http://172.19.0.4:8080\" -d '{\"deviceID\":\"%s\", \"%s\":%d}' " %  (device, service, value)
         print(myCmd)
         #output = os.popen(myCmd).read()
    elif( id.find("BU") == 0 or id.find("RU") == 0):
         myCmd = ""
         myCmd = "./send_gearman_temperature.sh \"%s\" \"%s\"  \"%s | status=%s;%s;%s;0;2\" %d" % (id, service, value, value, port_warn, port_crit, 0)
         print(myCmd)
         #output = os.popen(myCmd).read()
    else:
         print("service not match")
    logger.info(myCmd)


def verify_msg(raw_msg):
    try:
        id = raw_msg.topic.split("/")
        print(id[-1])
    except:
        print("cannot split str")
    print("------------")
    text = raw_msg.payload.decode("utf-8")
    #print(text)
    logger.info(text)
    #logger.info(text)
    if (is_json(text) == False):
        print("not json")
        logger.info("Not JASON")
        return -1
    else:
        print("msg is json")
        logger.info("MSG is JSON")
        # dumps the json object into an element
        #json_str = json.dumps(text)
        # print(json_str)
        # load the json to a string
        good_msg = json.loads(text)
        return good_msg




def process_id(resp):
    print("process_id() ")
    resp = verify_msg(resp)
    if (resp == -1):
        return -1
    try:
        key = list(resp)
    except:
        print("list(resp) error resp=", resp)
        return -1
    print( "key = {}".format(key) )
    if ( list(resp)[0] == "module" ):
         serviceName = list(resp)[3]
         print( "serviceName = {}".format(serviceName) )
    else:
         print("no module service Name in msg resp")
         return -1
    print ("serial_number:%s siteID:%s serviceName:%s value:%s " % ( resp['serial_number'], resp["site_id"], serviceName, resp[serviceName]) )
    servName = resp[serviceName]
    #print ("\n ***  serviceName = {}".format(servName) )
    #print ("\n ***** type servName ", type(servName ) )
    key = list(servName)
    #print( "key = {}".format(key) )
    #print( "key[0] = {}\n key[1] = {}\n".format( key[0], key[1] ) )
    #for msg_json in resp:
    #    print( "user = {}".format(msg_json ))
    #print("module = {}".format( resp["module"] ))
    #print("serial_number = {}".format( resp["serial_number"] ))
    serial_number = resp["serial_number"]
    #print("type = {} serial_number = {}".format( type(serial_number),  serial_number) )
    #print("site_id = {}".format( resp["site_id"] ))
    #print("digital_service = {}".format( resp["digital_service"] ))
    #print("analog_service = {}".format( resp["analog_service"] ))

    digital_service = resp["digital_service"]
    analog_service = resp["analog_service"]
    #print("digital_service[0] = {} \n digital_service[1] = {} ".format(digital_service[0], digital_service[1]) )
    for msg in digital_service:
        #print(msg)
        #print(type(msg))
        print( msg["service"], msg["value"], msg["filter"] )
        check_cmd2( resp["serial_number"], msg["service"], msg["value"]  )
        #is_json(msg2)

    for msg in analog_service:
        #print(msg)
        print( msg["service"], msg["value"], msg["filter"] )
        check_cmd2( resp["serial_number"], msg["service"], msg["value"]  )



def on_message(client, userdata, msg):
    global q
    q.put(msg) #put messages on queue
    print("put msg in Queue")
    #process_id(resp)
    print("------------")
    q.join()

############################################
#    Create thread worker waiting to       #
#    read the msg in queue                 #
############################################
for i in range(num_threads):
  print("worker start ....")
  worker = threading.Thread(target=do_stuff, args=(q, i,), daemon=True)
  worker.start()


################################################
#             MQTT Broker config               #
################################################
user = 'xx:xxxxx'
password = 'yyyyyyy'
keepAlive = 3
client = paho.Client(client_id="id20210903", clean_session=False)
client.username_pw_set(user, password=password)  # set username and password
client.on_subscribe = on_subscribe
client.on_message = on_message
client.connect("mq1.mycompany.com", 1883, keepAlive, bind_address = "" )
client.subscribe("/mycompany/ActDAS/#", qos=1)
client.loop_forever()