Import data from Twe-lite(usb) to raspberry pi

tocos-wireless.com/jp/tech/misc/python_twelite/twelite_read_write.py

# python3 thisScript.py /dev/ttyUSB?
where ttyUSB? is your assigned ToCoStick port, which can be addressed by dmsg or lsusb.

UART reader, writer for TWE-Lite Std App
[Reading]
   Read ‘:??81????’ data from serial port and interpret its content
[Writing]
   Read from standard input line by line and output to serial port
   q[Enter] --> exit the script
   :??????[Enter] --> output ‘:…’ command

from serial import *
from sys import stdout, stdin, stderr, exit
import threading

ser = None # serial port obj
t1 = None  # reading thread
bTerm = False # flag to term the script

def printPayload(l):
    # Output message ‘as is...'
    if len(l) < 3: return False # check data size
    
    print (" command = 0x%02x (other)" % l[1])
    print ("  src     = 0x%02x" % l[0])
    
    # display the payload in HEX strings
    print ("  payload =",)
    for c in l[2:]:
        print ("%02x" % c,)
    print ("(hex)")
    return True
        
def printPayload_0x81(l):
    # 0x81 message, interpret and display
    if len(l) != 23: return False # check data size
    
    ladr = l[5] << 24 | l[6] << 16 | l[7] << 8 | l[8]
    print ("  command   = 0x%02x (data arrival)" % l[1])
    print ("  src       = 0x%02x" % l[0])
    print ("  src long  = 0x%08x" % ladr)
    print ("  dst       = 0x%02x" % l[9])
    print ("  pktid     = 0x%02x" % l[2])
    print ("  prtcl ver = 0x%02x" % l[3])
    print ("  LQI       = %d / %.2f [dbm]" % (l[4], (7*l[4]-1970)/20.))
    ts = l[10] << 8 | l[11]
    print ("  time stmp = %.3f [s]" % (ts / 64.0))
    print ("  relay flg = %d" % l[12])
    vlt = l[13] << 8 | l[14]
    print ("  volt      = %04d [mV]" % vlt)
    
    # check DI1..4
    dibm = l[16]
    dibm_chg = l[17]
    di = {} # list of DI state (1 as Lo)
    di_chg = {} # change state (once a port gets Lo, then set 1)
    for i in range(1,5):
        di[i] = 0 if (dibm & 0x1) == 0 else 1
        di_chg[i] = 0 if (dibm_chg & 0x1) == 0 else 1
        dibm >>= 1
        dibm_chg >>= 1
    
    print ("  DI1=%d/%d  DI2=%d/%d  DI3=%d/%d  DI4=%d/%d" % (di[1], di_chg[1], di[2], di_chg[2], di[3], di_chg[3], di[4], di_chg[4]))
    
    # AD1..4 data
    ad = {}
    er = l[22]
    for i in range(1,5):
        av = l[i + 18 - 1]
        if av == 0xFF:
            # -1 means `AD port is unused (>2.0V)
            ad[i] = -1
        else:
            # merge lower bit for accuracy
            ad[i] = ((av * 4) + (er & 0x3)) * 4
        er >>= 2
    print ("  AD1=%04d AD2=%04d AD3=%04d AD4=%04d [mV]" % (ad[1], ad[2], ad[3], ad[4]))
    
    return True

def readThread():
    # thread handling serial port line by line.
    global ser, bTerm
    while True:
        if bTerm: return # quit the script
        line = ser.readline().rstrip() # read line by line and strip line bread chars (blocking read)
        
        bCommand = False
        bStr = False
        
        if len(line) > 0:
            c = line[0]
            if isinstance(c, str):
                if c == ':': bCommand = True
                bStr = True
            else:
                # in python3, c will be ‘bytes’ type
                if c == 58: bCommand = True
                
            print ("\n%s" % line)
        
        if not bCommand: continue
    
        try:
            lst = {}
            if bStr:
                # for Python2.7
                lst = map(ord, line[1:].decode('hex')) # decode HEX string and make byte list
            else:
                # for Python3
                import codecs
                s = line[1:].decode("ascii") # bytes -> str
                lst = codecs.decode(s, "hex_codec") # hex_codec でバイト列に変換 (bytes)
                
            csum = sum(lst) & 0xff # check sum (sum & 0xff should be zero)
            lst = lst[0:len(lst)-1] # remove check sum from the list (unable pop() in python3)
            if csum == 0:
                if lst[1] == 0x81:
                    printPayload_0x81(lst) # display if data is IO data (0x81) message
                else:
                    printPayload(lst) # display message other than 0x81.
            else:
                print ("checksum ng")
        except:
            if len(line) > 0:
                print ("...skip (%s)" % line) # error

def DoTerminate():
    # termination the script
    global t1, bTerm

    # set the flag to stop the thread
    bTerm = True
    
    print ("... quitting")
    time.sleep(0.5) # wait a while to thread termination
        
    exit(0)

if __name__=='__main__':
    # check the param
    #   The first ARG: name of serial port
    if len(sys.argv) != 2:
        print ("%s {serial port name}" % sys.argv[0])
        exit(1)
    
    # open the serial port
    try:
        ser = Serial(sys.argv[1], 115200, timeout=0.1)
        print ("open serial port: %s" % sys.argv[1])
    except:
        print ("cannot open serial port: %s" % sys.argv[1])
        exit(1)
        
    # start the serial port thread
    t1=threading.Thread(target=readThread)
    t1.setDaemon(True)
    t1.start()
    
    # input from stdin
    while True:
        try:
            # read a line from stdin
            l = stdin.readline().rstrip()
            
            if len(l) > 0:
                if l[0] == 'q': # ‘q’ to terminate
                    DoTerminate()
                    
                if l[0] == ':': # :command
                    cmd = l + "\r\n"
                    print ("--> "+ l)
                    ser.write(cmd)
        except KeyboardInterrupt: # Ctrl+C
            DoTerminate()
        except SystemExit:
            exit(0)
        except:
            # other exception
            print ("... unknown exception detected")
            break
    
    # just in a case
    exit(0)