Python多线程UHF读卡器报错：缺少必填位置参数'data'

一、修复TypeError: readTagOnce()缺少'data'参数 #

问题根源 #

1. readTagOnce方法声明了必填参数data，但启动线程时未传递该参数；
2. 方法内部同时出现data=0（赋值为整数）和data.put(value)（队列操作）的矛盾逻辑，说明data应为队列实例，但未正确初始化传递。

修复操作 #

• 在类初始化方法中添加队列实例：

  def __init__(self):
      self.connected = True 
      self.tagJson = {}
      self.previousTag = ""
      self.q = Queue()  # 新增队列初始化
      self.thread=threading.Thread(target=self.connect,args=(), daemon=True)
      self.thread.start()
      self.thread.join()
  

• 启动readTagOnce线程时传入队列参数：

  self.thread=threading.Thread(target=self.readTagOnce,args=(self.q,), daemon=True)
  

• 修改readTagOnce方法，删除错误的data=0赋值，统一使用队列：

  def readTagOnce(self, data):  
      try:
          while True:
              time.sleep(0.5)  # 移除data=0
              # 后续代码保持不变，仅将data作为队列使用
  

二、修复self.tagJson无法存入enter_record的问题 #

问题根源 #

WriteTag方法中仅定义了enter_record局部变量，未做任何持久化或存储操作，变量执行完即销毁，导致数据无法留存。

修复操作 #

在WriteTag方法中，定义enter_record后添加存储逻辑，比如将其设为类属性：

try:
    self.enter_record = {
        "uhfdata": self.tagJson,
        "laneNo" :"1"
    }
    # 可选：添加写入文件/数据库的逻辑，例如
    # import json
    # with open("record.json", "w") as f:
    #     json.dump(self.enter_record, f)
except:
    print("data not found")

另外，调用WriteTag时需传递所需参数，否则会触发新的参数缺失错误：

a=Uhfthread()
a.WriteTag(None, None)  # 根据实际需求传入data和instance参数

修改后的完整代码 #

from re import X
import socket
import threading
import time
from queue import Queue

class Uhfthread():
    def __init__(self):
        self.connected = True 
        self.tagJson = {}
        self.previousTag = ""
        self.q = Queue()  # 初始化队列
        self.thread=threading.Thread(target=self.connect,args=(), daemon=True)
        self.thread.start()
        self.thread.join()

    def connect(self):
        while self.connected:
            try:
                self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                self.s.connect(("192.168.1.232", 100))
                self.connected = True
                print( "connected" )
                # 传递队列参数给readTagOnce
                self.thread=threading.Thread(target=self.readTagOnce,args=(self.q,), daemon=True)
                self.thread.start()
                self.thread.join()
            except Exception as i:
                print(i)
                self.connected = False    
                print("connection lost... reconnecting")  
                while not self.connected: 
                    try:  
                        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                        self.s.connect(("192.168.1.232", 100))  
                        self.connected = True  
                        print( "re-connection successful" )  
                    except socket.error:  
                        time.sleep(2)
                        print("uhf is not connected")

    def checksum(self,msg):
        sm = 0
        for i in range(0, len(msg)):
            sm = sm + msg[i]
        high, low = divmod(int(sm), 0x100)
        csm = (0xff - low + 1)
        return csm

    def readTagOnce(self, data):  
        try:
            while True:
                time.sleep(0.5)  # 移除错误的data=0赋值
                message = bytearray(b'\x0A\xFF\x02\x44')
                csm = self.checksum(message)  # 改为实例方法调用，无需传self
                message.append(csm)
                self.s.send(message)
                msg = self.s.recv(32)
                msgStr = msg.hex()
                message = bytearray(b'\x0A\xFF\x0B\x84\x0E\x00\x00\x00\x08\x00\x0C\x00\x20')
                csm = self.checksum(message)  # 同上修改
                message.append(csm)
                self.s.send(message)
                msg = self.s.recv(32)
                msgStr = msg.hex()
                if str(msgStr) == '0be80400000108':  
                    message = bytearray(b'\x0A\xFF\x03\x41\x10')
                    csm = self.checksum(message)  # 同上修改
                    message.append(csm)
                    self.s.send(message)
                    msg = self.s.recv(1024)
                    msgStr = msg.hex()
                    uhfTag = msgStr[14:38]
                    res = "{0:08b}".format(int(uhfTag, 16)).rjust(96, '0')
                    fasTageCheck = int(res[14:38], 2)
                    print('fastag--> ',fasTageCheck)
                    if fasTageCheck == 8907272:
                        print("FASTAG DETECTED")
                        if self.previousTag != uhfTag:
                            self.previousTag = uhfTag
                            uhfEpcId = msgStr[38:70]
                            print(uhfEpcId)
                            uhfTid = msgStr[70:118]
                            print(uhfTid)
                            uhfUserData = msgStr[118:246]
                            self.tagJson['uhfTag'] = uhfTag
                            self.tagJson['uhfEpcId'] = uhfEpcId
                            self.tagJson['uhfTid'] = uhfTid
                            self.tagJson['uhfUserData'] = uhfUserData
                            print(self.tagJson)
                            value=[uhfEpcId,uhfTid,uhfUserData]       
                            data.put(value)  # 队列存储数据
                            print(data) 
                        else:
                            print("same TAG")
                            uhfEpcId = msgStr[38:70]
                            uhfTid = msgStr[70:118]
                            uhfUserData = msgStr[118:246]
                            self.tagJson['uhfTag'] = uhfTag
                            self.tagJson['uhfEpcId'] = uhfEpcId
                            self.tagJson['uhfTid'] = uhfTid
                            self.tagJson['uhfUserData'] = uhfUserData
                            print(self.tagJson)
                            value=[uhfEpcId,uhfTid,uhfUserData]  
                            data.put(value) 
                            print(data)      
                    else:
                        print('Not a FASTAG')
        except Exception as i:
            print("uhf data not found",i)
    def WriteTag(self,data,instance):
        try:
            if self.q.empty():
                print("data not found",data)
            else:
                h=self.q.get()
                print(h)
        except:
                print("blank data sent")
        try:
            # 将enter_record设为类属性，方便外部访问或后续存储
            self.enter_record = {
            "uhfdata": self.tagJson,
            "laneNo" :"1"
            }
            # 可选：添加持久化逻辑，比如写入文件
            # import json
            # with open("uhf_records.json", "a") as f:
            #     json.dump(self.enter_record, f)
            #     f.write("\n")
        except:
            print("data not found")
a=Uhfthread()
a.WriteTag(None, None)  # 传递所需参数

额外优化说明 #

• 将Uhfthread.checksum(self,message)改为self.checksum(message)，因为checksum是实例方法，无需通过类名调用并传递self。

内容的提问来源于stack exchange，提问作者SHRUTI PARAB