# -*- coding: utf-8 -*- # APNEA_M USB Serial I/F # by ik-gadgets import sys import re import os import serial import serial.tools.list_ports import codecs import time import msvcrt from datetime import datetime command_read_mic_level = 0xA3 command_stop = 0xA5 command_read_mV = 0xAC command_run = 0x5A command_read_measure = 0x53 command_sensitivity = 0x5C def auto_detect_serial_port(): ports = list(serial.tools.list_ports.comports()) print("### Search results ###") for p in ports: print(" ",p) if 'USB Serial Port' in p.description: return p.device return None def Tx_byte(ser,d): ser.write(d.to_bytes(1,'big')) # time.sleep(0.01) def Rx_byte(ser): while True: bytesWaiting = ser.inWaiting() if bytesWaiting != 0: onebyte = ser.read(1) oneint = int.from_bytes(onebyte,'big') # time.sleep(0.01) return oneint def Rx_2byte(ser): D = Rx_byte(ser) D <<= 8 D += Rx_byte(ser) return D def print_usage(): print("Command input") print(" s : stop measurement & display result") print(" w : write result to file") print(" r : run (itialization and start measurement)") print(" k : read and set sensitivity") print(" p : run & continuous print of mic_level readings") print(" m : continuous print of breath-holding period") print(" v : read battery voltage") print(" [Enter]: end mesurement or end program") print("") def main(): sensitivity = 0 rx_list = [] print() serial_port_name = None file_name_list = [] wfile_name = "" if len(sys.argv) >= 2: for file_name in sys.argv[1:]: if re.search(r'\.txt$',file_name): wfile_name = file_name sep = " " elif re.search(r'\.csv$',file_name): wfile_name = file_name sep = "," elif re.search(r'^(com)|(COM)[0-9]+$',file_name): serial_port_name = file_name print() if wfile_name: wfile = codecs.open(wfile_name, 'w', 'utf-8') if not serial_port_name: serial_port_name = auto_detect_serial_port() if not serial_port_name: print() print("### ERROR ### USB-Serial port not recognized") print(" Please check the connection or specify the USB-Serial port name") print(" ex. python APNEA_M.py com3") print() sys.exit() print() print("### Try to communicate",serial_port_name,"###") print() ser = serial.Serial(serial_port_name,9600) print_usage() while True: key_data = input("Command(s,w,r,k,p,m,v,e,[Enter]) = ") if key_data == "s": ##### stop measurement and get result ##### Tx_byte(ser,command_stop) stop_cnt = 0 max_stop_period = 0 print("") stop_sum = 0 sensitivity = Rx_byte(ser) while True: sec = Rx_2byte(ser) stop_sec = Rx_byte(ser) stop_sum += stop_sec if stop_sec == 0: break stop_cnt += 1 if stop_sec > max_stop_period: max_stop_period = stop_sec H = sec//3600 M = (sec%3600)//60 S = sec%60 print(f"{H:02d}:{M:02d}:{S:02d} {stop_sec:3d}sec") H = sec//3600 M = (sec%3600)//60 S = sec%60 print(f"MEASURE PERIOD {H:02d}:{M:02d}:{S:02d}") print("STOP COUNT",stop_cnt) print("STOP MAX",max_stop_period) if stop_cnt == 0: ave = 0 else: ave = stop_sum/stop_cnt print("STOP AVERAGE %.2f" % (ave)) if sec ==0: sec = 1.0 print("STOP COUNT/Hour %.2f" % (stop_cnt/(sec/3600))) print("SENSITIVITY %3d" % (sensitivity)) print("") elif key_data == "w": ##### stop measurement and write result to file ##### Tx_byte(ser,command_stop) wfile_name = input(" Output File Name(*.txt or *.csv or [Enter]) = ") if wfile_name == "": now = datetime.now() wfile_name = now.strftime("%Y%m%d") + ".csv" if re.search(r'\.csv$',wfile_name): sep = "," else: sep = " " print("") print(" Output File Name = ",wfile_name) print("") wfile = codecs.open(wfile_name, 'w', 'utf-8') stop_cnt = 0 max_stop_period = 0 stop_sum = 0 sensitivity = Rx_byte(ser) while True: sec = Rx_2byte(ser) stop_sec = Rx_byte(ser) stop_sum += stop_sec if stop_sec == 0: break stop_cnt += 1 if stop_sec > max_stop_period: max_stop_period = stop_sec H = sec//3600 M = (sec%3600)//60 S = sec%60 wfile.write(f"{H:02d}:{M:02d}:{S:02d}{sep}{stop_sec:3d}sec\n") H = sec//3600 M = (sec%3600)//60 S = sec%60 wfile.write(f"MEASURE PERIOD{sep}{H:02d}:{M:02d}:{S:02d}\n") wfile.write("STOP COUNT"+sep+str(stop_cnt)+"\n") wfile.write("STOP MAX"+sep+str(max_stop_period)+"\n") if stop_cnt == 0: wfile.write(f"STOP AVERAGE{sep}0\n") else: wfile.write(f"STOP AVERAGE{sep}{stop_sum/stop_cnt:.2f}\n") wfile.write(f"STOP COUNT/Hour{sep}{stop_cnt/(sec/3600):.2f}\n") wfile.write(f"SENSITIVITY {sensitivity:3d}") wfile.write("") wfile.close() elif key_data == "p": ##### read mic level ##### M_MAX = 0 M_MIN = 0xFFFFFF while True: Tx_byte(ser,command_read_mic_level) #start = time.time() MH = Rx_byte(ser) #end = time.time() #print(f"time={(end-start)*1000:.3f}ms") ML = Rx_byte(ser) M = (MH<<8) + ML if M_MAX < M : M_MAX = M if M_MIN > M : M_MIN = M print(M) time.sleep(1) if msvcrt.kbhit(): key_data = input("") print("Pressure Max = ",M_MAX," Pressure Min = ",M_MIN," Delta = ",M_MAX-M_MIN) print("") break elif key_data == "m": ##### read measure status ##### while True: Tx_byte(ser,command_read_measure) #start = time.time() stop_period_sec = Rx_byte(ser) print("Stop Period = ",stop_period_sec,"sec") time.sleep(1) if msvcrt.kbhit(): key_data = input("") break elif key_data == "v": Tx_byte(ser,command_read_mV) mV = Rx_byte(ser) mV <<= 8 mV += Rx_byte(ser) print(f"{mV/1000:.2f}V") print("") elif key_data == "r": Tx_byte(ser,command_sensitivity) sensitivity = Rx_byte(ser) Tx_byte(ser,0) # 0ならチップ側で無視しsensitivity 書き換え無し Tx_byte(ser,command_run) print("APNEA CHECK START") print(" sensitivity =",sensitivity) print("") elif key_data == "k": s = input(" Sensitivity(>0,<=255,[Enter] for read only) = ") Tx_byte(ser,command_sensitivity) old_sensitivity = Rx_byte(ser) if s == "": Tx_byte(ser,0) print(" sensitivity = ",old_sensitivity) else: sensitivity = int(s) if (0 < sensitivity) and (sensitivity < 255): Tx_byte(ser,sensitivity) print(f" sensitivity changed {old_sensitivity:3d} to {sensitivity:3d}") else: Tx_byte(ser,0) # 0ならチップ側で無視しsensitivity 書き換え無し print(" value error (0