Pyserial issues with proper loops and binary

[deanhystad]

The arguments passed are wrong.

In write_command(), port is a Serial object, device is the LIDAR device(s) to receive the command, and command is the command you want the lidar device to execute. device and command are defined in the document: LDROBOT_LD07_Development manual_v1.0_en.pdf".

In your example, your serial device is called "device". I suggest you rename because "device" should be the lidar device. You want to send a command to device = 1 (you only have one LIDAR device), and the command you want to send is 2, PACK_GET_DISTANCE. To write a command to request distance information from you 1 lidar device you should call.:

write_command(device, 1, 2)

Inside write_command(), it will compute a checksum = 3, pack a command of bytes = b'\xAA\xAA\xAA\xAA\x01\x02\x00\x00\x00\x00\x03' and wrote these bytes out the serial port.

It might be worthwhile defining some commands at the top of the file, like:

MY_LIDAR = 1
PACK_GET_DISTANCE = 0x2
PACK_STOP = 0XF
PACK_GET_CDE = 0X12
PACK_ACL = 0x10

This lets you write a command like this in your code:

 # Which is easier to understand?  This:
write_command(serial_port, MY_LIDAR, PACK_GET_DISTANCE)
or this:
write_command(device, 1, 2)

I would rewrite your code above to look (somewhat) like this:

import serial
import struct

# define some LIDAR stuff
MY_LIDAR = 1
PACK_GET_DISTANCE = 0x2
PACK_STOP = 0XF
PACK_GET_CDE = 0X12
PACK_ACL = 0x10

serial_port = serial.Serial('/dev/ttyUSB0', baudrate=921600, timeout=10.0, bytesize=8, parity='N', stopbits=1)


def write_command(port, device, command):
    checksum = (device + command) & 15
    msg_bytes = struct.pack("<LBBHHB", 0xAAAAAAAA, device, command, 0, 0, checksum)
    port.write(msg_bytes)

def read_data(port):
    # Read the header
    header = port.read(size=10)
    if len(header) < 10:
        raise serial.SerialTimeoutException("Timed out reading header")
  
    # Verify the header is mostly correct
    start_chars, device, command, offset, size = struct.unpack("<LBBHH", header)
    if start_chars != 0xAAAAAAAA:
        raise serial.SerialException("Start characters don't match")
    if size > 800:
        raise serial.SerialException("Data cannot exceed 800 bytes.")
    # You could also check if the command code is something you recognize.
  
    # Read the data.  Data is followed by the message checksum.
    data = port.read(size=size+1)
    if len(data) < size+1:
        raise serial.SerialTimeoutException("Timed out reading data")
  
    # Verify the checksum.  Checksum is the sum of the entire message EXCEPT the 4 byte start character. and checksum.
    checksum = (sum(header[4:]) + sum(data[:-1])) & 15
    if checksum != data[-1]:
        raise serial.SeriaException("Checksum is invalid")
  
    return device, command, data[:-1]



write_command(serial_port, MY_LIDAR, PACK_GET_DISTANCE)
device, command, data = read_data(serial_port) 
print (f'reply from {device}.{command} = {data}')

I also found a bug while looking at the code.

# data = port-read(size=size+1)
data = port.read(size=size+1)

I can't run any of this so there are likely other bugs. Heck, the byte order might not even be right for the pack/unpack calls.

[jttolleson]

Hello....I have been away from my side project doing realer stuff and have gotten back toward lidar again .... a couple of weeks later...
I read the code less tired and i understand much better....however i do not know struct soo....

So i dove into struct, which is to 'interpret bytes of packed binary data' and decided to update the struct code...

here is my new struct code which runs much the same as your code with a timeout error....i am working to resolve this, however right now i do not return a header.

lidartest.py

