Code:
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.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" )
self.thread=threading.Thread(target=self.readTagOnce,args=(), 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:
data = 0
time.sleep(0.5)
message = bytearray(b'\x0A\xFF\x02\x44')
csm = Uhfthread.checksum(self,message)
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 = Uhfthread.checksum(self,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 = Uhfthread.checksum(self,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 = 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 = {
"uhfdata": self.tagJson,
"laneNo" :"1"
}
except:
print("data not found")
a=Uhfthread()
a.WriteTag()
Data is received from UHF reader and stored in self.json. When I just print without data, I get all the details from self.json; but not in entryrecord().
I want to get the data from self.json and store it in entryrecord.
I'm getting following error:
Exception in thread Thread-157 (readTagOnce):
Traceback (most recent call last):
File "C:\Users\kamle\AppData\Local\Programs\Python\Python310\lib\threading.py", line 1009, in _bootstrap_inner
self.run()
File "C:\Users\kamle\AppData\Local\Programs\Python\Python310\lib\threading.py", line 946, in run
self._target(*self._args, **self._kwargs)
TypeError: Uhfthread.readTagOnce() missing 1 required positional argument: 'data'