#!/usr/bin/python # tz_pulse_server.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-- # October 4, 2011 bar # October 8, 2011 bar put the log in the current dir # October 10, 2011 bar finger samples # print latest clients # spoofer # get rid of the .py in the log file name # October 11, 2011 bar we only need so many spoof samples. Don't overdo it. # restart writing to a new file every hour # November 7, 2011 bar close log in main # November 10, 2011 bar log 4-bit Y mismatches # --flush # November 11, 2011 bar pass verbose to the server # November 13, 2011 bar show_samples # November 14, 2011 bar typo # November 28, 2011 bar client .when is not elapsed_time() # November 29, 2011 bar pyflake cleanup # January 10, 2012 bar don't fuss about an unopenable port if it's given in argv # March 13, 2012 bar dance around a serial numbered cardiox cp2102 serial to USB converter # --eodstamps-- ## \file # # # This is a web server for streaming real time pulse data for graphing. # # import os import random import sys import tzlib import tz_http_server import tz_stream_graph_server LOG_FILE_NAME = os.path.splitext(os.path.basename(__file__))[0] + ".log" def set_main_page(me, title = None) : title = title or "Pulse" protocol = "http://" url = '' # "%s%s:%d" % ( protocol, tz_http_server.get_local_ip_address(), me.http_port ) me.main_file = tz_stream_graph_server.MAIN_PAGE % ( url, title, title, url, 'pulse' ) class a_spoofer(object) : def __init__(me, file_names = []) : me.file_names = {} me.add_files(file_names) me.samples = [] me.si = 0 me.start_now() def add_files(me, file_names = []) : me.file_names.update(tzlib.make_dictionary(file_names or [])) def start_now(me) : me.when = tzlib.elapsed_time() def get_sample(me) : t = tzlib.elapsed_time() if t - me.when >= 1.0 / tz_cms50.SAMPLE_RATE : me.when += 1.0 / tz_cms50.SAMPLE_RATE if me.si >= len(me.samples) : if len(me.file_names) and (len(me.samples) < tz_cms50.SAMPLE_RATE * 2000) : for i in xrange(len(me.file_names.keys())) : fa = [ ( os.path.getsize(fn), fn ) for fn in me.file_names.keys() ] fa.sort() fn = random.choice(fa[:max(1, (len(fa) / 4) + i)])[1] # print "@@@@", fn r = me.file_names[fn] if r is True : r = tz_cms50.a_recording.parse_file(fn) # note: in case we don't whack the file and append its samples me.file_names[fn] = r if r : sa = [ s for s in r.samples if hasattr(s, 'y') ] if len(sa) > tz_cms50.SAMPLE_RATE * 100 : del(me.file_names[fn]) # to not run up the memory use, forget this file and append its samples me.samples += sa me.samples = me.samples[:tz_cms50.SAMPLE_RATE * 4000] print "Spoofing", fn, len(sa), len(me.samples) break pass del(me.file_names[fn]) if not len(me.file_names) : break pass pass me.si = random.randint(0, len(me.samples) / 2) if me.si < len(me.samples) : s = me.samples[me.si] me.si += 1 return(s) pass return(None) # a_spoofer help_str = """ python %s (options) (output_file_name) Options: --title Set web page title. --http_port port_number Set the server port number. --port port_number Set the COM port. --port_list List possible ports (twice, list all available ports) --spoof file_name Add files matching the ambiguous name to the list of spoof files. Spoof files are used for data when the 'finger' is out. --flush Flush the output file every sample. --verbose Raise the verbosity level. Web server for pulse information streaming to a browser. """ if __name__ == '__main__' : import time import serial import TZKeyReady import tz_server_logger import TZCommandLineAtFile import tz_cms50 import tz_usb program_name = sys.argv.pop(0) TZCommandLineAtFile.expand_at_sign_command_line_files(sys.argv) http_port = tz_stream_graph_server.HTTP_PORT port = 0 # my COM port, not yours port_list = 0 verbose = 0 do_flush = 0 title = None spoofer = a_spoofer() if tzlib.array_find(sys.argv, [ "--help", "-h", "-?", "/?", "-?" ] ) >= 0 : print help_str % ( os.path.basename(program_name), ) sys.exit(254) while True : oi = tzlib.array_find(sys.argv, [ "--http_port", "--httpport", "--http-port", ]) if oi < 0 : break del sys.argv[oi] http_port = int(sys.argv.pop(oi)) 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", "--pl", ] ) if oi < 0 : break del sys.argv[oi] port_list += 1 while True : oi = tzlib.array_find(sys.argv, [ "--verbose", "-v", ] ) if oi < 0 : break del sys.argv[oi] verbose += 1 while True : oi = tzlib.array_find(sys.argv, [ "--flush", "-f", ] ) if oi < 0 : break del sys.argv[oi] do_flush += 1 while True : oi = tzlib.array_find(sys.argv, [ "--title", "-t", ] ) if oi < 0 : break del sys.argv[oi] title = sys.argv.pop(oi) while True : oi = tzlib.array_find(sys.argv, [ "--spoof", ] ) if oi < 0 : break del sys.argv[oi] afn = sys.argv.pop(oi) fns = tzlib.ambiguous_file_list(afn) if not fns : afn += tz_cms50.a_recording.FILE_EXT fns = tzlib.ambiguous_file_list(afn) if not fns : print "No spoof files matching: [%s]!" % afn sys.exit(105) spoofer.add_files(fns) 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 = tz_cms50.USB_VENDOR_ID, product_id = tz_cms50.USB_PRODUCT_ID, list_level = port_list) gport = port if not port : nports = tz_usb.find_likely_COM_ports(vendor_id = tz_cms50.USB_VENDOR_ID, product_id = tz_cms50.USB_PRODUCT_ID, serial_number = "ca8d") # kludge to avoid cardiox driver board in one incarnation ports = [ p for p in tz_usb.find_likely_COM_ports(vendor_id = tz_cms50.USB_VENDOR_ID, product_id = tz_cms50.USB_PRODUCT_ID) if p not in nports ] if len(ports) : port = ports[0] print "Using port", port, if len(ports) > 1 : print "Found ports", ports, print 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 io = None 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 : if not gport : print "Port %s [%s] cannot be opened!" % ( str(port), str(cport) ) sys.exit(111) pass cm = tz_cms50.a_comm(io) logger = tz_server_logger.a_logger( http_port = http_port, file_name = LOG_FILE_NAME) me = tz_stream_graph_server.a_server(http_port = http_port, logger = logger, show_exceptions = True, verbose = verbose) set_main_page(me, title) ipadr = tz_http_server.get_local_ip_address() print "Connect to http://localhost:%u/ or http://%s:%u/" % ( me.http_port, ipadr, me.http_port ) me.start() samples = tz_cms50.a_recording() if ofile_name : fn = tz_cms50.get_output_file_name(ofile_name, program_name = program_name) print "Outputting to: ", samples.open_write_file(fn) prg = tz_cms50.a_progress_rtn() stopped = 1000 finger = False ts = tzlib.elapsed_time() rx_when = ts cm.start_usb() # in case he's not turned it on (though we'll do this every half second of silence from the device, anyway st = tz_stream_graph_server.a_stream('pulse') me.add_stream(st) show_samples = False while True : no_finger = False s = None try : def mismatch(s, b, ay) : logger.log("; Mismatch b=%d ay=%d sample=%s" % ( b, ay, str(s) ) ) s = cm.read_sample(progress_rtn = prg.show_progress, verbose = verbose, mismatch_callback = mismatch) if s : st.append(s.y) if not finger : samples.append(tz_cms50.a_finger_sample(True)) samples.append(s) if do_flush : samples.flush_file() ys = (' ' * int((100.0 * (s.y - 0)) / max(1.0, (128 - 0)))) + '*' avya = [ ox for ox in samples.samples[-5 * tz_cms50.SAMPLE_RATE : ] if hasattr(ox, 'y') ] avy = sum([ ox.y for ox in avya ]) / float(max(1, len(avya))) if not finger : finger = True samples.flush_file() logger.flush() print "Finger" stopped = max(stopped - 10, 0) if show_samples : print "%s %s" % ( s.print_str(), ys ) # (100.0 * s.y) / (s.y + (s.bc & 0xf)), ys ) if len(samples.samples) > 3600 * tz_cms50.SAMPLE_RATE * 2 : # note: it's an accident of code that the 1st file will be 2 hours of samples and each subsequent file will be 1 hour samples.flush_file() samples.forget_old_samples(3600 * tz_cms50.SAMPLE_RATE) # keep some around so we can update clients with them (though we only need a few seconds worth if ofile_name : fn = tz_cms50.get_output_file_name(ofile_name, program_name = program_name) print "Outputting to: ", samples.open_write_file(fn, only_new_samples = True) pass rx_when = tzlib.elapsed_time() t = tzlib.elapsed_time() if t - ts > 59 : ts = t samples.flush_file() logger.flush() pass except tz_cms50.a_cms50_no_finger_exception : no_finger = True except tz_cms50.a_cms50_data_exception, msg : no_finger = True samples.flush_file() print msg # those bits are not, apparently, dupes of each other if not stopped : sys.exit(199) stopped -= 1 except ( tz_cms50.a_cms50_exception, tz_cms50.a_cms50_timeout_exception, ) : t = tzlib.elapsed_time() if t - rx_when > 0.5 : rx_when = t samples.flush_file() cm.start_usb() no_finger = True pass if no_finger : if finger : finger = False samples.append(tz_cms50.a_finger_sample(False)) spoofer.start_now() print "No finger" samples.flush_file() logger.flush() pass if (not s) and (not finger) : s = spoofer.get_sample() if s : st.append(s.y) pass if len(cm.data) : samples.flush_file() while len(cm.data) : dt = cm.data.pop(0) if len(dt.samples) : fn = tz_cms50.get_output_file_name(ofile_name, program_name = program_name, ext = ".dat") 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 stopped = 1000 k = TZKeyReady.key_ready() if k : print samples.flush_file() logger.flush() if k.lower() in [ 'q', '\033', ] : break if k.lower() in [ '?', ] : ca = list(me.clients.values()) ca.sort(lambda a, b : cmp(b.when, a.when)) lsa = [ "%s : %s:%s %s %s" % ( c.ip_adr, str(c.ci), c.name or "", time.asctime(time.localtime(time.time() - (tzlib.elapsed_time() - c.when))), c.host or '', ) for c in ca[-20:] ] for ls in lsa : logger.log(ls, flush = True) print "Show samples %s (key: v to toggle)" % show_samples if k.lower() in [ 'v', ] : show_samples = not show_samples print "Show samples %s (key: v to toggle)" % show_samples pass pass me.stop() samples.close_write_file() cm.blind_close() logger.close() # # # # eof