import sys # Test to use local customized version of pc-ble-driver-py... # Don't know how to do so for now, I just: # pip install pc-ble-driver-py # then directly motified the files in C:\Python38\Lib\site-packages\pc_ble_driver_py ... #sys.path.insert(0, "./pc-ble-driver-py") #import pc_ble_driver_py #print("TEST : {}".format(pc_ble_driver_py.__file__)) import time from datetime import datetime import logging import re import json import os import queue from traceback import print_tb from pc_ble_driver_py.observers import (BLEDriverObserver, BLEAdapterObserver, gen_conn_params_str) import threading import tkinter as tk from tkinter.scrolledtext import ScrolledText from tkinter.ttk import Treeview import wave, struct import winsound import numpy as np import scipy import scipy.io.wavfile import scipy.fftpack VERSION = "V0.0.7" logger = logging.getLogger(__name__) # https://docs.python.org/3/library/logging.html#logging-levels LOGGING_LEVEL_TEST_RESULT_SUCCESS = 25 LOGGING_LEVEL_TEST_RESULT_FAILURE = 26 logging.addLevelName(LOGGING_LEVEL_TEST_RESULT_SUCCESS, "TEST_RESULT_SUCCESS") logging.addLevelName(LOGGING_LEVEL_TEST_RESULT_FAILURE, "TEST_RESULT_FAILURE") # Load configuration CONFIG = {} try: with open(sys.argv[1] if len(sys.argv) == 2 else 'config.json') as config_file: CONFIG = json.load(config_file) except: pass # Preparing test results logging test_results_file = CONFIG.get("test_results_file", "") if test_results_file == "automatic": TEST_RESULTS_FOLDER = './test_results' if not os.path.isdir(TEST_RESULTS_FOLDER): os.makedirs(TEST_RESULTS_FOLDER) test_results_file = datetime.now().strftime(TEST_RESULTS_FOLDER + "/%Y%m%d_%H%M%S.csv") if len(test_results_file) > 0: logger.debug("Saving test results to {}".format(test_results_file)) test_results_file = open(test_results_file, "a+") else: test_results_file = None TARGET_DEV_NAME = "xxxxxxx" TARGET_FW_VERSION = "2.0" CFG_TAG = 1 TARGET_KEYS = { "GUIDE" : {"page":0x0C, "code": 0x8D}, "POWER" : {"page":0x0C, "code": 0x30}, "1" : {"page":0x07, "code": 0x1E}, "2" : {"page":0x07, "code": 0x1F}, "3" : {"page":0x07, "code": 0x20}, "4" : {"page":0x07, "code": 0x21}, "5" : {"page":0x07, "code": 0x22}, "6" : {"page":0x07, "code": 0x23}, "7" : {"page":0x07, "code": 0x24}, "8" : {"page":0x07, "code": 0x25}, "9" : {"page":0x07, "code": 0x26}, "MOVIE" : {"page":0x07, "code": 0x3F}, "0" : {"page":0x07, "code": 0x27}, "TV" : {"page":0x0C, "code": 0x89}, "UP" : {"page":0x0C, "code": 0x42}, "LEFT" : {"page":0x0C, "code": 0x44}, "OK" : {"page":0x0C, "code": 0x41}, "RIGHT" : {"page":0x0C, "code": 0x45}, "DOWN" : {"page":0x0C, "code": 0x43}, "BACK" : {"page":0x0C, "code":0x224}, "MIC" : {"page":0x0C, "code":0x221}, "HOME" : {"page":0x0C, "code":0x223}, "NETFLIX" : {"page":0x07, "code": 0x43}, "VOL_P" : {"page":0x0C, "code": 0xE9}, "PLAY_PAUSE" : {"page":0x0C, "code": 0xCD}, "CH_P" : {"page":0x0C, "code": 0x9C}, "VOL_M" : {"page":0x0C, "code": 0xEA}, "MUTE" : {"page":0x0C, "code": 0xE2}, "CH_M" : {"page":0x0C, "code": 0x9D}, } TARGET_KEYS_COORDINATES = { "GUIDE" : {"x": 22, "y": 72, "w":40, "h":40}, "POWER" : {"x":125, "y": 72, "w":40, "h":40}, "1" : {"x": 22, "y":116, "w":40, "h":40}, "2" : {"x": 72, "y":116, "w":40, "h":40}, "3" : {"x":125, "y":116, "w":40, "h":40}, "4" : {"x": 22, "y":161, "w":40, "h":40}, "5" : {"x": 72, "y":161, "w":40, "h":40}, "6" : {"x":125, "y":161, "w":40, "h":40}, "7" : {"x": 22, "y":207, "w":40, "h":40}, "8" : {"x": 72, "y":207, "w":40, "h":40}, "9" : {"x":125, "y":207, "w":40, "h":40}, "MOVIE" : {"x": 22, "y":252, "w":40, "h":40}, "0" : {"x": 72, "y":252, "w":40, "h":40}, "TV" : {"x":125, "y":252, "w":40, "h":40}, "UP" : {"x": 72, "y":295, "w":40, "h":40}, "LEFT" : {"x": 22, "y":347, "w":40, "h":40}, "OK" : {"x": 72, "y":347, "w":40, "h":40}, "RIGHT" : {"x":125, "y":347, "w":40, "h":40}, "DOWN" : {"x": 72, "y":390, "w":40, "h":40}, "BACK" : {"x": 22, "y":441, "w":40, "h":40}, "MIC" : {"x": 72, "y":441, "w":40, "h":40}, "HOME" : {"x":125, "y":441, "w":40, "h":40}, "NETFLIX" : {"x": 53, "y":492, "w":77, "h":24}, "VOL_P" : {"x": 22, "y":527, "w":40, "h":40}, "PLAY_PAUSE" : {"x": 72, "y":527, "w":40, "h":40}, "CH_P" : {"x":125, "y":527, "w":40, "h":40}, "VOL_M" : {"x": 22, "y":572, "w":40, "h":40}, "MUTE" : {"x": 72, "y":572, "w":40, "h":40}, "CH_M" : {"x":125, "y":572, "w":40, "h":40}, "AUDIO" : {"x": 72, "y": 72, "w":40, "h":40}, } TARGET_TEST_LIST = [*TARGET_KEYS.keys(), "AUDIO"] # noinspection PyGlobalUndefined global config, BLEDriver, BLEAdvData, BLEEvtID, BLEAdapter, BLEEnableParams, BLEGapTimeoutSrc, BLEUUID, BLEConfigCommon, BLEConfig, BLEConfigConnGatt, BLEGapScanParams from pc_ble_driver_py import (config, exceptions as BLEException) config.__conn_ic_id__ = "NRF52" # noinspection PyUnresolvedReferences from pc_ble_driver_py.ble_driver import ( BLEDriver, BLEAdvData, BLEEvtID, BLEEnableParams, BLEGapTimeoutSrc, BLEUUIDBase, BLEUUID, BLEGapScanParams, BLEConfigCommon, BLEConfig, BLEConfigConnGatt, BLEGattStatusCode, BLEGapRoles, BLEGapConnParams, BLEConfigConnGap, ) # noinspection PyUnresolvedReferences from pc_ble_driver_py.ble_adapter import BLEAdapter global nrf_sd_ble_api_ver nrf_sd_ble_api_ver = config.sd_api_ver_get() HID_BASE_UUID = BLEUUID(0x2A4D) BATTERY_UUID = BLEUUID(0x2A19) ATVV_BASE_UUID = BLEUUIDBase([0xAB, 0x5E, 0x00, 0x01, 0x5A, 0x21, 0x4F, 0x05, 0xBC, 0x7D, 0xAF, 0x01, 0xF6, 0x17, 0xB6, 0x64]) ATVV_TX_CHAR = BLEUUID(0x0002, ATVV_BASE_UUID) ATVV_RX_CHAR = BLEUUID(0x0003, ATVV_BASE_UUID) ATVV_CTL_CHAR = BLEUUID(0x0004, ATVV_BASE_UUID) ATVV_FREQUENCY = 8000 def wav_frequency_peak(filename): # Get signal data Fs, signal = scipy.io.wavfile.read(filename) N = signal.shape[0] Ts = 1.0 / Fs # Compute signal FFT FFT = abs(scipy.fft.fft(signal)) FFT_side = FFT[range(N // 2)] # Compute corresponding FFT frequency scale freqs = scipy.fftpack.fftfreq(signal.size, Ts) freqs_side = freqs[range(N // 2)] # Keep only frequency > 50Hz freqs_side = freqs_side[freqs_side > 50] FFT_side = FFT_side[-len(freqs_side):] # Find peak peak_idx = np.where(FFT_side == max(FFT_side)) peak_frequency = max(freqs_side[peak_idx]) return peak_frequency from serial.tools.list_ports import comports def serial_ports(vendor_id: int = None, product_id: int = None): return [port.device for port in comports() if ((vendor_id is None or vendor_id == port.vid) and (product_id is None or product_id == port.pid))] class ADPCMDecoder: STEPSIZE_LUT = [ 7, 8, 9, 10, 11, 12, 13, 14, 16, 17, 19, 21, 23, 25, 28, 31, 34, 37, 41, 45, 50, 55, 60, 66, 73, 80, 88, 97, 107, 118, 130, 143, 157, 173, 190, 209, 230, 253, 279, 307, 337, 371, 408, 449, 494, 544, 598, 658, 724, 796, 876, 963, 1060, 1166, 1282, 1411, 1552, 1707, 1878, 2066, 2272, 2499, 2749, 3024, 3327, 3660, 4026, 4428, 4871, 5358, 5894, 6484, 7132, 7845, 8630, 9493, 10442, 11487, 12635, 13899, 15289, 16818, 18500, 20350, 22385, 24623, 27086, 29794, 32767, ] INDEX_LUT = [ -1, -1, -1, -1, 2, 4, 6, 8, -1, -1, -1, -1, 2, 4, 6, 8, ] @staticmethod def clamp(sample): sample = sample if sample < 32767 else 32767 sample = sample if sample > -32767 else -32767 return sample def __init__(self): self.audio_frame = [] self.audio_pcm_frame_num = 0 @staticmethod def decode_frame(adpcm): siDec = adpcm[2] if siDec & 0x80: siDec = (siDec - 256) pvDec = ((adpcm[0] << 8) | adpcm[1]) if pvDec & 0x8000: pvDec = (pvDec - 65536) def decode_nibble(nibble): nonlocal siDec nonlocal pvDec step = ADPCMDecoder.STEPSIZE_LUT[siDec] cumDiff = (step >> 3) siDec += ADPCMDecoder.INDEX_LUT[nibble] if siDec < 0: siDec = 0 elif siDec >= len(ADPCMDecoder.STEPSIZE_LUT): siDec = (len(ADPCMDecoder.STEPSIZE_LUT) - 1) if (nibble & 4) != 0: cumDiff += step if (nibble & 2) != 0: cumDiff += (step >> 1) if (nibble & 1) != 0: cumDiff += (step >> 2) if (nibble & 8) != 0: if pvDec < (cumDiff - 32767): pvDec = -32767 else: pvDec -= cumDiff else: if pvDec > (32767 - cumDiff): pvDec = 32767 else: pvDec += cumDiff return ADPCMDecoder.clamp(pvDec) pcm = [] i = 3 while i < len(adpcm): pcm.append(decode_nibble(adpcm[i] >> 4)) pcm.append(decode_nibble(adpcm[i] & 0x0F)) i = i + 1 return pcm def feed(self, data): pcm = None ''' if len(self.audio_frame) >= 128: self.audio_frame = [] self.audio_frame.extend(data) print("len data : {}, len audio_frame : {}".format(len(data), len(self.audio_frame))) if len(data) <= 20: if len(self.audio_frame) == 128: pcm = ADPCMDecoder.decode_frame(self.audio_frame[3:]) self.audio_pcm_frame_num += 1 print(self.audio_pcm_frame_num) ''' self.audio_frame.extend(data) if len(data) < 20: if len(self.audio_frame) == 134: pcm = ADPCMDecoder.decode_frame(self.audio_frame) if len(self.audio_frame) >= 134: self.audio_frame = [] return pcm class BLE(BLEDriverObserver, BLEAdapterObserver): def __init__(self, gui, port = None): super(BLE, self).__init__() self.gui = gui self.gui.set_ble(self) if port == None: ports = BLEDriver.enum_serial_ports() # This is bugged if len(ports) <= 0: ports = serial_ports(0x1915, 0xC00A) if len(ports) > 0: port = ports[0] else: port = ports[0].port if port == None: raise IOError("BLE dongle serial port not found !") self.gui.set_port(port) self.conn_q = queue.Queue(maxsize=1) # Used to sync connection event self.adpcm_decoder = None self.scanned_devices = {} self.pressed_keys = { "KBRD" : [], "CSMR" : [], } self.adapter = BLEAdapter(BLEDriver(serial_port=port, auto_flash=False, baud_rate=1000000, log_severity_level="info")) self.adapter.observer_register(self) self.adapter.driver.observer_register(self) self.adapter.driver.open() gatt_cfg = BLEConfigConnGatt() gatt_cfg.att_mtu = 250 gatt_cfg.tag = CFG_TAG self.adapter.driver.ble_cfg_set(BLEConfig.conn_gatt, gatt_cfg) conn_cfg = BLEConfigConnGap() conn_cfg.conn_count = 1 conn_cfg.event_length = 320 self.adapter.driver.ble_cfg_set(BLEConfig.conn_gap, conn_cfg) self.adapter.driver.ble_enable() self.adapter.driver.ble_vs_uuid_add(ATVV_BASE_UUID) def start_scan(self): self.scanned_devices = {} self.adapter.driver.ble_gap_scan_start(scan_params=BLEGapScanParams(interval_ms=200, window_ms=150, timeout_s=0)) def stop_scan(self): self.adapter.driver.ble_gap_scan_stop() def disconnect(self, conn): self.adapter.disconnect(conn) def connect(self, mac, on_connected = None): def connection_thread(dev, on_connected): conn = None self.adapter.connect(dev["addr"], scan_params=BLEGapScanParams(interval_ms=200, window_ms=150, timeout_s=5), conn_params=BLEGapConnParams(7.5, 7.5, 1500, 0), tag=CFG_TAG) try: conn = self.conn_q.get(timeout=5) except queue.Empty: pass if on_connected is not None: on_connected(dev, conn) thread = threading.Thread(target=connection_thread, args=(self.scanned_devices[mac], on_connected)) thread.isDaemon = True thread.start() def configure_device(self, conn, on_configured): self.pressed_keys["KBRD"] = [] self.pressed_keys["CSMR"] = [] def configuration_thread(conn, on_configured): try: self.adapter.service_discovery(conn) for s in self.adapter.db_conns[conn].services: logger.debug(str(s)) for c in s.chars : logger.debug(str(c)) except BLEException.NordicSemiException as err: if on_configured is not None: on_configured(conn, "Couldn't discover services : {}".format(err)) try: self.adapter.authenticate(conn, _role=BLEGapRoles.central) except BLEException.NordicSemiException as err: if on_configured is not None: on_configured(conn, "Couldn't pair : {}".format(err)) try: self.adapter.enable_notification(conn, HID_BASE_UUID, attr_handle=0x2C) self.adapter.enable_notification(conn, HID_BASE_UUID, attr_handle=0x30) self.adapter.enable_notification(conn, ATVV_RX_CHAR) self.adapter.enable_notification(conn, ATVV_CTL_CHAR) self.adapter.write_cmd(conn, ATVV_TX_CHAR, [0x0A, 0x01, 0x00, 0x00, 0x03, 0x01]) # Request capacity except BLEException.NordicSemiException as err: if on_configured is not None: on_configured(conn, "Couldn't enable notifications : {}".format(err)) if on_configured is not None: on_configured(conn) thread = threading.Thread(target=configuration_thread, args=(conn, on_configured)) thread.isDaemon = True thread.start() def get_device_fw_version(self, conn): status_code, value = self.adapter.read_req(conn, BLEUUID(0x2A26)) if value is None: logger.error("Couldn't read device firmware version : {}".format(status_code)) else: value = "".join(chr(e) for e in value) return value def on_gap_evt_connected(self, ble_driver, conn_handle, peer_addr, role, conn_params): logger.debug("New connection: {}".format(conn_handle)) self.conn_q.put(conn_handle) def on_gap_evt_disconnected(self, ble_driver, conn_handle, reason): try: self.conn_q.get_nowait() except: pass self.gui.on_disconnected(reason) def on_gap_evt_adv_report(self, ble_driver, conn_handle, peer_addr, rssi, adv_type, adv_data): if BLEAdvData.Types.complete_local_name in adv_data.records: dev_name_list = adv_data.records[BLEAdvData.Types.complete_local_name] elif BLEAdvData.Types.short_local_name in adv_data.records: dev_name_list = adv_data.records[BLEAdvData.Types.short_local_name] else: dev_name_list = [] name = "".join(chr(e) for e in dev_name_list) mac = "".join("{0:02X}:".format(b) for b in peer_addr.addr)[:-1] self.scanned_devices[mac] = {"name": name, "mac" : mac, "addr": peer_addr, "rssi": rssi} self.gui.on_scanned(self.scanned_devices) def start_voice(self, conn, from_gui=False): try: logger.info("Starting voice") self.adpcm_decoder = ADPCMDecoder() self.adapter.write_cmd(conn, ATVV_TX_CHAR, [0x0C, 0x01], attr_handle=0x1B) self.cnt = 0 except: return False return True def stop_voice(self, conn): try: logger.info("Stopping voice") self.adapter.write_cmd(conn, ATVV_TX_CHAR, [0x0D], attr_handle=0x1B) except: return False return True def on_notification_handle(self, ble_adapter, conn_handle, uuid, attr_handle, data): if uuid == HID_BASE_UUID: keys = [] page_name = "KBRD" page_code = 0x07 if attr_handle == 0x30: data = [(data[0] | ((data[1] & 0xF) << 8)), (((data[1] & 0xF0) >> 4) | (data[2] << 4))] page_name = "CSMR" page_code = 0x0C for k in data: if k: for label, value in TARGET_KEYS.items(): if value["page"] == page_code and value["code"] == k: keys.append(label) for k in self.pressed_keys[page_name]: if not k in keys: self.gui.on_key_released(conn_handle, k) self.pressed_keys[page_name].remove(k) for k in keys: if not k in self.pressed_keys[page_name]: self.gui.on_key_pressed(conn_handle, k) self.pressed_keys[page_name].append(k) elif uuid == ATVV_RX_CHAR: self.cnt += 1 logger.debug(str(self.cnt)) if data is not []: self.voice_received(conn_handle, data) elif uuid == ATVV_CTL_CHAR: logger.debug("CTL : {}".format(' '.join('{:02X}'.format(b) for b in data))) if data == [0]: self.gui.on_key_released(conn_handle, "MIC", False) elif uuid == BATTERY_UUID: logger.debug("Battery level : {}".format(data[0])) else: logger.warning("Unknow notification : {} {}".format(uuid, data)) def voice_received(self, conn, data): if self.adpcm_decoder: pcm = self.adpcm_decoder.feed(data) if pcm is not None: self.gui.on_audio_received(conn, pcm) def on_indication(self, ble_adapter, conn_handle, uuid, char_handle, data): logger.debug("Indication: {}, {} ({}) = {}".format(conn_handle, uuid, char_handle, data)) def on_conn_param_update_request(self, ble_adapter, conn_handle, conn_params): logger.debug("Conn param update request: {}\n Params({})".format(conn_handle, gen_conn_params_str(conn_params))) def stop(self): self.adapter.driver.close() class GUI(): class LoggingQueueHandler(logging.Handler): def __init__(self, st): super().__init__() self.log_queue = queue.Queue() self.st = st self.window = st while self.window.master is not None: self.window = self.window.master self.window.after(50, self.poll_log_queue) def emit(self, record): self.log_queue.put(record) def poll_log_queue(self): while True: try: record = self.log_queue.get(block=False) except queue.Empty: break else: self.display_log(record) self.window.after(50, self.poll_log_queue) def display_log(self, record): msg = self.format(record) scroll = self.st.bbox("end-1c") is not None self.st.configure(state='normal') self.st.insert(tk.END, msg + '\n', record.levelname) self.st.update_idletasks() self.st.configure(state='disabled') if scroll or CONFIG.get("force_logs_autoscroll", False): self.st.yview(tk.END) self.st.update_idletasks() def __init__(self): self.ble = None self.tested_devices = [] self.current_tested_device = None self.audio_data = None self.test_list = None self.clicked_keys = {} self.window = tk.Tk() self.window.title("BLE Test {} ({})".format(TARGET_DEV_NAME, VERSION)) self.window.grid_rowconfigure(0, weight=1) self.window.grid_columnconfigure(0, weight=1) self.state_panel = tk.Frame(self.window) self.state_panel.grid(row=0, column=0, sticky=(tk.N + tk.E + tk.W + tk.S)) self.state_panel.grid_rowconfigure(0, weight=1) self.state_panel.grid_columnconfigure(0, weight=1) self.state_panel.rowconfigure(0, weight=1) self.state_panel.columnconfigure(0, weight=1) # Setup GUI logging st = ScrolledText(self.state_panel, state='disabled') st.configure(font='TkFixedFont') st.tag_config('INFO', foreground='black') st.tag_config('DEBUG', foreground='gray') st.tag_config('WARNING', foreground='orange') st.tag_config('ERROR', foreground='red') st.tag_config('CRITICAL', foreground='red', underline=1) st.tag_config('TEST_RESULT_SUCCESS', background='green', font=("TkFixedFont", 16)) st.tag_config('TEST_RESULT_FAILURE', background='red', font=("TkFixedFont", 16)) st.grid(row=0, column=0, sticky=(tk.N + tk.E + tk.W + tk.S)) self.logging_handler = GUI.LoggingQueueHandler(st) self.logging_handler.setFormatter(logging.Formatter('%(asctime)s: %(message)s')) logger.addHandler(self.logging_handler) logging.getLogger("pc_ble_driver_py.observers").addHandler(self.logging_handler) logging.getLogger("pc_ble_driver_py.ble_adapter").addHandler(self.logging_handler) logger.debug(CONFIG) self.status_label = tk.Label(self.state_panel, text="") self.set_status("Initializing...") # DUT panel self.dut_panel = tk.Frame(self.window) self.devices_treeview = Treeview(self.window) self.devices_treeview["columns"] = ("name", "rssi") self.devices_treeview.column("#0", width=120, minwidth=120, stretch=tk.NO) self.devices_treeview.column("name", width=150, minwidth=150) self.devices_treeview.column("rssi", width=45, minwidth=45, stretch=tk.NO) self.devices_treeview.heading("#0", text="MAC") self.devices_treeview.heading("name", text="Name") self.devices_treeview.heading("rssi", text="RSSI") self.devices_treeview.bind("<>", self.on_device_selected) self.devices_treeview.grid(row=0, column=1, sticky=(tk.N + tk.S)) self.fw_version_label = tk.Label(self.dut_panel, text="FW Version : {}".format("UNKNOW")) self.fw_version_label.grid(row=0, column=0) self.remote_img = tk.PhotoImage(file="xxxxx.png") # We must keep a reference for some reason self.remote_canvas = tk.Canvas(self.dut_panel, width=self.remote_img.width(), height=self.remote_img.height()) self.remote_canvas.create_image((0, 0), image=self.remote_img, anchor=tk.NW) self.remote_canvas.grid(row=1, column=0) #self.dut_panel.grid(row=0, column=1) def set_status(self, status): logger.info(status) self.status_label["text"] = status self.status_label.grid(row=1, column=0) def on_device_selected(self, event): try: item = self.devices_treeview.selection() except IndexError as err: logger.error("{}".format(err)) return mac = self.devices_treeview.item(item, "text") self.devices_treeview.delete(*self.devices_treeview.get_children()) self.devices_treeview.grid_forget() self.set_status("Stopping scan...") self.ble.stop_scan() self.set_status("Connecting to {}...".format(mac)) self.ble.connect(mac, self.on_connected) def mainloop(self): return self.window.mainloop() def set_ble(self, ble): self.ble = ble def set_port(self, port): logger.debug("Using port {}".format(port)) self.window.title("{} ({})".format(self.window.title(), port)) def on_configured(self, conn, error = None): if error is None: self.set_status("Device configured") version = self.ble.get_device_fw_version(conn) if version is None: logger.error("Couldn't read device firmware version") self.ble.disconnect(conn) else: logger.info("FW Version : {}".format(version)) self.fw_version_label["text"] = "FW Version : {}".format(version) if version != TARGET_FW_VERSION: logger.error("Invalid firmware version : {}".format(version)) self.ble.disconnect(conn) self.test_list = TARGET_TEST_LIST.copy() self.clicked_keys = {} for k in TARGET_KEYS_COORDINATES.keys(): self.clicked_keys[k] = 0 self.set_status("Testing...") self.dut_panel.grid(row=0, column=1) else: logger.error("Configuration failed : {}".format(error)) self.ble.disconnect(conn) def on_connected(self, dev, conn): if conn is not None: self.set_status("Connected to {}".format(dev["mac"])) self.current_tested_device = dev self.set_status("Configuring device...") self.ble.configure_device(conn, self.on_configured) else: logger.error("Couldn't connect to {}".format(dev["mac"])) self.set_status("Scanning...") self.ble.start_scan() def on_disconnected(self, reason): self.set_status("Disconnected ({})".format(reason)) if len(self.test_list) > 0: self.on_test_end(False, self.current_tested_device, None) self.current_tested_device = None #winsound.PlaySound("1KHz.wav", winsound.SND_FILENAME | winsound.SND_PURGE) winsound.PlaySound(None, winsound.SND_PURGE) self.dut_panel.grid_forget() self.remote_canvas.delete(tk.ALL) self.remote_canvas.create_image((0, 0), image=self.remote_img, anchor=tk.NW) self.devices_treeview.grid(row=0, column=1, sticky=(tk.N + tk.S)) self.set_status("Scanning...") self.ble.start_scan() SCAN_REFRESH_PERIOD_SEC = 0.5 next_scan_refresh = 0 def on_scanned(self, devices): if GUI.next_scan_refresh < time.time(): GUI.next_scan_refresh = (time.time() + GUI.SCAN_REFRESH_PERIOD_SEC) self.devices_treeview.delete(*self.devices_treeview.get_children()) for _, dev in sorted(devices.items(), key=lambda item: -item[1]["rssi"]): if dev["name"] == TARGET_DEV_NAME: if dev["rssi"] >= CONFIG.get("scan_rssi_threshold", float("-inf")): if not (CONFIG.get("hide_successfully_tested", False) and (dev["mac"] in self.tested_devices)): self.devices_treeview.insert("", tk.END, text=dev["mac"], values=(dev["name"], dev["rssi"])) self.devices_treeview.grid(row=0, column=1, sticky=(tk.N + tk.S)) if CONFIG.get("autoconnect_nearest", False): children = self.devices_treeview.get_children() if len(children) > 0: self.devices_treeview.selection_set(children[0]) def voice_end(self, conn): self.ble.stop_voice(conn) def on_key_pressed(self, conn, key): logger.debug("Key {} pressed".format(key)) coord = TARGET_KEYS_COORDINATES[key] self.remote_canvas.create_rectangle(coord["x"], coord["y"], coord["x"] + coord["w"], coord["y"] + coord["h"], fill="yellow") if key == "MIC": #threading.Timer(7, self.voice_end, args=[conn]).start() self.audio_data = [] #winsound.PlaySound("1KHz.wav", winsound.SND_FILENAME | winsound.SND_ASYNC | winsound.SND_NODEFAULT | winsound.SND_LOOP) self.ble.start_voice(conn) def on_key_released(self, conn, key, mic_state = True): if key == "MIC" and not mic_state: #self.ble.stop_voice(conn) #winsound.PlaySound("1KHz.wav", winsound.SND_FILENAME | winsound.SND_PURGE) #winsound.PlaySound(None, winsound.SND_PURGE) def save_voice_record(self, conn): if (self.audio_data is None) or (len(self.audio_data) <= 0): logger.warning("No audio data to save") return audio_file = wave.open("test.wav", "w") audio_file.setnchannels(1) audio_file.setsampwidth(2) audio_file.setframerate(ATVV_FREQUENCY) for s in self.audio_data: audio_file.writeframes(struct.pack('h', ADPCMDecoder.clamp(s))) audio_file.close() self.audio_data = None #winsound.PlaySound("test.wav", winsound.SND_FILENAME | winsound.SND_ASYNC | winsound.SND_NODEFAULT) freq_peak = wav_frequency_peak("test.wav") color = "red" if (freq_peak < 950) or (freq_peak > 1050): logger.warning("Wrong frequency peak : {}".format(freq_peak)) else: logger.debug("Frequency peak : {}".format(freq_peak)) color = "green" if "AUDIO" in self.test_list: self.test_list.remove("AUDIO") if len(self.test_list) <= 0: self.on_test_end(True, self.current_tested_device, conn) coord = TARGET_KEYS_COORDINATES["AUDIO"] self.remote_canvas.create_rectangle(coord["x"], coord["y"], coord["x"] + coord["w"], coord["y"] + coord["h"], fill=color) if self.audio_data is not None: threading.Timer(0.1, save_voice_record, [self, conn]).start() else: logger.debug("Key {} released".format(key)) COLORS = ["green", "blue", "grey", "cyan", "brown", "purple", "#404040"] color = COLORS[self.clicked_keys[key] % len(COLORS)] self.clicked_keys[key] += 1 coord = TARGET_KEYS_COORDINATES[key] self.remote_canvas.create_rectangle(coord["x"], coord["y"], coord["x"] + coord["w"], coord["y"] + coord["h"], fill=color) if key in self.test_list: self.test_list.remove(key) if len(self.test_list) <= 0: self.on_test_end(True, self.current_tested_device, conn) def on_audio_received(self, conn, pcm): if self.audio_data is not None: self.audio_data.extend(pcm) def on_test_end(self, success, dev=None, conn=None): if success: self.set_status("TEST SUCCESS") logger.log(LOGGING_LEVEL_TEST_RESULT_SUCCESS, "TEST SUCCESS") if dev is not None: self.tested_devices.append(dev["mac"]) else: self.set_status("TEST FAILURE") logger.log(LOGGING_LEVEL_TEST_RESULT_FAILURE, "TEST FAILURE") if dev is not None: if test_results_file is not None: test_results_file.write("{};{};{}\n".format(dev["mac"], success, datetime.now().strftime("%Y/%m/%d %H:%M:%S"))) test_results_file.flush() if conn is not None: self.ble.disconnect(conn) if __name__ == "__main__": logging.basicConfig( level="DEBUG", format="%(asctime)s [%(thread)d/%(threadName)s] %(message)s", ) logging.getLogger("pc_ble_driver_py.observers").setLevel("INFO") logging.getLogger("pc_ble_driver_py.ble_adapter").setLevel("INFO") log_file = CONFIG.get("log_file", "") if log_file == "automatic": LOGS_FOLDER = './logs' if not os.path.isdir(LOGS_FOLDER): os.makedirs(LOGS_FOLDER) log_file = datetime.now().strftime(LOGS_FOLDER + "/%Y%m%d_%H%M%S.log") if len(log_file) > 0: logger.debug("Saving logs to {}".format(log_file)) fh = logging.FileHandler(log_file) fh.setFormatter(logging.Formatter('%(asctime)s: %(message)s')) fh.setLevel(logging.DEBUG) logger.addHandler(fh) logging.getLogger("pc_ble_driver_py.observers").addHandler(fh) logging.getLogger("pc_ble_driver_py.ble_adapter").addHandler(fh) gui = GUI() ble = BLE(gui, CONFIG.get("com_port", None)) gui.set_status("Scanning...") ble.start_scan() gui.mainloop() ble.stop()