#!/usr/bin/python # tz_cms50.py # --copyright-- Copyright 2011 (C) Tranzoa, Co. All rights reserved. Warranty: You're free and on your own here. This code is not necessarily up-to-date or of public quality. # --url-- http://www.tranzoa.net/tzpython/ # --email-- pycode is the name to send to. tranzoa.com is the place to send to. # --bodstamps-- # September 19, 2011 bar # September 20, 2011 bar figure out bottom 4 bits of 3rd byte (dupes of top 4 of 2nd byte, Y value) # 8O1 instead of 8N1 - like PC program # fix timeouts # fix > 127 heart rates # September 21, 2011 bar .png file output # put the sample getting inside the thing # spin off find_likely_COM_ports() to tz_usb.py # September 22, 2011 bar continue to fight auto-upload and upload in general # September 23, 2011 bar rememmber the a0, b0, c0 cmds that come in every 255th sample in upload # remember the hr==ox==0 samples # --verbose # September 24, 2011 bar time stamp the file name of the main wave sample output file # run the .png file # auto-restart the com port in case it was yanked # September 25, 2011 bar don't write .dat files and whack empty .plsoxi files after they've been written # September 27, 2011 bar allow no io to be given at create time # September 28, 2011 bar print average wave values or something else # September 29, 2011 bar upload needs slower timeout # October 2, 2011 bar put the USB/serial ids in constants at the top # October 4, 2011 bar write_new_samples when flushing a recording file # October 10, 2011 bar finger samples # rework sample parsing # October 11, 2011 bar able to restart writing to a new file # October 14, 2011 bar allow simple sample ox of 255 (as in recordings) # November 9, 2011 bar be robust in the face of no pygooglechart installed # and allow graph with the finger "samples" in there # November 9, 2011 bar try multiplying the low 3 bits of the 1st byte in to the waveform values # November 11, 2011 bar only use one bit from byte 3 as the high bit of the heart rate - ignore the byte 3 high bit # except that the byte 3 high bit will cause us now to ignore the sample entirely # November 13, 2011 bar try sending an E to the device (another device responds with version info) # November 29, 2011 bar pyflake cleanup # May 27, 2012 bar doxygen namespace # January 18, 2016 bar except syntax change # December 16, 2016 bar helpfull prompt # December 3, 2017 bar put the 'when' in samples # January 25, 2018 bar a doc string for a_finger_sample() # December 16, 2019 bar parse samples with extra stuff at the ends of the sample text lines - to allow star-graph waveforms to be stored in the sample files # parse a float in the tm record at the top of a recording file # December 17, 2019 bar sigh - would that I could code - and also, let's have yet another go at setting a recording time # write_graph option # December 19, 2019 bar tweak tm at file close time - this whole logic needs a rethink. a_recording() is too smart and assumes too much # January 2, 2020 bar print why the port didn't open # March 3, 2023 bar python3 # get rid of tz_google_chart stuff (it no longer works) # --eodstamps-- ## \file # \namespace tzpython.tz_cms50 # # # Communicate with a CMS50E Pulse Oximeter # # Notes: # # When recording, occasionally a USB stream has dropped samples - several. You can see the jump in the waveform. # When screen changes happen, ditto. # See the note: comments in upload code for information about what the device sends. # # TODO: # Jeez. Why didn't I put a full float time stamp at the top of the file and then just use elapsed times subsequently? Yeah. Reading the file couldn't be done by just randomly jumping in to the file and starting after the next \n, but ... that's easy to fix. # # figure out what the 6 low bits in the 1st byte are # 0x08 and 0x10 seem to be triggered by valsalva - PC program says "Searching" and flat-lines the waveform display # it's the low nibble values of 5, 6, and 7 that come through for no easy-to-understand reason # figure out the a0, b0, c0 cmds in upload data # figure out what the top 0x30 bits in the 3rd byte are (top 3 bits have been detected set, but the top bit is not necessarily what we make of it (hr + 256)) # figure out commands to device (use \winbin\portmon on PC - note: cannot run from a net drive) # Any byte sent to it seems to turn on USB streaming (f5 and f6 do it - some others did, too) # # Upload from portmon: # >> f5 f5 These seem to also be sent out in the middle of the dump from the device # << 86 09 01 # << 47 5f # << 86 08 01 # << 47 5f # . # . # . # >> f6 f6 f6 # # from __future__ import print_function import math import os import re import socket import sys import time import serial # from pySerial import output_files import tz_usb import tzlib USB_VENDOR_ID = 0x10c4 USB_PRODUCT_ID = 0xea60 SAMPLE_RATE = 60 class a_cms50_exception(Exception) : pass class a_cms50_timeout_exception(a_cms50_exception) : pass # there's nothing to be had for now class a_cms50_data_exception(a_cms50_exception) : pass # the data isn't ready yet class a_cms50_no_finger_exception(a_cms50_exception) : pass # a "finger" isn't in the device class a_simple_sample(object) : p_attrs = [ 'hr', 'ox', ] def __init__(me, hr = 0, ox = 0) : me.hr = hr me.ox = ox if (me.ox > 255) : # in practice, notice the 'when' value in a_finger_sample() is gonna always be over 255 (and a dot-float, to boot, but we don't even need to check that) - anyway, we'll not want the 'when' to be 0, though the code allows it raise ValueError("ox:%s" % str(me.ox)) # also, 'ox' value is sometimes 255 and there's a bogus 'hr' value going with it - possibly 127, for instance - also, it appears [0,0] samples are placeholders while the finger is out if me.hr is None : raise ValueError("hr:%s" % str(me.hr)) pass def print_str(me) : return("%2u%% %3ubpm" % ( me.ox, me.hr ) ) def __str__(me) : aa = [ getattr(me, atr, None) for atr in me.p_attrs ] return(repr(aa)) # a_simple_sample class a_finger_sample(a_simple_sample) : """ A "sample" that indicates when a finger goes on or off. """ p_attrs = [ 'on', 'when', ] def __init__(me, on = False, when = None) : me.when = when or time.time() me.on = on def print_str(me) : return("Finger %s %f %s" % ( ((me.on and "on") or "off"), me.when, time.strftime('"%H:%M:%S %b %d, %Y"', time.localtime(me.when)), ) ) # a_finger_sample class a_full_sample(a_simple_sample) : csv_header = '"Sample", "Beat", "Value", "Oxi%", "BPM", "?b1", "?b3hi"' p_attrs = a_simple_sample.p_attrs + [ 'num', 'y', 'beat', 'bc', 'ac', ] def __init__(me, hr = 0, ox = 0, num = 0, y = 0, beat = False, bc = 0, ac = 0) : super(a_full_sample, me).__init__(hr = hr, ox = ox) me.num = num me.y = y me.beat = beat me.bc = bc # low 6 bits of 1st byte me.ac = ac # 0x30 bits of 3rd byte # me.y = ((me.y) * ((me.bc & 0xf) + 1)) / 4.5 def csv_str(me) : return("%d, %s, %d, %d, %d, %d, %d" % ( me.num, ((me.beat and '"B"') or ''), me.y, me.ox, me.hr, me.bc, me.ac, ) ) def print_str(me) : return("%u: b=%02x ac=%02x %s y=%3d %s" % ( me.num, me.bc, me.ac, ((me.beat and 'B') or ' '), me.y, super(a_full_sample, me).print_str(), ) ) # a_full_sample def _parse_sample(s) : try : return(eval(s)) except ( TypeError, ValueError ) : pass return(None) def _create_sample(aa, cls) : if len(aa) == len(cls.p_attrs) : kwargs = dict(zip(cls.p_attrs, aa)) try : return(cls(**kwargs)) except ( TypeError, ValueError ) : pass pass return(None) def parse_sample(s) : aa = _parse_sample(s) ss = _create_sample(aa, a_full_sample) or _create_sample(aa, a_simple_sample) or _create_sample(aa, a_finger_sample) return(ss) class a_recording(object) : FILE_EXT = ".plsoxi" def __init__(me, tm = None, samples = [], fd = "", write_graph = False) : if tm is None : tm = time.time() me.tm = tm me.now = tm # the time of the next sample we'll get if me.now : scnt = 0.0 tm = sys.maxsize for s in samples : when = getattr(s, 'when', None) if when is None : s.when = me.now + (scnt / SAMPLE_RATE) else : me.now = s.when scnt = 0.0 if hasattr(s, 'y') : scnt += 1.0 tm = min(tm, s.when) # find the earliest sample time pass if tm < sys.maxsize : me.tm = tm # make our time the earliest real sample time me.now = me.now + (scnt / SAMPLE_RATE) # set the time it'll be when we get the next sample me.samples = samples or [] # a_???_sample()s me.fd = fd or "" # raw data me.fo = None # file we are outputting to, if any me.write_graph = write_graph # whether we put out the waveform stars in the file def forget_old_samples(me, how_many) : how_many = max(0, int(how_many)) me.samples = me.samples[how_many:] # let's not run out of memory if this thing runs for days if me.fo : me.wfsi = max(0, me.wfsi - how_many) pass def append(me, sample) : sample.when = getattr(sample, 'when', me.now) me.now = sample.when # get back in sync so we don't drift if not hasattr(sample, 'when') : # note: we don't do this logic now - why? Because we're just stepping me.now and we've set it at the start sample.when = time.time() # rather than using 'now', for now, just assume we're appending real-time samples. yuck. if hasattr(sample, 'y') : me.tm = min(me.tm, sample.when) # set our .tm to the earliest real sample time we've seen me.now += 1.0 / SAMPLE_RATE # set the time it'll be when we get the next sample me.samples.append(sample) def tm_str(me) : if me.tm < 3600 * 24 * 365 : tm = me.tm % ( 3600 * 24 ) return("%02u:%02u" % ( tm // 3600, (tm // 60) % 60 ) ) # hour:minute return(time.strftime('"%H:%M:%S %b %d, %Y"', time.localtime(me.tm))) def flush_file(me) : me.write_new_samples() if me.fo : me.fo.flush() pass def write_new_samples(me) : if not me.fo : return(False) nsc = len(me.samples) cnt = nsc - me.wfsi for sa in me.samples[me.wfsi:nsc] : if me.wfna != sa.p_attrs : me.wfna = sa.p_attrs me.fo.write("fields %s\n" % str(me.wfna)) if me.write_graph and hasattr(sa, 'y') : ys = (' ' * int((100.0 * (sa.y - 0)) / max(1.0, (128 - 0)))) + '*' me.fo.write("s %-35s | %s\n" % ( str(sa), ys, )) else : me.fo.write("s %s\n" % str(sa)) pass me.wfsc += cnt me.wfsi = nsc return(cnt) def close_write_file(me) : if me.fo : me.write_new_samples() me.fo.close() me.fo = None if not me.wfsc : os.remove(me.fn) # whack the file if there were no samples written if len(me.samples) : me.tm = max(me.tm, me.samples[-1].when) else : me.tm = max(me.tm, time.time()) return(True) return(False) def open_write_file(me, fn, only_new_samples = False) : me.close_write_file() me.wfsi = (only_new_samples and len(me.samples)) or 0 me.wfsc = 0 me.wfna = [] me.fn = os.path.splitext(fn)[0] + me.FILE_EXT # force the proper ext (?) me.fo = output_files.a_file(me.fn) me.fo.write("tm %f %s\n\n" % ( me.tm, me.tm_str() ) ) return(me.fn) def write_file(me, fn) : fn = me.open_write_file(fn) if me.close_write_file() : return(fn) return(None) s_vals_parse_re = re.compile(r'^s\s+(\[.+\])', re.DOTALL) @staticmethod def parse(fd, write_graph = None) : samples = [] for li in re.split(r"\r?\n", fd) : g = a_recording.s_vals_parse_re.search(li) if g : s = parse_sample(g.group(1)) if s : samples.append(s) pass pass if len(samples) : g = re.search(r"\ntm (" + tzlib.float_regx_str + r")", fd) if g : tm = float(g.group(1)) return(a_recording(tm, samples, write_graph = write_graph)) pass return(None) @staticmethod def parse_file(fn, write_graph = None) : fd = tzlib.read_whole_text_file(fn) return(a_recording.parse(fd, write_graph = write_graph)) # a_recording class a_comm(object) : def __init__(me, io = None, timeout = 0.01) : me.io = io me.timeout = timeout me.ibuf = "" me.scnt = 1 me.rx_when = tzlib.elapsed_time() - 10000.0 me.data = [] # queue of a_recording's driven by data coming in from the device if not me.io : me.reopen() pass def tx(me, s) : s = tzlib.convert_to_latin1_bytes(s) return(me.io.write(s)) def rx(me, how_many = 1) : try : return(me.io.read(how_many).decode('latin1')) except ( serial.SerialException, serial.serialutil.SerialException, OSError, IOError, AttributeError ) : pass return('') def read(me, how_many = 1, timeout = None) : timeout = timeout or me.timeout t = nt = tzlib.elapsed_time() r = "" while True : rr = me.ibuf[:how_many] # pull in anything we have buffered - what we've read from the device, but haven't returned from this routine me.ibuf = me.ibuf[how_many:] how_many -= len(rr) rc = len(rr) r += rr if not how_many : break rr = me.rx(how_many) how_many -= len(rr) rc += len(rr) r += rr # and pull in as much as asked for from the device, itself if not how_many : # until we've satisfied the caller's request break if rc : t = nt if nt - t >= timeout : if len(r) : break raise a_cms50_timeout_exception("read") nt = tzlib.elapsed_time() if not rc : time.sleep(min(0.1, timeout / 2.0)) pass # print("rxing", len(r), "%04x" % ( len(r) ), hexify(r)) return(r) def read_upload(me, progress_rtn = None, verbose = 0) : timeout = 10 while True : try : r = me.read(1, timeout = timeout) timeout = 0.1 a1 = me.read(1, timeout = timeout) a2 = me.read(1, timeout = timeout) cmd = ord(r) if (cmd == 0xf2) and (ord(a1) & 0x80) : me.ibuf = r + a1 + a2 + me.ibuf break elif cmd == 0xf0 : raise a_cms50_data_exception("No upload header") else : me.ibuf = a1 + a2 + me.ibuf except ( a_cms50_exception, a_cms50_timeout_exception, ) : return(None) pass fd = "" tm = 0 prv = 0 mxc = 10000000 samples = [] while True : try : r = me.read(1, timeout = timeout) + me.read(1, timeout = timeout) + me.read(1, timeout = timeout) cmd = ord(r[0]) hr = ord(r[1]) ox = ord(r[2]) if not (cmd & 0x80) : # print("@@@@ bad cmd %02x:%02x:%02x" % ( cmd, ord(r[1]), ord(r[2]) )) break sa = None if (not hr) and (not ox) : # a = hr | (ox << 8) # print("@@@@ @%u toss out msg: 2nd/3rd no hi bit cmd %02x arg=%02x:%02x d=%d x=%04x" % ( len(fd) // 3, cmd, hr, ox, a, a )) sa = a_full_sample(hr, ox, bc = cmd) r = "" # this may be a bug in the device. not sure. not sure whether tis correct to toss the data out of mxc tracking. Too, a ox of 0xff came up in a recording that had 6 extra samples even after this, which happened at 255-bytes in. elif cmd == 0xf2 : tm = 60 * ((60 * (hr & 0x1f)) + ox) prv = 0xf2 else : if prv == 0xf2 : mxc = ((cmd & 0x3f) << 14) | ((hr & 0x7f) << 7) | ox # print("@@@@ @%u mxc=%u %02x:%02x:%02x" % ( len(fd), mxc, cmd, hr, ox, )) mxc = mxc + len(fd) - 9 elif ((cmd & 0xf0) != 0xf0) : # a = hr | (ox << 8) # print("@@@@ info@%u: cmd %02x arg=%02x:%02x le:%d:0x%04x" % ( len(fd), cmd, hr, ox, a, a )) if (cmd == 0x80) and (len(fd) + 3 >= mxc) : me.ibuf = r + me.ibuf r = "" # !!!! kludge to remove ending, regular, 5-byte samples mxc -= 3 elif (cmd == 0x80) and (hr == 0) : sa = a_full_sample(hr, ox, bc = cmd) mxc -= 3 # !!!! is this sample a reflection of a glitch in the recording? and, if so, why doesn't the byte count reflect it? (or does it mark something? or what?) else : sa = a_full_sample(hr, ox, bc = cmd) # note: every 256th sample (after the 1st, 253rd - 256, including the two F2 and the 8x length "samples") is one of these - cmds are in [ 0xa0, 0xb0, 0xc0, ], hr and oxi are normal pass else : hr = ((cmd & 0x3) << 7) | (hr & 0x7f) sa = a_simple_sample(hr, ox) # note: in each 256 sample batch, the 85th sample and the 170th sample are ox==255 prv = cmd if sa : samples.append(sa) fd += r if progress_rtn : progress_rtn(len(fd), mxc) if len(fd) >= mxc : # print("@@@@ ctmxc", len(fd), mxc) break pass except ( a_cms50_exception, a_cms50_timeout_exception, ) : # print("@@@@ timeout") break pass if progress_rtn : progress_rtn(len(fd), len(fd)) # print("@@@@", len(fd), "of", mxc, " ", len(fd) // 3, "msgs of", (mxc + 2) // 3) return(a_recording(tm = tm, samples = samples, fd = fd)) def read_sample(me, progress_rtn = None, verbose = 0, mismatch_callback = None) : s = None try : b = ord(me.read()) me.rx_when = tzlib.elapsed_time() if b == 128 : ba = [ b ] t = me.rx_when while (len(ba) < 5) and (t - me.rx_when < 0.1) : t = tzlib.elapsed_time() ba.append(ord(me.read())) me.scnt += 1 if False and (ba[1] in [ 0, 0xff ]) : # 321390_Kanograf_(8400).pdf if len(ba) == 5 : print("@@@@ %02x:%02x:%02x:%02x:%02x" % ( ba[0], ba[1], ba[2], ba[3], ba[4] )) else : print("@@@@", str(ba)) pass raise a_cms50_no_finger_exception if b < 128 : # print("@@@@ toss read %02x" % b) pass else : y = ord(me.read()) ay = ord(me.read()) if (b == 0xf2) and (y & 0x80) : me.ibuf = chr(b) + chr(y) + chr(ay) + me.ibuf dt = me.read_upload(progress_rtn = progress_rtn, verbose = verbose) if dt : me.data.append(dt) return(None) hr = ord(me.read()) if (ay == 0xf2) and (hr & 0x80) : me.ibuf = chr(ay) + chr(hr) + me.ibuf dt = me.read_upload(progress_rtn = progress_rtn, verbose = verbose) if dt : me.data.append(dt) return(None) hr |= ((ay & 0x40) << 1) ox = ord(me.read()) if ay & 0x80 : # me.ibuf = chr(ay) + chr(hr) + chr(ox) + me.ibuf # note: the few times this has happened have not indicated that this is a good idea - nor that interpreting the ay|hr|ox bytes as uploaded data is a good idea return(None) # punt as best we can bc = (b & ~(64 | 128)) ac = ay >> 4 # print("@@@@ %02x" % b, hr, ox, me.scnt, y, ay, bc, ac) s = a_full_sample(hr, ox, me.scnt, y, ((b & 64) and True) or False, bc = bc, ac = ac & 3) if (ay & 0x0f) != (y // 8) : # apparently, these 4 bits are a dupe of the top 4 or the 7 bits in "y" - let's force that to be so - crashes when device dumps memory # print("@@@@ b=%02x ay & 0xf != y/8 ay:0x%02x != y:0x%02x" % ( b, ay, y ) # can happen if upload is starting) if mismatch_callback : mismatch_callback(s, b, ay) # tell the caller this happened if he wants to know s = None me.scnt += 1 pass except ( a_cms50_exception, a_cms50_timeout_exception, ) : if tzlib.elapsed_time() - me.rx_when > 0.5 : me.scnt = 1 raise return(s) def close(me) : if me.io : c = me.io me.io = None try : c.close() except socket.error : raise a_cms50_exception("Close error!") pass pass def blind_close(me) : try : me.close() except a_cms50_exception : pass pass def reopen(me, port = None) : me.blind_close() if not port : port = tz_usb.find_likely_COM_ports(vendor_id = USB_VENDOR_ID, product_id = USB_PRODUCT_ID) if port : port = port[0] pass if not port : return(None) try : port = int(port) cport = port - 1 except ValueError : cport = port try : me.io = serial.Serial(port = cport, baudrate = 19200, parity = "O", timeout = 0.001) # note: PC program sets 8O1. serial.Serial() multiplies timeout by 1000 before passing to windows (This 1 mill is minimum for windows. I don't know about other OS's.) except serial.SerialException : return(False) return(True) def start_usb(me) : try : me.tx("\xf5") # try to get the USB streaming going again in case it's off except ( OSError, IOError, AttributeError, serial.SerialException ) : me.reopen() pass pass # a_comm def _program_version_str(ident) : return("V" + re.sub(r".*([\d\.]+).*", r"\1", ident)) def get_output_file_name(ofile_name, program_name = None, ext = a_recording.FILE_EXT) : if not ofile_name : ofile_name = os.path.basename(program_name or __file__) ofile_name = os.path.splitext(ofile_name)[0] + ("_%010u%s" % ( int(time.time()), ext )) return(ofile_name) def hms(t) : it = int(t) return("%2u:%02u:%04.1f" % ( it // 3600, (it // 60) % 60, t - ((it // 60) * 60) ) ) class a_progress_rtn(object) : def __init__(me) : me.pc = -1 me.sh = False def show_progress(me, how_many, to_do) : pc = (100 * how_many) // max(1, to_do) if me.pc != pc : me.pc = pc if how_many != to_do : me.sh = True sys.stdout.write("%3u%%\r" % me.pc) sys.stdout.flush() elif me.sh : sys.stdout.write('%3u%%\n' % me.pc) sys.stdout.flush() pass pass # a_progress_rtn def main(ident) : import TZCommandLineAtFile import TZKeyReady import tz_timer if ident : sys.argv.insert(1, ident) sys.argv.insert(1, '--ident') program_name = sys.argv.pop(0) TZCommandLineAtFile.expand_at_sign_command_line_files(sys.argv) ident = "" port = 0 # my COM port, not yours port_list = 0 verbose = 0 how_long = float(sys.maxsize) help_str = """ %s (options) (output_files_base_name) I get the streaming data from a CMS50E Pulse Oximeter. Options: --port port_number Set the COM port number --port_list List possible ports (twice, list all available ports) --version Print the program version number. --how_long seconds Only run for this long. (default: forever) """ % ( os.path.basename(program_name) ) oi = tzlib.array_find(sys.argv, [ "--help", "-?", "?", "-h", "/h", "/?", "?" ] ) if oi >= 0 : print(help_str) sys.exit(254) while True : oi = tzlib.array_find(sys.argv, [ "--ident", "-i" ] ) if oi < 0 : break del sys.argv[oi] if (oi >= len(sys.argv)) or not len(sys.argv[oi]) : print("Program IDENT info not given!") sys.exit(101) ident = sys.argv.pop(oi) while True : oi = tzlib.array_find(sys.argv, [ "--version", "-v" ] ) if oi < 0 : break del sys.argv[oi] print("version", _program_version_str(ident)) if not sys.argv : sys.exit(0) pass while True : oi = tzlib.array_find(sys.argv, [ "--port", "-p" ] ) if oi < 0 : break del sys.argv[oi] if (oi >= len(sys.argv)) or not len(sys.argv[oi]) : print("No COM port given!") sys.exit(102) port = sys.argv.pop(oi) while True : oi = tzlib.array_find(sys.argv, [ "--port_list" ] ) if oi < 0 : break del sys.argv[oi] port_list += 1 while True : oi = tzlib.array_find(sys.argv, [ "--how_long", "--how-long", "--howlong", ] ) if oi < 0 : break del sys.argv[oi] how_long = float(sys.argv.pop(oi)) while True : oi = tzlib.array_find(sys.argv, [ "--verbose" ] ) if oi < 0 : break del sys.argv[oi] verbose += 1 ofile_name = None if len(sys.argv) >= 1 : ofile_name = sys.argv.pop(0) if ofile_name.startswith('-') : print("Put the whole path or a dot/slash before the output file name. Dashes are confusing: [%s]" % ofile_name) sys.exit(104) pass if len(sys.argv) : print("I don't understand %s (--help for options)!" % ( sys.argv )) sys.exit(104) if port_list : tz_usb.find_likely_COM_ports(vendor_id = USB_VENDOR_ID, product_id = USB_PRODUCT_ID, list_level = port_list) if not port : port = tz_usb.find_likely_COM_ports(vendor_id = USB_VENDOR_ID, product_id = USB_PRODUCT_ID) if port : print( "Using port", port[0],) if len(port) > 1 : print("Found ports", port, end = ' ') print('') port = port[0] pass if not port : print("Please tell me a COM port to use with the --port option (e.g. --port 2 )!") sys.exit(103) try : port = int(port) cport = port - 1 except ValueError : cport = port try : io = serial.Serial(port = cport, baudrate = 19200, parity = "O", timeout = 0.001) # note: PC program sets 8O1. serial.Serial() multiplies timeout by 1000 before passing to windows (This 1 mill is minimum for windows. I don't know about other OS's.) except serial.SerialException : tzlib.print_exception() print("Port %s [%s] cannot be opened!" % ( str(port), str(cport) )) print("Best guess: Make this user able to open the serial port with: 'sudo usermod -a -G dialout ${USER}'") sys.exit(111) me = a_comm(io) samples = a_recording() if ofile_name : if os.path.splitext(ofile_name)[1].lower() == ".csv" : fo = output_files.a_file(ofile_name) fo.write(a_full_sample.csv_header + "\n") else : fo = None fn = get_output_file_name(ofile_name, program_name = program_name) print("Outputting to: ", samples.open_write_file(fn)) pass prg = a_progress_rtn() stopped = 1000 finger = False lay = [] mx = -10000 mn = 10000 msa = [ 64 ] * 300 mss = float(sum(msa)) ts = tzlib.elapsed_time() rx_when = ts png_drs = [ 7.5, 15.0, 30.0, 60.0, 120.0, 10 * 60.0, 15 * 60.0, 60 * 60.0, 2 * 60 * 60.0, 6 * 60 * 60.0, 12 * 60 * 60.0, ] png_di = 0 sb = 0xf5 me.start_usb() # in case he's not turned it on (though we'll do this every half second of silence from the device, anyway print("Type ? for help") start_time = tz_timer.tick() while tz_timer.tick() - start_time < how_long : try : s = me.read_sample(progress_rtn = prg.show_progress, verbose = verbose) if s : yy = s.y if False : msa.append(s.y) mss += s.y mss -= msa[0] del(msa[0]) else : mss = 64 if False : yy = ((s.y) * ((s.bc & 0xf) + 1)) / 4.5 # tends to flatten out in the middle when there is a change to low amplitude waves if False : if yy < 0 : yy = -math.log(-yy + math.e) else : yy = math.log( yy + math.e) pass pass mx = max(mx, yy) mn = min(mn, yy) if s.ac & 3 : # print("@@@@ ac=%u" % s.ac) lay.append(s.ac) if not finger : samples.append(a_finger_sample(True)) samples.append(s) if ofile_name : if fo : fo.write("%s\n" % s.csv_str()) samples.write_new_samples() ys = (' ' * int((100.0 * (yy - mn)) / max(1.0, (mx - mn)))) + '*' # avya = [ ox for ox in samples.samples[-5 * SAMPLE_RATE : ] if hasattr(ox, 'y') ] # avy = sum([ ox.y for ox in avya ]) / float(max(1, len(avya))) print("%s %s" % ( s.print_str(), ys )) # (100.0 * yy) / (yy + (s.bc & 0xf)), ys ) if not finger : finger = True samples.flush_file() print("Finger") stopped = max(stopped - 10, 0) if len(samples.samples) > png_drs[-1] * SAMPLE_RATE * 2 : samples.flush_file() samples.forget_old_samples(png_drs[-1] * SAMPLE_RATE) rx_when = tzlib.elapsed_time() t = tzlib.elapsed_time() if t - ts > 59 : ts = t samples.flush_file() pass except a_cms50_no_finger_exception : if finger : finger = False samples.append(a_finger_sample(False)) print("No finger") samples.flush_file() pass except a_cms50_data_exception as msg : samples.flush_file() print(msg) # those bits are not, apparently, dupes of each other if not stopped : sys.exit(199) stopped -= 1 except ( a_cms50_exception, a_cms50_timeout_exception, ) : t = tzlib.elapsed_time() if t - rx_when > 0.5 : rx_when = t ts = t mx = -10000 mn = 10000 samples.flush_file() me.start_usb() pass if len(me.data) : samples.flush_file() while len(me.data) : dt = me.data.pop(0) # print("@@@@", len(dt.samples), len(dt.fd), "at", dt.tm // 3600, (dt.tm // 60) % 60) if len(dt.samples) : fn = get_output_file_name(ofile_name, program_name = program_name, ext = ".dat") # tzlib.write_whole_binary_file(fn, dt.fd) dfn = dt.write_file(fn) if not dfn : print("Probably no data to write, so file not written.") else : print("Wrote driven upload to", fn, "and", dfn, len(dt.samples)) pass pass mx = -10000 mn = 10000 ts = tzlib.elapsed_time() stopped = 1000 k = TZKeyReady.key_ready() if k : print('') if k in [ 'q', ] : break if k == '?' : pcmd = (""" p Write %s file with latest data. """ % ( a_recording.FILE_EXT, )).rstrip() print(""" q Quit. ESC Quit. ? Help. r Reset some values. + Up the .png duration - currently %s - Lower the .png duration.%s d Wait for data upload to write to %s file. """ % ( hms(png_drs[png_di]), pcmd, a_recording.FILE_EXT ) ) print("ymx=%d ymn=%d lay=%s %.1f samples per second" % ( mx, mn, str(lay), me.scnt / max(1, (tzlib.elapsed_time() - ts)) )) # 60 per second if k == 'r' : samples.flush_file() mx = -10000 mn = 10000 ts = tzlib.elapsed_time() lay = [] me.scnt = 1 if k == 'd' : samples.flush_file() stopped = 1000 fn = get_output_file_name(ofile_name, program_name = program_name, ext = ".dat") print("Uploading to", fn, "You may need to turn on 'upload' in the menus.") dt = me.read_upload(progress_rtn = prg.show_progress, verbose = verbose) # print("@@@@", len(fd), "at", tm // 3600, (tm // 60) % 60) if dt and len(dt.samples) : # tzlib.write_whole_binary_file(fn, dt.fd) dfn = dt.write_file(fn) if not dfn : print("Probably no data to write, so file not written.") else : print("Wrote upload to", fn, "and", dfn, len(dt.samples)) pass else : print("Failed to upload.") pass if k == 'E' : print("Sending E") me.tx("E") if k == 'F' : print("Sending F") me.tx("F") if k == 'G' : print("Sending G") me.tx("G") if k == 'H' : print("Sending H") me.tx("H") if k == 's' : if False : me.tx(chr(sb)) print("s %02x:%u" % ( sb, sb )) sb -= 1 if sb < 0 : sb = 255 pass else : me.tx("\xf5\xf5") pass if k == 'S' : if False : me.tx(chr(sb)) print("s %02x:%u" % ( sb, sb )) sb += 1 if sb > 255 : sb = 0 pass else : me.tx("\xf6\xf6\xf6") pass if k in [ '+', '=' ] : png_di = min(len(png_drs) - 1, png_di + 1) print(".png duration %s" % hms(png_drs[png_di])) if k in [ '-', '_' ] : png_di = max(0, png_di - 1) print(".png duration %s" % hms(png_drs[png_di])) if k == 'p' : if (len(samples.samples) > 60) and (mn < mx) : fn = get_output_file_name(ofile_name, program_name = program_name, ext = ".png") sma = [ sa for sa in samples.samples if hasattr(sa, 'ox') ] tw = min(int(png_drs[png_di] * SAMPLE_RATE), len(sma)) sma = sma[len(sma) - tw : ] if not len(sma) : "No data to write/graph" else : dt = a_recording(samples = sma) dfn = dt.write_file(fn) if not dfn : print("Probably no data to write, so file not written.") else : print("Wrote %s" % dfn) pass pass pass pass pass if ofile_name : if fo : fo.close() samples.close_write_file() me.blind_close() if __name__ == '__main__' : main("") # # # eof