!/usr/bin/env python3
import serial
from serial import Serial
import struct
from struct import unpack
import time
#####################################################################################################################
command = [0xAA,0xAA,0xAA,0xAA,0x01,0x02,0x00,0x00,0x00,0x00,0x03]
MY_LIDAR = 1
PACK_GET_DISTANCE = 0x2
PACK_STOP = 0XF
PACK_GET_CDE = 0X12
PACK_ACL = 0x10
ld07_distances = []
ld07_confidences = []
#####################################################################################################################
serial_port = serial.Serial('/dev/ttyUSB0', baudrate=921600, timeout=10.0, bytesize=8, parity='N', stopbits=1)
print ('serial port name: '+str(serial_port.name))
serial_port.flush()
#####################################################################################################################
def write_command(port, device, command):
    checksum = (device + command) & 15
    msg_bytes = struct.pack("<LBBHHHHB", 0xAAAAAAAA, device, command, 0, 0, 0, 0, checksum)
    port.write(msg_bytes)
def read_data(port):
    # Read the header
    header = port.read(size=10)
    print (header)
    print (len(header))
    if len(header) < 10:
        raise serial.SerialTimeoutException("Timed out reading header")
    # Verify the header is mostly correct
    start_chars, device, command, offset, size = struct.unpack("<LBBHH", header)
    if start_chars != 0xAAAAAAAA:
        raise serial.SerialException("Start characters don't match")
    if size > 800:
        raise serial.SerialException("Data cannot exceed 800 bytes.")
    # You could also check if the command code is something you recognize.
    # Read the data.  Data is followed by the message checksum.
    data = port.read(size=size+1)
    if len(data) < size+1:
        raise serial.SerialTimeoutException("Timed out reading data")
    # Verify the checksum.  Checksum is the sum of the entire message EXCEPT the 4 byte start character. and checksum.
    checksum = (sum(header[4:]) + sum(data[:-1])) & 15
    if checksum != data[-1]:
        raise serial.SeriaException("Checksum is invalid")
    return device, command, data[:-1]
write_command(serial_port, MY_LIDAR, PACK_GET_DISTANCE)
device, command, data = read_data(serial_port) 
print (f'reply from {device}.{command} = {data}')

my next step is to include 'serial' module 'in_waiting()' to wait for 10 bytes......

[deanhystad]

