benfasoli
2/29/2016 - 8:46 PM

Serial data logging in Python.

Serial data logging in Python.

#!/usr/bin/python
# Ben Fasoli
import datetime
import os
import serial as ser
import string
from time import sleep

def saveLine(tobj, line, name):
    fname = tobj.strftime(name)
    with (open(fname, 'a+')) as f:
        f.write(line)

class serialDevice:
    def __init__(self, device, baud=9600, name='~/%Y_%m_%d_data.dat'):
        self.loc     = loc
        self.name    = name
        self.s       = ser.Serial(device, baud)
    def read(self):
        line = self.s.readline()
        tcol  = datetime.datetime.utcnow()
        line  = filter(lambda x: x in set(string.printable), line).replace('\r', '').replace('\n', '')
        save = ','.join([str(tcol), line]) + '\n'
        saveLine(tcol, save, self.name)
        print(save)
    def end(self):
        print('\n\nStopping data collection...\n\n')
        self.s.flush()
        self.s.close()

if __name__ == '__main__':
    device = '/dev/ttyUSB0'
    baud   = 9600
    delim  = '\r\n'
    name   = '~/%Y_%m_%d_data.dat'
    dev = serialDevice(device, baud, delim, loc, name)
    while True:
        try:
            dev.read()
        except (KeyboardInterrupt, SystemExit):
            raise
        except:
            print('Unknown error. Trying again.')
    dev.end()
#!/usr/bin/python
# Ben Fasoli
import datetime
import os
import serial as ser
import string
from time import sleep

def saveLine(tobj, line, loc, name):
    fname = loc + tobj.strftime(name)
    with (open(fname, 'a+')) as f:
        f.write(line)

class serialDevice:
    def __init__(self, device, baud=9600, delim='\r\n', loc='~/', name='%Y_%m_%d_data.dat'):
        self.allowed = set(string.printable)
        self.buf     = str()
        self.delim   = delim
        self.loc     = loc
        self.name    = name
        self.s       = ser.Serial(device, baud)
    def read(self):
        while self.delim not in self.buf:
            sleep(0.05)
            self.buf = self.buf + self.s.read(self.s.inWaiting())
        tcol  = datetime.datetime.utcnow()
        parts = self.buf.partition(self.delim)
        line  = parts[0].replace('\r', '').replace('\n', '')
        line  = filter(lambda x: x in self.allowed, line)
        self.buf = parts[2]
        save = ','.join([str(tcol), line]) + '\n'
        saveLine(tcol, save, self.loc, self.name)
        print(save)
    def end(self):
        print('\n\nStopping data collection...\n\n')
        self.s.flush()
        self.s.close()

if __name__ == '__main__':
    device = '/dev/ttyUSB0'
    baud   = 9600
    delim  = '\r\n'
    loc    = './'
    name   = '%Y_%m_%d_data.dat'
    dev = serialDevice(device, baud, delim, loc, name)
    while True:
        try:
            dev.read()
        except (KeyboardInterrupt, SystemExit):
            raise
    dev.end()