How can i use multithreading for my oem_data program for supporting multiple device?

user_11020596 0
Viewed 73

I have made and python program which basically decodes hex data send by the device into a readable form and that decoded data is sent over MQTT to the server now the code is working fine with the single device but when I connect more than one device the problem starts to occur the program is unable to start the thread for hex decode program

In this below program oem_data file is the hex data decoder

this is my program

import socket
import os
import logging
from _thread import *
from oem_data import packet_splitter,finaldic
import paho.mqtt.client as mqtt
import json
broker_ip="127.0.0.1"
broker_port=1883

ServerSideSocket = socket.socket()
HOST = '192.168.0.108'  # Standard loopback interface address (localhost)
PORT = 65432        # Port to listen on (non-privileged ports are > 1023)

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
logger.info(" Welcome to server")
ThreadCount = 0
try:
    ServerSideSocket.bind((HOST, PORT))
except socket.error as e:
    print(str(e))

print('Socket is listening..')
ServerSideSocket.listen(5)

def multi_threaded_client(connection):
    connection.send(str.encode('Server is working:'))
    while True:
           data = connection.recv(2048)
           a = data.hex().upper()
           y = packet_splitter(a)  # hex data is pass to the oem_data file for convertion

          topic = "device/{}/message".format(finaldic['data2']['tlr']['IMEI'])
          payload = {
              "param1": finaldic,  # converted data from oem_data to be sent to MQTT
            }
         if client.is_connected():
              client.publish(topic, json.dumps(payload))
              logger.info("{} - message sent: {} - {}".format(client, topic, json.dumps(payload)))
         else:
            logger.info("{} is not connected to the broker!".format(client))
    
        if not data:
            break
        connection.sendall(bytes.fromhex(y))
   connection.close()

while True:
    Wlient, address = ServerSideSocket.accept()
    client = mqtt.Client(str(address))  # create new instance
    client.connect(broker_ip, broker_port, 60)  # connect to broker
    client.loop_start()
    print('Connected to: ' + address[0] + ':' + str(address[1]))
    start_new_thread(multi_threaded_client, (Wlient, ))
    ThreadCount += 1
    print('Thread Number: ' + str(ThreadCount))
ServerSideSocket.close()

so my question is how can I start the thread for the oem_data converter file for individual devices which are connecting to the socket I am new to this so please help me out.

this is my oem_data code:

import re
import json
from posmessage import position_message,protocol_header,positiondict
from terminal_login import terminal_login_request,logindic


def packet_splitter(data1):
       command_id={'AA02':terminal_login_request,'AA00':position_message}
        # Replace all occurrences of character s with an empty string
        data=re.sub('DBDC', 'C0', data1)
        packet_header=data[2:26]
        packet_payload=data[26:-2]

        command_data=protocol_header(packet_header)
        for k,v in command_id.items():
             if k == command_data[0]:
                 return command_id[k](packet_payload,command_data[1])

 finaldic={}
 # packet=packet_splitter(position_data)
 finaldic["data1"]=positiondict
 finaldic["data2"]=logindic
mqtt python-3.x sockets multithreading python
1 Answers

The built-in socketserver module makes it easier to build threaded servers. I can't guarantee this'll work for you out of the box (e.g. if you really do need one MQTT client per incoming connection instead of a shared one), but it's a starting point.

Also, that finaldic import seems suspect -- if it's really a module-level variable and should be specified per-client, it definitely will not work in a threaded context!

import socketserver
import logging
from oem_data import packet_splitter, finaldic
import paho.mqtt.client as mqtt
import json

broker_ip = "127.0.0.1"
broker_port = 1883
HOST = "192.168.0.108"
PORT = 65432

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
logger.info(" Welcome to server")
mqtt_client = mqtt.Client()
mqtt_client.connect(broker_ip, broker_port, 60)
mqtt_client.loop_start()


class ConnectionHandler(socketserver.StreamRequestHandler):
    def handle(self) -> None:
        connection = self.request
        connection.send(str.encode("Server is working:"))
        while True:
            data = connection.recv(2048)
            if not data:
                break
            a = data.hex().upper()
            y = packet_splitter(a)  # hex data is pass to the oem_data file for convertion

            topic = "device/{}/message".format(finaldic["data2"]["tlr"]["IMEI"])
            payload = {
                "param1": finaldic,  # converted data from oem_data to be sent to MQTT
            }

            if mqtt_client.is_connected():
                mqtt_client.publish(topic, json.dumps(payload))
                logger.info("{} - message sent: {} - {}".format(mqtt_client, topic, json.dumps(payload)))
            else:
                logger.info("{} is not connected to the broker!".format(mqtt_client))

            connection.sendall(bytes.fromhex(y))
            connection.close()


server = socketserver.ThreadingTCPServer(
    server_address=(HOST, PORT),
    RequestHandlerClass=ConnectionHandler,
    bind_and_activate=True,
)
server.serve_forever(1)
AKX_516850
Related