Corrupted and wrong data on MB560-VL

[yassine-fy94]

Hello,

when i try to get the attendance data from my device using get_attendance(), i get corrupted data.

this is my function it works fine with other devices but not with the MB560-VL :

!/usr/bin/python3

from zk import ZK import json import argparse import sys import traceback import time

def MyFunction(): try:

------------------------------- les arguments ------------------------------

    parser = argparse.ArgumentParser(description='ZK Basic Reading Tests')
    parser.add_argument('-a', '--address', help='PYZK IP', default=None)
    parser.add_argument('-p', '--port', type=str, help='device port [4370]', default=4370)
    parser.add_argument('-pro', '--protocole', type=str, help='protocole', default='tcp')
    parser.add_argument('-ping', '--ping', type=str, help='if ping or not', default='no_ping')
    args = parser.parse_args()
    conn = None
    # Maximum number of retry attempts
    MAX_RETRIES = 3
    retries = 0
    tout_les_pointages = []
    attendance = None
    # --------------------------- les autres variables --------------------------- #
    force_udp = True if args.protocole=="udp" else False
    ping = True if args.ping=="no_ping" else False # ommit_ping = True donc pas de ping
    while retries < MAX_RETRIES:
        try:
            # create ZK instance
            zk = ZK(args.address, port=int(args.port), timeout=10, password=0, force_udp=force_udp, ommit_ping=ping)
            # connect to device
            conn = zk.connect()
            # Si la connection est ok
            if conn:
                # disable device, this method ensures no activity on the device while the process is run
                conn.disable_device()
                # On stoque les pointages dans une liste
                attendance = conn.get_attendance()
                for att in attendance:
                    tout_les_pointages.append({'created_at':"{}".format(att.timestamp),'matricule':att.user_id,'pointeuse':args.address})
                # re-enable device after all commands already execute
                conn.enable_device()
                conn.disconnect()
                conn = None
                # Renvoie du resultat
                print(json.dumps({'tout_les_pointages':tout_les_pointages,'connexion':'ok'}))
                break  # exit the while loop when connection and operations are successful
            # Si la connection n'est pas ok
            else:
                conn = None
                zk = None
                retries += 1
                time.sleep(3)  # wait for 3 second before retrying  
                # if an exception occurs during the connection or operation, print the error and retry
        except:
            conn = None
            zk = None
            retries += 1
            time.sleep(3)  # wait for 3 second before retrying  
    # If we've exhausted the retry attempts, print an error message
    if retries == MAX_RETRIES:
        print(json.dumps({'connexion':'no', 'error':'maximum retry attempts reached'}))    
except :
    exc_info = sys.exc_info()
    print(json.dumps({'connexion':'no' ,'erreur':'except','message':''.join(traceback.format_exception(*exc_info))}))

Appel à la fonction de syncrhonisation

MyFunction() it return something like :

{ "tout_les_pointages": [ { "created_at": "2023-06-27 18:48:54", "matricule": "3", "pointeuse": "device_one" }, { "created_at": "2000-01-01 00:00:00", "matricule": "55", "pointeuse": "device_one" }, { "created_at": "2000-01-01 00:00:00", "matricule": "", "pointeuse": "device_one" }, { "created_at": "2000-02-08 16:26:04", "matricule": "", "pointeuse": "device_one" }, { "created_at": "2027-10-09 22:21:51", "matricule": "", "pointeuse": "device_one" }, { "created_at": "2000-01-01 00:00:04", "matricule": "", "pointeuse": "device_one" }, { "created_at": "2000-01-01 00:00:00", "matricule": "", "pointeuse": "device_one" } ], "connexion": "ok" } "Firmware Version": "Ver 6.60 Aug 19 2021", "Device Name": "MB560-VL/ID"

[kurenai-ryu]

please include verbose mode output, AND, pcap capture file to compare, but probably related to ol' #56

I think it's the third time this issue is found, If you can confirm that the approach is ok, I'll make a patch and merge it.

[yassine-fy94]

It does not show any error on the output, and all other function are working, only getting data is corrupted, i ve had the same issue on a MA300 but i applied one of your solution and i updated my scripts and it works but for the MB560-vl it still showing wrond data. this is my current function ( it works fine with MA300, IN01, F18, F22, K14,K40, MB360)

def get_attendance(self,if_ma300=False): """ return attendance record

    :return: List of Attendance object
    """
    self.read_sizes()
    if self.records == 0:
        return []
    users = self.get_users()
    if self.verbose: print (users)
    attendances = []
    attendance_data, size = self.read_with_buffer(const.CMD_ATTLOG_RRQ)
    if size < 4:
        if self.verbose: print ("WRN: no attendance data")
        return []
    total_size = unpack("I", attendance_data[:4])[0]
    record_size = total_size/self.records
    if self.verbose: print ("record_size is ", record_size)
    attendance_data = attendance_data[4:]
    if record_size == 8:
        while len(attendance_data) >= 8:
            if(if_ma300==True) :
                uid, status, punch, timestamp = unpack('<HBB4s', attendance_data.ljust(8, b'\x00')[:8]) # Cas d'une mA300
            elif(if_ma300==False) :
                uid, status, timestamp, punch = unpack('HB4sB', attendance_data.ljust(8, b'\x00')[:8])
            if self.verbose: print (codecs.encode(attendance_data[:8], 'hex'))
            attendance_data = attendance_data[8:]
            tuser = list(filter(lambda x: x.uid == uid, users))
            if not tuser:
                user_id = str(uid)
            else:
                user_id = tuser[0].user_id
            timestamp = self.__decode_time(timestamp)
            attendance = Attendance(user_id, timestamp, status, punch, uid)
            attendances.append(attendance)
    elif record_size == 16:
        while len(attendance_data) >= 16:
            user_id, timestamp, status, punch, reserved, workcode = unpack('<I4sBB2sI', attendance_data.ljust(16, b'\x00')[:16])
            user_id = str(user_id)
            if self.verbose: print(codecs.encode(attendance_data[:16], 'hex'))
            attendance_data = attendance_data[16:]
            tuser = list(filter(lambda x: x.user_id == user_id, users))
            if not tuser:
                if self.verbose: print("no uid {}", user_id)
                uid = str(user_id)
                tuser = list(filter(lambda x: x.uid == user_id, users))
                if not tuser:
                    uid = str(user_id)
                else:
                    uid = tuser[0].uid
                    user_id = tuser[0].user_id
            else:
                uid = tuser[0].uid
            timestamp = self.__decode_time(timestamp)
            attendance = Attendance(user_id, timestamp, status, punch, uid)
            attendances.append(attendance)
    else:
        while len(attendance_data) >= 40:
            uid, user_id, status, timestamp, punch, space = unpack('<H24sB4sB8s', attendance_data.ljust(40, b'\x00')[:40])
            if self.verbose: print (codecs.encode(attendance_data[:40], 'hex'))
            user_id = (user_id.split(b'\x00')[0]).decode(errors='ignore')
            timestamp = self.__decode_time(timestamp)

            attendance = Attendance(user_id, timestamp, status, punch, uid)
            attendances.append(attendance)
            attendance_data = attendance_data[40:]
    return attendances

in my case record_size = 49

[kurenai-ryu]

ok, I think someone pushed a fix for 49 bytes packets...