def read_data(port):
    # Read the header
    header = port.read(size=10)
    if len(header) < 10:
        print("read buffer contains ", port.read(size=port.in_waiting)
        raise serial.SerialTimeoutException("Timed out reading header")
   
    # Verify the header is mostly correct
    start_chars, device, command, offset, size = struct.unpack("<LBBHH", header)
    if start_chars != 0xAAAAAAAA:
        raise serial.SerialException("Start characters don't match")
    if size > 800:
        raise serial.SerialException("Data cannot exceed 800 bytes.")
    # You could also check if the command code is something you recognize.
   
    # Read the data.  Data is followed by the message checksum.
    data = port.read(size=size+1)
    if len(data) < size+1:
        raise serial.SerialTimeoutException("Timed out reading data")
   
    # Verify the checksum.  Checksum is the sum of the entire message EXCEPT the 4 byte start character. and checksum.
    checksum = (sum(header[4:]) + sum(data[:-1])) & 15
    if checksum != data[-1]:
        raise serial.SeriaException("Checksum is invalid")
   
    return device, command, data[:-1]

[jttolleson]

Hi all, Jay T here again...I have some lidar chips and batteries on a table with a raspberry pi and will run the code via ssh again today. It has been some time................i caught some huge fish without the fishfinder too.
   

[jttolleson]

#!/usr/bin/env python3
import serial
from serial import Serial
import struct
from struct import unpack
import time
#####################################################################################################################
command = [0xAA,0xAA,0xAA,0xAA,0x01,0x02,0x00,0x00,0x00,0x00,0x03]
MY_LIDAR = 1
PACK_GET_DISTANCE = 0x2
PACK_STOP = 0XF
PACK_GET_CDE = 0X12
PACK_ACL = 0x10
ld07_distances = []
ld07_confidences = []
#####################################################################################################################
port = serial.Serial('/dev/ttyUSB0', baudrate=921600, timeout=10.0, bytesize=8, parity='N', stopbits=1)
print ('serial port name: '+str(port.name))
port.flush()
#####################################################################################################################
def write_command(port, device, command):
    checksum = (device + command) & 15
    msg_bytes = struct.pack("<LBBHHHHB", 0xAAAAAAAA, device, command, 0, 0, 0, 0, checksum)
    port.write(msg_bytes)



def read_data(port):
    # Read the header
    header = port.read(size=10)
    if len(header) < 10:
        print ("read buffer contains ", port.read(size=port.in_waiting))
        raise serial.SerialTimeoutException("Timed out reading header")
    
    # Verify the header is mostly correct
    start_chars, device, command, offset, size = struct.unpack("<LBBHH", header)
    if start_chars != 0xAAAAAAAA:
        raise serial.SerialException("Start characters don't match")
    if size > 800:
        raise serial.SerialException("Data cannot exceed 800 bytes.")
    # You could also check if the command code is something you recognize.
    
    # Read the data.  Data is followed by the message checksum.
    data = port.read(size=size+1)
    if len(data) < size+1:
        raise serial.SerialTimeoutException("Timed out reading data")
    
    # Verify the checksum.  Checksum is the sum of the entire message EXCEPT the 4 byte start character. and checksum.
    checksum = (sum(header[4:]) + sum(data[:-1])) & 15
    if checksum != data[-1]:
        raise serial.SeriaException("Checksum is invalid")
    
    return device, command, data[:-1]



device, command, data[:-1] = read_data(port) 
print (f'reply from {device}.{command} = {data[:-1]}')

And this is what i get:
   

I will work to resolve this and make a fishfinder with lidar.....right now it is a small lidar on a 4 pin usb chip with a blinking red light....3v power and tx rx...with ground plugged into my laptop for testing....    

thanks for the assistance so far ... i will update ... later....

[jttolleson]

OK, So I have read through our post thoroughly and have recompiled code i understand:

lidar8.py

import serial
import time
import binascii
import math
####################################################################################################################
interface_ld07 = serial.Serial('/dev/ttyUSB0', baudrate=921600, timeout=1.0, bytesize=8, parity='N', stopbits=1)

if interface_ld07.isOpen() :
    print ("open success")
    print ('serial port name: '+str(interface_ld07.name))
else :
    print ("open failed") 
####################################################################################################################
ld07_getDistance = b'0xAA/0xAA/0xAA/0xAA/0x01/0x02/0x00/0x00/0x00/0x00/0x03'
ld07_stopGetDistance = b'0xAA/0xAA/0xAA/0xAA/0x01/0x0F/0x00/0x00/0x00/0x00/0x10'
ld07_distances = []
ld07_confidences = []
####################################################################################################################
print ('StartMeasurement: ', ld07_getDistance)
interface_ld07.flush()
distance = interface_ld07.write(ld07_getDistance)
ld07_startCharacter = int.from_bytes(interface_ld07.read(4), byteorder='little')
print ('start character: '+str(ld07_startCharacter))
ld07_deviceAddress = int.from_bytes(interface_ld07.read(1), byteorder='little')
ld07_cmdCode = int.from_bytes(interface_ld07.read(1), byteorder='little')
ld07_packetOffsetAddress = int.from_bytes(interface_ld07.read(2), byteorder='little')
ld07_dataLength = int.from_bytes(interface_ld07.read(2), byteorder='little')
ld07_timestamp = int.from_bytes(interface_ld07.read(4), byteorder='little')
print ('ld07_deviceAddress,ld07_cmdCode,ld07_packetOffsetAddress,ld07_dataLength,ld07_timestamp: ',ld07_deviceAddress,ld07_cmdCode,ld07_packetOffsetAddress,ld07_dataLength,ld07_timestamp)    
for measurement in range(16):
    tmpMeasurement = int.from_bytes(interface_ld07.read(2), byteorder='little')
    tmpConfidence = (tmpMeasurement >> 9) <<1
    tmpDistance = tmpMeasurement & 0x1ff
    ld07_distances.append(tmpDistance)
    ld07_confidences.append(tmpConfidence)
    print (f"{str(measurement)}:\t{bin(tmpMeasurement)}\t d: {str(tmpDistance)} mm\t Conf: {str(tmpConfidence)}")
interface_ld07.flush()

and my result is all zeros on my lidar, i do not know ....however this is actually working in the right direction and now I am working again....


   

darn, all zeros again.

[deanhystad]

I get the same results with nothing connected to a serial port. Success opening a serial port is guaranteed if you enter a valid file descriptor. int.from_bytes(b'', byteorder='little') returns 0. I don't think your lidar is responding at all.